def recv_udp(listener, bufsize):
     debug3('Accept UDP using socket_ext recvmsg.\n')
     srcip, data, adata, flags = listener.recvmsg(
         (bufsize,), socket.CMSG_SPACE(24))
     dstip = None
     family = None
     for a in adata:
         if a.cmsg_level == socket.SOL_IP and a.cmsg_type == IP_ORIGDSTADDR:
             family, port = struct.unpack('=HH', a.cmsg_data[0:4])
             port = socket.htons(port)
             if family == socket.AF_INET:
                 start = 4
                 length = 4
             else:
                 raise Fatal("Unsupported socket type '%s'" % family)
             ip = socket.inet_ntop(
                 family, a.cmsg_data[start:start + length])
             dstip = (ip, port)
             break
         elif a.cmsg_level == SOL_IPV6 and a.cmsg_type == IPV6_ORIGDSTADDR:
             family, port = struct.unpack('=HH', a.cmsg_data[0:4])
             port = socket.htons(port)
             if family == socket.AF_INET6:
                 start = 8
                 length = 16
             else:
                 raise Fatal("Unsupported socket type '%s'" % family)
             ip = socket.inet_ntop(
                 family, a.cmsg_data[start:start + length])
             dstip = (ip, port)
             break
     return (srcip, dstip, data[0])

def _check_nmb(hostname, is_workgroup, is_master):
    return
    global _nmb_ok
    if not _nmb_ok:
        return
    argv = ['nmblookup'] + ['-M'] * is_master + ['--', hostname]
    debug2(' > n%d%d: %s\n' % (is_workgroup, is_master, hostname))
    try:
        p = ssubprocess.Popen(argv, stdout=ssubprocess.PIPE, stderr=null)
        lines = p.stdout.readlines()
        rv = p.wait()
    except OSError:
        _, e = sys.exc_info()[:2]
        log('%r failed: %r\n' % (argv, e))
        _nmb_ok = False
        return
    if rv:
        log('%r returned %d\n' % (argv, rv))
        return
    for line in lines:
        m = re.match(r'(\d+\.\d+\.\d+\.\d+) (\w+)<\w\w>\n', line)
        if m:
            g = m.groups()
            (ip, name) = (g[0], g[1].lower())
            debug3('<    %s -> %s\n' % (name, ip))
            if is_workgroup:
                _enqueue(_check_smb, ip)
            else:
                found_host(name, ip)
                check_host(name)

def _check_revdns(ip):
    debug2(' > rev: %s\n' % ip)
    try:
        r = socket.gethostbyaddr(ip)
        debug3('<    %s\n' % r[0])
        check_host(r[0])
        found_host(r[0], ip)
    except socket.herror:
        pass

def _check_dns(hostname):
    debug2(' > dns: %s\n' % hostname)
    try:
        ip = socket.gethostbyname(hostname)
        debug3('<    %s\n' % ip)
        check_host(ip)
        found_host(hostname, ip)
    except socket.gaierror:
        pass

def _nb_clean(func, *args):
    try:
        return func(*args)
    except OSError as e:
        if e.errno not in (errno.EWOULDBLOCK, errno.EAGAIN):
            raise
        else:
            debug3('%s: err was: %s\n' % (func.__name__, e))
            return None

def _check_smb(hostname):
    return
    global _smb_ok
    if not _smb_ok:
        return
    argv = ['smbclient', '-U', '%', '-L', hostname]
    debug2(' > smb: %s\n' % hostname)
    try:
        p = ssubprocess.Popen(argv, stdout=ssubprocess.PIPE, stderr=null)
        lines = p.stdout.readlines()
        p.wait()
    except OSError:
        _, e = sys.exc_info()[:2]
        log('%r failed: %r\n' % (argv, e))
        _smb_ok = False
        return

    lines.reverse()

    # junk at top
    while lines:
        line = lines.pop().strip()
        if re.match(r'Server\s+', line):
            break

    # server list section:
    #    Server   Comment
    #    ------   -------
    while lines:
        line = lines.pop().strip()
        if not line or re.match(r'-+\s+-+', line):
            continue
        if re.match(r'Workgroup\s+Master', line):
            break
        words = line.split()
        hostname = words[0].lower()
        debug3('<    %s\n' % hostname)
        check_host(hostname)

    # workgroup list section:
    #   Workgroup  Master
    #   ---------  ------
    while lines:
        line = lines.pop().strip()
        if re.match(r'-+\s+', line):
            continue
        if not line:
            break
        words = line.split()
        (workgroup, hostname) = (words[0].lower(), words[1].lower())
        debug3('<    group(%s) -> %s\n' % (workgroup, hostname))
        check_host(hostname)
        check_workgroup(workgroup)

    if lines:
        assert(0)

def _check_smb(hostname):
    return
    global _smb_ok
    if not _smb_ok:
        return
    argv = ["smbclient", "-U", "%", "-L", hostname]
    debug2(" > smb: %s\n" % hostname)
    try:
        p = ssubprocess.Popen(argv, stdout=ssubprocess.PIPE, stderr=null)
        lines = p.stdout.readlines()
        p.wait()
    except OSError as e:
        log("%r failed: %r\n" % (argv, e))
        _smb_ok = False
        return

    lines.reverse()

    # junk at top
    while lines:
        line = lines.pop().strip()
        if re.match(r"Server\s+", line):
            break

    # server list section:
    #    Server   Comment
    #    ------   -------
    while lines:
        line = lines.pop().strip()
        if not line or re.match(r"-+\s+-+", line):
            continue
        if re.match(r"Workgroup\s+Master", line):
            break
        words = line.split()
        hostname = words[0].lower()
        debug3("<    %s\n" % hostname)
        check_host(hostname)

    # workgroup list section:
    #   Workgroup  Master
    #   ---------  ------
    while lines:
        line = lines.pop().strip()
        if re.match(r"-+\s+", line):
            continue
        if not line:
            break
        words = line.split()
        (workgroup, hostname) = (words[0].lower(), words[1].lower())
        debug3("<    group(%s) -> %s\n" % (workgroup, hostname))
        check_host(hostname)
        check_workgroup(workgroup)

    if lines:
        assert 0

 def __init__(self, rsock, wsock, connect_to=None, peername=None):
     global _swcount
     _swcount += 1
     debug3('creating new SockWrapper (%d now exist)\n' % _swcount)
     self.exc = None
     self.rsock = rsock
     self.wsock = wsock
     self.shut_read = self.shut_write = False
     self.buf = []
     self.connect_to = connect_to
     self.peername = peername or _try_peername(self.rsock)
     self.try_connect()

def _check_etc_hosts():
    debug2(' > hosts\n')
    for line in open('/etc/hosts'):
        line = re.sub(r'#.*', '', line)
        words = line.strip().split()
        if not words:
            continue
        ip = words[0]
        names = words[1:]
        if _is_ip(ip):
            debug3('<    %s %r\n' % (ip, names))
            for n in names:
                check_host(n)
                found_host(n, ip)

def _check_netstat():
    debug2(' > netstat\n')
    argv = ['netstat', '-n']
    try:
        p = ssubprocess.Popen(argv, stdout=ssubprocess.PIPE, stderr=null)
        content = p.stdout.read().decode("ASCII")
        p.wait()
    except OSError as e:
        log('%r failed: %r\n' % (argv, e))
        return

    for ip in re.findall(r'\d+\.\d+\.\d+\.\d+', content):
        debug3('<    %s\n' % ip)
        check_host(ip)

def _check_etc_hosts():
    debug2(" > hosts\n")
    for line in open("/etc/hosts"):
        line = re.sub(r"#.*", "", line)
        words = line.strip().split()
        if not words:
            continue
        ip = words[0]
        names = words[1:]
        if _is_ip(ip):
            debug3("<    %s %r\n" % (ip, names))
            for n in names:
                check_host(n)
                found_host(n, ip)

def _check_netstat():
    debug2(" > netstat\n")
    argv = ["netstat", "-n"]
    try:
        p = ssubprocess.Popen(argv, stdout=ssubprocess.PIPE, stderr=null)
        content = p.stdout.read()
        p.wait()
    except OSError as e:
        log("%r failed: %r\n" % (argv, e))
        return

    for ip in re.findall(r"\d+\.\d+\.\d+\.\d+", content):
        debug3("<    %s\n" % ip)
        check_host(ip)

def expire_connections(now, mux):
    for chan, timeout in dnsreqs.items():
        if timeout < now:
            debug3('expiring dnsreqs channel=%d\n' % chan)
            del mux.channels[chan]
            del dnsreqs[chan]
    debug3('Remaining DNS requests: %d\n' % len(dnsreqs))
    for peer, (chan, timeout) in udp_by_src.items():
        if timeout < now:
            debug3('expiring UDP channel channel=%d peer=%r\n' % (chan, peer))
            mux.send(chan, ssnet.CMD_UDP_CLOSE, '')
            del mux.channels[chan]
            del udp_by_src[peer]
    debug3('Remaining UDP channels: %d\n' % len(udp_by_src))

 def try_connect(self):
     if self.connect_to and self.shut_write:
         self.noread()
         self.connect_to = None
     if not self.connect_to:
         return  # already connected
     self.rsock.setblocking(False)
     debug3('%r: trying connect to %r\n' % (self, self.connect_to))
     try:
         self.rsock.connect(self.connect_to)
         # connected successfully (Linux)
         self.connect_to = None
     except socket.error:
         _, e = sys.exc_info()[:2]
         debug3('%r: connect result: %s\n' % (self, e))
         if e.args[0] == errno.EINVAL:
             # this is what happens when you call connect() on a socket
             # that is now connected but returned EINPROGRESS last time,
             # on BSD, on python pre-2.5.1.  We need to use getsockopt()
             # to get the "real" error.  Later pythons do this
             # automatically, so this code won't run.
             realerr = self.rsock.getsockopt(socket.SOL_SOCKET,
                                             socket.SO_ERROR)
             e = socket.error(realerr, os.strerror(realerr))
             debug3('%r: fixed connect result: %s\n' % (self, e))
         if e.args[0] in [errno.EINPROGRESS, errno.EALREADY]:
             pass  # not connected yet
         elif e.args[0] == 0:
             # connected successfully (weird Linux bug?)
             # Sometimes Linux seems to return EINVAL when it isn't
             # invalid.  This *may* be caused by a race condition
             # between connect() and getsockopt(SO_ERROR) (ie. it
             # finishes connecting in between the two, so there is no
             # longer an error).  However, I'm not sure of that.
             #
             # I did get at least one report that the problem went away
             # when we added this, however.
             self.connect_to = None
         elif e.args[0] == errno.EISCONN:
             # connected successfully (BSD)
             self.connect_to = None
         elif e.args[0] in NET_ERRS + [errno.EACCES, errno.EPERM]:
             # a "normal" kind of error
             self.connect_to = None
             self.seterr(e)
         else:
             raise  # error we've never heard of?!  barf completely.

def _handle_diversion(divertsock, dnsport):
    p, tag = divertsock.recvfrom(4096)
    src, dst = _udp_unpack(p)
    debug3('got diverted packet from %r to %r\n' % (src, dst))
    if dst[1] == 53:
        # outgoing DNS
        debug3('...packet is a DNS request.\n')
        _real_dns_server[0] = dst
        dst = ('127.0.0.1', dnsport)
    elif src[1] == dnsport:
        if islocal(src[0], divertsock.family):
            debug3('...packet is a DNS response.\n')
            src = _real_dns_server[0]
    else:
        log('weird?! unexpected divert from %r to %r\n' % (src, dst))
        assert(0)
    newp = _udp_repack(p, src, dst)
    divertsock.sendto(newp, tag)

 def add_rules(anchor, rules):
     assert isinstance(rules, bytes)
     debug3("rules:\n" + rules.decode("ASCII"))
     pfctl('-a %s -f /dev/stdin' % anchor, rules)

#.........这里部分代码省略.........
    hw.leftover = b('')

    def hostwatch_ready(sock):
        assert(hw.pid)
        content = hw.sock.recv(4096)
        if content:
            lines = (hw.leftover + content).split(b('\n'))
            if lines[-1]:
                # no terminating newline: entry isn't complete yet!
                hw.leftover = lines.pop()
                lines.append(b(''))
            else:
                hw.leftover = b('')
            mux.send(0, ssnet.CMD_HOST_LIST, b('\n').join(lines))
        else:
            raise Fatal('hostwatch process died')

    def got_host_req(data):
        if not hw.pid:
            (hw.pid, hw.sock) = start_hostwatch(data.strip().split())
            handlers.append(Handler(socks=[hw.sock],
                                    callback=hostwatch_ready))
    mux.got_host_req = got_host_req

    def new_channel(channel, data):
        (family, dstip, dstport) = data.decode("ASCII").split(',', 2)
        family = int(family)
        dstport = int(dstport)
        outwrap = ssnet.connect_dst(family, dstip, dstport)
        handlers.append(Proxy(MuxWrapper(mux, channel), outwrap))
    mux.new_channel = new_channel

    dnshandlers = {}

    def dns_req(channel, data):
        debug2('Incoming DNS request channel=%d.\n' % channel)
        h = DnsProxy(mux, channel, data)
        handlers.append(h)
        dnshandlers[channel] = h
    mux.got_dns_req = dns_req

    udphandlers = {}

    def udp_req(channel, cmd, data):
        debug2('Incoming UDP request channel=%d, cmd=%d\n' % (channel, cmd))
        if cmd == ssnet.CMD_UDP_DATA:
            (dstip, dstport, data) = data.split(",", 2)
            dstport = int(dstport)
            debug2('is incoming UDP data. %r %d.\n' % (dstip, dstport))
            h = udphandlers[channel]
            h.send((dstip, dstport), data)
        elif cmd == ssnet.CMD_UDP_CLOSE:
            debug2('is incoming UDP close\n')
            h = udphandlers[channel]
            h.ok = False
            del mux.channels[channel]

    def udp_open(channel, data):
        debug2('Incoming UDP open.\n')
        family = int(data)
        mux.channels[channel] = lambda cmd, data: udp_req(channel, cmd, data)
        if channel in udphandlers:
            raise Fatal('UDP connection channel %d already open' % channel)
        else:
            h = UdpProxy(mux, channel, family)
            handlers.append(h)
            udphandlers[channel] = h
    mux.got_udp_open = udp_open

    while mux.ok:
        if hw.pid:
            assert(hw.pid > 0)
            (rpid, rv) = os.waitpid(hw.pid, os.WNOHANG)
            if rpid:
                raise Fatal(
                    'hostwatch exited unexpectedly: code 0x%04x\n' % rv)

        ssnet.runonce(handlers, mux)
        if latency_control:
            mux.check_fullness()

        if dnshandlers:
            now = time.time()
            remove = []
            for channel, h in dnshandlers.items():
                if h.timeout < now or not h.ok:
                    debug3('expiring dnsreqs channel=%d\n' % channel)
                    remove.append(channel)
                    h.ok = False
            for channel in remove:
                del dnshandlers[channel]
        if udphandlers:
            remove = []
            for channel, h in udphandlers.items():
                if not h.ok:
                    debug3('expiring UDP channel=%d\n' % channel)
                    remove.append(channel)
                    h.ok = False
            for channel in remove:
                del udphandlers[channel]

    def setup_firewall(self, port, dnsport, nslist, family, subnets, udp):
        tables = []
        translating_rules = []
        filtering_rules = []

        if family != socket.AF_INET:
            raise Exception(
                'Address family "%s" unsupported by pf method_name'
                % family_to_string(family))
        if udp:
            raise Exception("UDP not supported by pf method_name")

        if len(subnets) > 0:
            includes = []
            # If a given subnet is both included and excluded, list the
            # exclusion first; the table will ignore the second, opposite
            # definition
            for f, swidth, sexclude, snet in sorted(
                    subnets, key=lambda s: (s[1], s[2]), reverse=True):
                includes.append(b"%s%s/%d" %
                                (b"!" if sexclude else b"",
                                    snet.encode("ASCII"),
                                    swidth))

            tables.append(
                b'table <forward_subnets> {%s}' % b','.join(includes))
            translating_rules.append(
                b'rdr pass on lo0 proto tcp '
                b'to <forward_subnets> -> 127.0.0.1 port %r' % port)
            filtering_rules.append(
                b'pass out route-to lo0 inet proto tcp '
                b'to <forward_subnets> keep state')

        if len(nslist) > 0:
            tables.append(
                b'table <dns_servers> {%s}' %
                b','.join([ns[1].encode("ASCII") for ns in nslist]))
            translating_rules.append(
                b'rdr pass on lo0 proto udp to '
                b'<dns_servers> port 53 -> 127.0.0.1 port %r' % dnsport)
            filtering_rules.append(
                b'pass out route-to lo0 inet proto udp to '
                b'<dns_servers> port 53 keep state')

        rules = b'\n'.join(tables + translating_rules + filtering_rules) \
                + b'\n'
        assert isinstance(rules, bytes)
        debug3("rules:\n" + rules.decode("ASCII"))

        pf_status = pfctl('-s all')[0]
        if b'\nrdr-anchor "sshuttle" all\n' not in pf_status:
            pf_add_anchor_rule(osdefs.PF_RDR, b"sshuttle")
        if b'\nanchor "sshuttle" all\n' not in pf_status:
            pf_add_anchor_rule(osdefs.PF_PASS, b"sshuttle")

        pfctl('-a sshuttle -f /dev/stdin', rules)
        if osdefs.platform == "darwin":
            o = pfctl('-E')
            _pf_context['Xtoken'] = \
                re.search(b'Token : (.+)', o[1]).group(1)
        elif b'INFO:\nStatus: Disabled' in pf_status:
            pfctl('-e')
            _pf_context['started_by_sshuttle'] = True

 def add_rules(self, rules):
     assert isinstance(rules, bytes)
     debug3("rules:\n" + rules.decode("ASCII"))
     pfctl('-a sshuttle -f /dev/stdin', rules)

def udp_done(chan, data, method, sock, dstip):
    (src, srcport, data) = data.split(b",", 2)
    srcip = (src, int(srcport))
    debug3('doing send from %r to %r\n' % (srcip, dstip,))
    method.send_udp(sock, srcip, dstip, data)