<commit_before>import errno
from idiokit import threado, sockets
from abusehelper.core import bot, events
from combiner import Expert

DEFAULT_KEYS = ("domain", "ip", "first seen", "last seen")

@threado.stream
def collect(inner):
    result = list()
    while True:
        try:
            data = yield inner
        except sockets.error, error:
            if error.errno == errno.ECONNRESET:
                break
            raise

        result.append(data)
    inner.finish("".join(result))

@threado.stream
def lookup(inner, host, port, eid, name, keys=DEFAULT_KEYS):
    sock = sockets.Socket()
    try:
        yield inner.sub(sock.connect((host, port)))
        sock.send(name + "\r\n")
        data = yield inner.sub(sock | collect())
    except sockets.error, error:
        return

    for line in data.splitlines():
        event = events.Event()
        for key, value in zip(keys, line.split("\t")):
            event.add(key, value)
        yield inner.send(eid, event)

class PassiveDNSExpert(Expert):
    host = bot.Param()
    port = bot.IntParam(default=43)

    @threado.stream
    def augment(inner, self):
        while True:
            eid, event = yield inner
            
            for name in set(event.values("domain") + 
                            event.values("ip") + 
                            event.values("soa")):
                yield inner.sub(lookup(self.host, self.port, eid, name))

if __name__ == "__main__":
    PassiveDNSExpert.from_command_line().execute()<commit_msg>contrib/experts/passivedns.py: Fix, don't forward incoming (eventid, event) pairs into the socket.<commit_after>import errno
from idiokit import threado, sockets
from abusehelper.core import bot, events
from combiner import Expert

DEFAULT_KEYS = ("domain", "ip", "first seen", "last seen")

@threado.stream
def collect(inner):
    result = list()
    while True:
        try:
            data = yield inner
        except sockets.error, error:
            if error.errno == errno.ECONNRESET:
                break
            raise

        result.append(data)
    inner.finish("".join(result))

@threado.stream
def lookup(inner, host, port, eid, name, keys=DEFAULT_KEYS):
    sock = sockets.Socket()
    try:
        yield inner.sub(sock.connect((host, port)))
        sock.send(name + "\r\n")
        data = yield inner.sub(sock | collect())
    except sockets.error, error:
        return

    for line in data.splitlines():
        event = events.Event()
        for key, value in zip(keys, line.split("\t")):
            event.add(key, value)
        yield inner.send(eid, event)

class PassiveDNSExpert(Expert):
    host = bot.Param()
    port = bot.IntParam(default=43)

    def augment(self):
        channel = threado.Channel()

        @threado.stream
        def collect(inner):
            while True:
                eid, event = yield inner

                for name in set(event.values("domain") + 
                                event.values("ip") + 
                                event.values("soa")):
                    channel.send(eid, name)

        @threado.stream
        def work(inner):
            while True:
                eid, name = yield inner, channel
                yield inner.sub(lookup(self.host, self.port, eid, name))

        return collect() | work()

if __name__ == "__main__":
    PassiveDNSExpert.from_command_line().execute()
{'fraction_non_alphanumeric': 0.11723122238586156, 'fraction_numerical': 0.00117820324005891, 'mean_word_length': 2.064981949458484, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '97082', 'n_tokens_mistral': 1013, 'n_tokens_neox': 975, 'n_words': 259}