flow-matrix.yaml

---
UUID: "bcc2e200-7a57-11e8-913c-28d2442e1325"
Name: "FlowMatrix"
Title: "Create a Flow Matrix of the running processes"
Abstract: "This workflow aims to create a Flow Matrix of the running processes. Flow Matrix will report all opened Sockets between client and server processes across hosts."
Description: >
    # How to use:
      1. If you also want to use the flows to build matrix, Select 1st Checkbox
      2. Select the Flow matrix output format -> Graph (Add nodes and edges to the current topology), Dot (Gives 'Graphviz output')
      3. Select the last Checkbox, 'To see debug logs in the JavaScript console'
      4. Hit the 'Execute' button to run the workflow
    # How It Works:
      1. Please follow the link - 'skydive.network/blog/flow-matrix.html'
      2. For more information about Skydive workflows please follow - 'skydive.network/blog/introduction-to-workflows.html'
Parameters:
  - Name: useFlows
    Description: Also use flows to build matrix
    Type: boolean
  - Name: format
    Description: Flow matrix output format
    Type: choice
    Default: graph
    Values:
      - Description: "Add nodes and edges to the current topology"
        Value: graph
      - Description: "Graphviz output"
        Value: dot
  - Name: debug
    Description: Show debug logs in the JavaScript console
    Type: boolean
Source: |
    function FlowMatrix(useFlows, format, debug) {
        format = format || "graph"

        var G = client.gremlin.G()
        var at;
        var listenPorts = []
        var peerSockets = {}
        var matrix = {}
        var endpoints = {}
        var hosts = {}
        var hostAddresses = {}

        var isLoopback = function(addr) {
            return addr.indexOf("127.") == 0 ||
                   addr.indexOf("::1/") == 0
        }

        var isListenSocket = function(socket) {
            return (socket.State == "LISTEN") || (
                    ((socket.Protocol == "UDP") &&
                    (socket.State == "CLOSE") &&
                    ((socket.RemoteAddress == "0.0.0.0") ||
                    (socket.RemoteAddress == "::"))))
        }

        var buildCaches = function(nodes, interfaces) {
            // Build a list of the port for which we have a socket listening
            // and a dict that maps outgoing connection local port to the remote socket
            for (var i in nodes) {
                var host = nodes[i]
                hosts[host.Host] = host

                var sockets = host.Metadata.Sockets
                for (var j in sockets) {
                    var socket = host.Metadata.Sockets[j]
                    var localPort = socket.LocalPort
                    if (isListenSocket(socket)) {
                        listenPorts.push(localPort)
                    } else {
                        if (!(localPort in peerSockets)) {
                            peerSockets[localPort] = []
                        }
                        peerSockets[localPort].push({
                            "host": host.Host,
                            "socket": socket,
                        })
                    }
                }
            }

            for (var intf in interfaces) {
                var intf = interfaces[intf]
                for (var i in intf.Metadata.IPV4) {
                    var ip = intf.Metadata.IPV4[i].split('/')[0]
                    hostAddresses[ip] = hosts[intf.Host]
                }
            }
        }

        var getSocket = function(node, protocol, localAddress, localPort) {
            if (!node.Metadata.Sockets) return

            for (var i in node.Metadata.Sockets) {
                var socket = node.Metadata.Sockets[i]
                if (socket.Protocol == protocol && socket.LocalAddress == localAddress && socket.LocalPort == localPort) {
                    return socket
                }
            }
        }

        var getSocketPeer = function(socket, fromHost) {
            var localPort = socket.RemotePort
            var localAddress = socket.RemoteAddress
            var protocol = socket.Protocol

            if (!(localPort in peerSockets)) {
                return
            }

            for (var i in peerSockets) {
                var sockets = peerSockets[i]
                for (var j in sockets) {
                    var s = sockets[j].socket
                    if ((s.Protocol == protocol) && (s.LocalAddress == localAddress) && (s.LocalPort == localPort)) {
                        if (isLoopback(localAddress)) {
                            if (sockets[j].host == fromHost) {
                                return sockets[j]
                            }
                        } else {
                            return sockets[j]
                        }
                    }
                }
            }

            if (protocol == "UDP") {
                var node = hostAddresses[localAddress]
                if (node) {
                    for (var i in node.Metadata.Sockets) {
                        var s = node.Metadata.Sockets[i]
                        if (s.Protocol == "UDP" && isListenSocket(s) && s.LocalPort == localPort) {
                            return {
                                "host": node.Host,
                                "socket": socket
                            }
                        }
                    }
                }
            }
        }

        var log = function() {
            if (debug) {
                console.log.apply(console, arguments)
            }
        }

        var addEntry = function(protocol, serviceHost, serviceAddress, servicePort, serviceProcess, serviceName,
                                clientHost, clientAddress, clientProcess, clientName) {
            log("Adding entry", protocol, serviceHost, serviceAddress, servicePort, serviceProcess, serviceName,
                clientHost, clientAddress, clientProcess, clientName, Math.random())

            var key = [ protocol, serviceHost, serviceAddress, servicePort, serviceProcess, serviceName,
                        clientHost, clientAddress, clientProcess, clientName ].join(',')

            var entry = { "protocol": protocol,
                          "serviceHost": serviceHost,
                          "serviceAddress": serviceAddress,
                          "servicePort": servicePort,
                          "serviceProcess": serviceProcess,
                          "serviceName": serviceName,
                          "clientHost": clientHost,
                          "clientAddress": clientAddress,
                          "clientProcess": clientProcess,
                          "clientName": clientName }

            matrix[key] = entry
        }

        var outputDot = function() {
            var endpoints = buildEndpoints()
            var s = "digraph g {\n"

            for (var k in endpoints) {
                var endpoint = endpoints[k]
                s += "subgraph cluster_" + k + " {\n"
                s += 'label = "' + k + '"\n'
                for (var name in endpoint) {
                    s += '"' + k+name + '" [label=' + name + ',shape=box]\n'
                }
                s += "}\n"
            }

            for (var i in matrix) {
                var entry = matrix[i]
                s += '"' + entry.clientHost+entry.clientName + '" -> ' +
                     '"' + entry.serviceHost+entry.serviceName + '" ' +
                     '[label="' + entry.protocol + ":" + entry.serviceAddress + ":" + entry.servicePort + '"]\n'
            }

            return s + "}\n"
        }

        var buildEndpoints = function() {
            for (var i in matrix) {
                var entry = matrix[i]

                if (!(entry.serviceHost in endpoints)) {
                    endpoints[entry.serviceHost] = {}
                }
                endpoints[entry.serviceHost][entry.serviceName] = entry

                if (!(entry.clientHost in endpoints)) {
                    endpoints[entry.clientHost] = {}
                }
                endpoints[entry.clientHost][entry.clientName] = entry
            }

            log("Endpoints: ", endpoints)
            return endpoints
        }

        var topologyMatrix = function(nodes) {
            for (var i in nodes) {
                var node = nodes[i]
                for (var i in node.Metadata.Sockets) {
                    var socket = node.Metadata.Sockets[i]
                    if ((listenPorts.indexOf(socket.RemotePort) != -1) && (socket.State == "ESTABLISHED") || (at && !isListenSocket(socket))) {
                        var peer = getSocketPeer(socket, node.Host)
                        if (!peer) continue

                        addEntry(socket.Protocol,
                                 peer.host, socket.RemoteAddress, socket.RemotePort, peer.socket.Process, peer.socket.Name,
                                 node.Host, socket.LocalAddress, socket.Process, socket.Name )
                    }
                }
            }
        }

        var flowMatrix = function(sockets) {
            for (var flowID in sockets) {
                var entries = sockets[flowID]
                for (var i in entries) {
                    var socket = entries[i]
                    if (listenPorts.indexOf(socket.RemotePort) != -1) {
                        var serverAddress = socket.RemoteAddress
                        var serverPort = socket.RemotePort

                        var node = hostAddresses[serverAddress]
                        if (!node) break

                        var serverSocket = getSocket(node, socket.Protocol, serverAddress, serverPort)
                        if (!serverSocket) break

                        var server = node.Host
                        var serverProc = serverSocket.Process
                        var serverProcName = serverSocket.Name

                        var clientAddress = socket.LocalAddress
                        node = hostAddresses[clientAddress]
                        if (!node) continue

                        var client = node.Host
                        var clientProc = socket.Process
                        var clientProcName = socket.Name

                        addEntry(socket.Protocol,
                                 server, serverAddress, serverPort, serverProc, serverProcName,
                                 client, clientAddress, clientProc, clientProcName)
                    }
                }
            }
        }

        var outputGraph = function() {
            log("Matrix", matrix)

            var endpoints = buildEndpoints()
            var g = new api.Graph()

            for (var i in matrix) {
                var entry = matrix[i]

                var metadata = {
                    "Name": entry.serviceName,
                    "Type": "process",
                    "ServiceType": "server",
                    "Process": entry.serviceProcess,
                    "Ports": []
                }

                var parentNodeID = new UUID(5, "ns:URL", entry.serviceHost+entry.serviceProcess+"service").toString()
                if (parentNodeID in g.nodes) {
                    metadata["Ports"] = g.nodes[parentNodeID].Metadata["Ports"]
                }
                if (metadata["Ports"].indexOf(entry.servicePort) == -1) {
                    metadata["Ports"].push(entry.servicePort)
                }

                var parentNode = new api.GraphNode()
                parentNode.ID = parentNodeID
                parentNode.Host = entry.serviceHost
                parentNode.Metadata = metadata
                g.nodes[parentNodeID] = parentNode

                var ownerEdgeID = new UUID(5, "ns:URL", hosts[entry.serviceHost].ID+parentNodeID).toString()
                var ownerEdge = new api.GraphEdge()
                ownerEdge.ID = ownerEdgeID
                ownerEdge.Host = entry.serviceHost
                ownerEdge.Metadata = { "RelationType": "ownership" }
                ownerEdge.Parent = hosts[entry.serviceHost].ID
                ownerEdge.Child = parentNodeID
                g.edges[ownerEdgeID] = ownerEdge

                var childNodeID = new UUID(5, "ns:URL", entry.clientHost+entry.clientProcess+"client").toString()
                var metadata = {
                    "Name": entry.clientName,
                    "Type": "process",
                    "ServiceType": "client",
                    "Process": entry.clientProcess
                }
                var childNode = new api.GraphNode()
                childNode.ID = childNodeID
                childNode.Host = entry.clientHost
                childNode.Metadata = metadata
                g.nodes[childNodeID] = childNode

                ownerEdgeID = new UUID(5, "ns:URL", hosts[entry.clientHost].ID+childNodeID).toString()
                ownerEdge = new api.GraphEdge()
                ownerEdge.ID = ownerEdgeID
                ownerEdge.Host =  entry.clientHost
                ownerEdge.Metadata = { "RelationType": "ownership" }
                ownerEdge.Parent = hosts[entry.clientHost].ID
                ownerEdge.Child = childNodeID
                g.edges[ownerEdgeID] = ownerEdge

                var edgeID = new UUID(5, "ns:URL", childNodeID+parentNodeID+entry.servicePort).toString()
                var edge = new api.GraphEdge()
                edge.ID = edgeID
                edge.Host = entry.serviceHost
                edge.Metadata = { "RelationType": "connection",
                                  "Port": entry.servicePort }
                edge.Parent = childNodeID
                edge.Child = parentNodeID

                g.edges[edgeID] = edge
            }

            return g
        }

        var outputResult = function() {
            log("Output result in " + format)

            try {
                switch (format) {
                    case "dot":
                        return outputDot()
                    case "graph":
                        return outputGraph()
                }
            } catch(error) {
                log("Error", error)
            }
        }

        var nodes = G.V().Has('Type', 'host').HasKey('Sockets').result()
        var interfaces = G.V().HasKey('IPV4').result()
        buildCaches(nodes, interfaces)
        topologyMatrix(nodes)
        if (useFlows) {
            var sockets = G.Flows().Has('Transport', NE('')).Sockets().result()
            flowMatrix(sockets)
        }
        return outputResult()
    }