<commit_before>import socket
import asyncio
import logging

import collections

logger = logging.getLogger(__name__)

import synapse.exc as s_exc
import synapse.common as s_common

import synapse.lib.base as s_base
import synapse.lib.const as s_const
import synapse.lib.msgpack as s_msgpack

readsize = 10 * s_const.megabyte

async def connect(host, port, ssl=None, hostname=None):
    '''
    Async connect and return a Link().
    '''
    info = {'host': host, 'port': port, 'ssl': ssl, 'hostname': hostname}
    reader, writer = await asyncio.open_connection(host, port, ssl=ssl, server_hostname=hostname)
    return await Link.anit(reader, writer, info=info)

async def listen(host, port, onlink, ssl=None):
    '''
    Listen on the given host/port and fire onlink(Link).

    Returns a server object that contains the listening sockets
    '''
    async def onconn(reader, writer):
        info = {'tls': bool(ssl)}
        link = await Link.anit(reader, writer, info=info)
        link.schedCoro(onlink(link))

    server = await asyncio.start_server(onconn, host=host, port=port, ssl=ssl)
    return server

async def unixlisten(path, onlink):
    '''
    Start an PF_UNIX server listening on the given path.
    '''
    info = {'path': path, 'unix': True}

    async def onconn(reader, writer):
        link = await Link.anit(reader, writer, info=info)
        link.schedCoro(onlink(link))
    return await asyncio.start_unix_server(onconn, path=path)

async def unixconnect(path):
    '''
    Connect to a PF_UNIX server listening on the given path.
    '''
    reader, writer = await asyncio.open_unix_connection(path=path)
    info = {'path': path, 'unix': True}
    return await Link.anit(reader, writer, info=info)

async def linkfile(mode='wb'):
    '''
    Connect a socketpair to a file-object and return (link, file).
    '''
    sock0, sock1 = socket.socketpair()

    file1 = sock1.makefile(mode)
    sock1.close()

    reader, writer = await asyncio.open_connection(sock=sock0)
    link0 = await Link.anit(reader, writer, info={'unix': True})

    return link0, file1

async def linksock():
    '''
    Connect a Link, socket pair.
    '''
    sock0, sock1 = socket.socketpair()
    reader, writer = await asyncio.open_connection(sock=sock0)
    link0 = await Link.anit(reader, writer, info={'unix': True})
    return link0, sock1

async def fromspawn(spawninfo):
    sock = spawninfo.get('sock')
    info = spawninfo.get('info', {})
    info['spawn'] = True
    reader, writer = await asyncio.open_connection(sock=sock)
    return await Link.anit(reader, writer, info=info)

class Link(s_base.Base):
    '''
    A Link() is created to wrap a socket reader/writer.
    '''
    async def __anit__(self, reader, writer, info=None):

        await s_base.Base.__anit__(self)

        self.iden = s_common.guid()

        self.reader = reader
        self.writer = writer

        self.rxqu = collections.deque()

        self.sock = self.writer.get_extra_info('socket')
        self.peercert = self.writer.get_extra_info('peercert')

        self._drain_lock = asyncio.Lock()

        if info is None:
            info = {}

        if not info.get('unix'):

            # disable nagle ( to minimize latency for small xmit )
            self.sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)

            # enable TCP keep alives...
            self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
            if hasattr(socket, 'TCP_KEEPIDLE'):
                # start sending a keep alives after 3 sec of inactivity
                self.sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 3)
                # send keep alives every 3 seconds once started
                self.sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 3)
                # close the socket after 5 failed keep alives (15 sec)
                self.sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 5)

        self.info = info
        self.hostname = self.info.get('hostname')

        self.unpk = s_msgpack.Unpk()

        async def fini():
            self.writer.close()
            try:
                await self.writer.wait_closed()
            except (BrokenPipeError, ConnectionResetError) as e:
                logger.debug('Link error waiting on close: %s', str(e))

        self.onfini(fini)

        if self.hostname is not None:
            if self.hostname != self.getTlsPeerCn():
                mesg = f'Expected: {self.hostname} Got: {self.getTlsPeerCn()}'
                await self.fini()
                raise s_exc.BadCertHost(mesg=mesg)

    def getTlsPeerCn(self):

        if self.peercert is None:
            return None

        for items in self.peercert.get('subject', ()):
            for name, valu in items:
                if name == 'commonName':
                    return valu

    async def getSpawnInfo(self):
        info = {}

        # selectively add info for pickle...
        if self.info.get('unix'):
            info['unix'] = True

        if self.info.get('tls'):
            info['unix'] = True
            link0, sock = await linksock()
            link0.onfini(sock.close)

            async def relay(link):
                async with link:
                    while True:
                        byts = await link.recv(1024)
                        if not byts:
                            break
                        await self.send(byts)

                    if not self.isfini:
                        await self.fini()

            self.schedCoro(relay(link0))

        else:
            sock = self.reader._transport._sock

        return {
            'info': info,
            'sock': sock,
        }

    def getAddrInfo(self):
        '''
        Get a summary of address information related to the link.
        '''
        ret = {'family': 'tcp',
               'addr': self.sock.getpeername(),
               }
        # Set family information
        if self.info.get('unix'):
            ret['family'] = 'unix'
            # Unix sockets don't use getpeername
            ret['addr'] = self.sock.getsockname()
        elif self.info.get('tls'):
            ret['family'] = 'tls'
        # Set ipver if needed
        if self.sock.family == socket.AF_INET:
            ret['ipver'] = 'ipv4'<commit_msg>remove incorrect link fini (#2141)<commit_after>import socket
import asyncio
import logging

import collections

logger = logging.getLogger(__name__)

import synapse.exc as s_exc
import synapse.common as s_common

import synapse.lib.base as s_base
import synapse.lib.const as s_const
import synapse.lib.msgpack as s_msgpack

readsize = 10 * s_const.megabyte

async def connect(host, port, ssl=None, hostname=None):
    '''
    Async connect and return a Link().
    '''
    info = {'host': host, 'port': port, 'ssl': ssl, 'hostname': hostname}
    reader, writer = await asyncio.open_connection(host, port, ssl=ssl, server_hostname=hostname)
    return await Link.anit(reader, writer, info=info)

async def listen(host, port, onlink, ssl=None):
    '''
    Listen on the given host/port and fire onlink(Link).

    Returns a server object that contains the listening sockets
    '''
    async def onconn(reader, writer):
        info = {'tls': bool(ssl)}
        link = await Link.anit(reader, writer, info=info)
        link.schedCoro(onlink(link))

    server = await asyncio.start_server(onconn, host=host, port=port, ssl=ssl)
    return server

async def unixlisten(path, onlink):
    '''
    Start an PF_UNIX server listening on the given path.
    '''
    info = {'path': path, 'unix': True}

    async def onconn(reader, writer):
        link = await Link.anit(reader, writer, info=info)
        link.schedCoro(onlink(link))
    return await asyncio.start_unix_server(onconn, path=path)

async def unixconnect(path):
    '''
    Connect to a PF_UNIX server listening on the given path.
    '''
    reader, writer = await asyncio.open_unix_connection(path=path)
    info = {'path': path, 'unix': True}
    return await Link.anit(reader, writer, info=info)

async def linkfile(mode='wb'):
    '''
    Connect a socketpair to a file-object and return (link, file).
    '''
    sock0, sock1 = socket.socketpair()

    file1 = sock1.makefile(mode)
    sock1.close()

    reader, writer = await asyncio.open_connection(sock=sock0)
    link0 = await Link.anit(reader, writer, info={'unix': True})

    return link0, file1

async def linksock():
    '''
    Connect a Link, socket pair.
    '''
    sock0, sock1 = socket.socketpair()
    reader, writer = await asyncio.open_connection(sock=sock0)
    link0 = await Link.anit(reader, writer, info={'unix': True})
    return link0, sock1

async def fromspawn(spawninfo):
    sock = spawninfo.get('sock')
    info = spawninfo.get('info', {})
    info['spawn'] = True
    reader, writer = await asyncio.open_connection(sock=sock)
    return await Link.anit(reader, writer, info=info)

class Link(s_base.Base):
    '''
    A Link() is created to wrap a socket reader/writer.
    '''
    async def __anit__(self, reader, writer, info=None):

        await s_base.Base.__anit__(self)

        self.iden = s_common.guid()

        self.reader = reader
        self.writer = writer

        self.rxqu = collections.deque()

        self.sock = self.writer.get_extra_info('socket')
        self.peercert = self.writer.get_extra_info('peercert')

        self._drain_lock = asyncio.Lock()

        if info is None:
            info = {}

        if not info.get('unix'):

            # disable nagle ( to minimize latency for small xmit )
            self.sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)

            # enable TCP keep alives...
            self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
            if hasattr(socket, 'TCP_KEEPIDLE'):
                # start sending a keep alives after 3 sec of inactivity
                self.sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 3)
                # send keep alives every 3 seconds once started
                self.sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 3)
                # close the socket after 5 failed keep alives (15 sec)
                self.sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 5)

        self.info = info
        self.hostname = self.info.get('hostname')

        self.unpk = s_msgpack.Unpk()

        async def fini():
            self.writer.close()
            try:
                await self.writer.wait_closed()
            except (BrokenPipeError, ConnectionResetError) as e:
                logger.debug('Link error waiting on close: %s', str(e))

        self.onfini(fini)

        if self.hostname is not None:
            if self.hostname != self.getTlsPeerCn():
                mesg = f'Expected: {self.hostname} Got: {self.getTlsPeerCn()}'
                await self.fini()
                raise s_exc.BadCertHost(mesg=mesg)

    def getTlsPeerCn(self):

        if self.peercert is None:
            return None

        for items in self.peercert.get('subject', ()):
            for name, valu in items:
                if name == 'commonName':
                    return valu

    async def getSpawnInfo(self):
        info = {}

        # selectively add info for pickle...
        if self.info.get('unix'):
            info['unix'] = True

        if self.info.get('tls'):
            info['unix'] = True
            link0, sock = await linksock()
            link0.onfini(sock.close)

            async def relay(link):
                async with link:
                    while True:
                        byts = await link.recv(1024)
                        if not byts:
                            break
                        await self.send(byts)

            self.schedCoro(relay(link0))

        else:
            sock = self.reader._transport._sock

        return {
            'info': info,
            'sock': sock,
        }

    def getAddrInfo(self):
        '''
        Get a summary of address information related to the link.
        '''
        ret = {'family': 'tcp',
               'addr': self.sock.getpeername(),
               }
        # Set family information
        if self.info.get('unix'):
            ret['family'] = 'unix'
            # Unix sockets don't use getpeername
            ret['addr'] = self.sock.getsockname()
        elif self.info.get('tls'):
            ret['family'] = 'tls'
        # Set ipver if needed
        if self.sock.family == socket.AF_INET:
            ret['ipver'] = 'ipv4'
{'fraction_non_alphanumeric': 0.11079228439343217, 'fraction_numerical': 0.005898294277060417, 'mean_word_length': 2.1359660084978755, 'pattern_counts': {'":': 0, '<': 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': '7278210', 'n_tokens_mistral': 3869, 'n_tokens_neox': 3739, 'n_words': 1006}