README.md

Start:

./manager.py
./data.py 1
./data.py 2
./worker.py 2
./worker.py 2
./worker.py 2
./worker.py 2

data.py

#!/usr/bin/env python

import sys
import zmq
from collections import defaultdict

port = 30001 + int(sys.argv[1])
# Socket to talk to server
context = zmq.Context()
socket = context.socket(zmq.PULL)
socket.bind("tcp://127.0.0.1:%d" % port)

manager = context.socket(zmq.PUB)
manager.connect("tcp://127.0.0.1:30001")

def log(msg):
    manager.send_multipart(["log", msg])
def NOTICE(msg):
    manager.send_multipart(["notice", msg])

# known hosts
HOSTS=set()
def new_host(host):
    if host not in HOSTS:
        print "New host:", host
        log("known host: %s" % host)
        HOSTS.add(host)
###

SCAN_ATTEMPTS = defaultdict(set)
# Scan detection
def scan(src, dest, port):
    print "attempt", src, dest, port
    attempts = SCAN_ATTEMPTS[src]
    attempts.add((dest, port))
    if len(attempts) == 10:
        print "Scan:", src, dest, port
        NOTICE("Scan from %s, %d destinations" % (src, len(attempts)))


print "Waiting..."
while True:
    msg = socket.recv_multipart()
    queue = msg[0]
    if queue == 'new_host':
        new_host(*msg[1:])
    if queue == 'scan':
        scan(*msg[1:])

manager.py

#!/usr/bin/env python

import sys
import zmq

port = 30001
# Socket to talk to server
context = zmq.Context()
socket = context.socket(zmq.SUB)

socket.bind("tcp://127.0.0.1:%d" % port)

socket.setsockopt(zmq.SUBSCRIBE, "notice")
socket.setsockopt(zmq.SUBSCRIBE, "log")

print "Waiting..."
while True:
    msg = socket.recv_multipart()
    print msg

worker.py

#!/usr/bin/env python

import binascii

import sys
import zmq
import time
import random
data_count = int(sys.argv[1])

def random_src(up=255):
    return "1.2.3.%d" % random.randint(1,up)

def random_dest(up=255):
    return "192.168.1.%d" % random.randint(1,up)

def random_scan():
    return random_src(5), random_dest(), str(random.choice((22,5900,3389)))

# Socket to talk to server
context = zmq.Context()
data_sockets = []
for i in range(data_count):
    port = 30002 + i
    socket = context.socket(zmq.PUSH)
    socket.connect("tcp://127.0.0.1:%d" % port)
    data_sockets.append(socket)

def send_unicast_hashed(event, key, *args):
    node = binascii.crc32(key) % len(data_sockets)
    socket = data_sockets[node]
    msg = [event] + list(args)
    socket.send_multipart(msg)

def new_host(ip):
    send_unicast_hashed("new_host", ip, ip)

def new_scan(src, dest, port):
    send_unicast_hashed("scan", src, src, dest, port)

while True:
    new_host(random_dest(20))
    new_scan(*random_scan())
    time.sleep(1)