#!/usr/bin/env python3
"""One interpreter start for the whole of 99-python3-pysocks.

Emits `key=value' lines, one per line, in file order; the suite reads them back
with grep.  A python3 start on this arm is tens of seconds and every measurement
here is a few hundred bytes of string and struct work against an in-memory
socket double, so the start is the entire cost of the suite and taking it once
is the difference between a 30-second run and a fifteen-minute one.

TWO RULES THE VALUES OBEY.

The runner parses its input line by line, so no value may contain a newline.
Every value below is a single line -- which is why the HTTP cases, whose subject
is a CRLF-separated request, emit the wire as HEX or with the separators spelled
out rather than as the text it is.

And an EMPTY RESULT MUST BE DISTINGUISHABLE FROM AN ABSENT KEY.  Several of this
module's answers are empty -- an unset proxy tuple's fields, a request with no
user id, an auth header that was not appended -- and `grep' followed by `cut'
cannot tell "" from a key the probe never reached.  So every string result is
emitted WRAPPED IN ANGLE BRACKETS: `<a/b>', and the empty string is `<>'.  A
refusal is `!<ExceptionName>' instead, and a case the probe skipped is simply
missing, which reads as neither.

THE WIRE IS MEMORY, NOT A PEER.  `socksocket' is a `socket.socket' subclass and
its three negotiators reach the network through exactly four calls --
`self.makefile(mode)', `self.sendall(data)', and, for SOCKS5, `conn.makefile'
on the connection handed in as a parameter.  `Doubled' below overrides those,
so every handshake in this file is driven through the module's own methods with
no peer, no listener and no port: the bytes the negotiator writes land in a
BytesIO, and the bytes it reads are a scripted reply the case supplies.  Nothing
here opens a connection.

ONE THING THIS PROBE MUST NEVER DO: pass a HOSTNAME with `rdns=False'.  That is
the one input on which this library resolves NAMES ITSELF -- `_write_SOCKS5_
address' calls `socket.getaddrinfo' and both `_negotiate_SOCKS4' and
`_negotiate_HTTP' call `socket.gethostbyname' -- and a guest with no resolver
blocks there for the platform's DNS timeout, inside a suite whose whole point is
that it needs no peer.  A hang costs the BATCH its verify, where a wrong answer
would cost one case.  Every hostname below is passed with `rdns=True', and every
`rdns=False' case below passes a NUMERIC address, for which the resolver branch
is never reached.  `rdns-false-is-untested' is the case that says so out loud.
"""

import io
import socket
import struct
import sys

emitted = []


def emit(key, value):
    emitted.append(key)
    print('%s=%s' % (key, value))


def call(fn, *a, **kw):
    """`<result>' on success, `!<ExceptionName>' on a refusal.

    The exception NAME rather than its message: the messages are asserted
    separately, by the cases that are ABOUT the message, so a release that
    reworded the rest does not read as a behavioural change."""
    try:
        return '<%s>' % (fn(*a, **kw),)
    except Exception as exc:
        return '!<%s>' % type(exc).__name__


def flat(text):
    """CR and LF spelled out rather than embedded.

    The runner parses its input line by line, so a value carrying a newline
    lands as one counted result plus stray uncounted lines.  Two values here are
    built from an HTTP status line and CANNOT avoid one: the reason phrase
    arrives with its CRLF still attached, and the 400/403/405 note is joined
    with a newline of the module's own."""
    return text.replace('\\', '\\\\').replace('\r', '\\r').replace('\n', '\\n')


def msg(fn, *a, **kw):
    """`str(exc)', which for a ProxyError is its `msg' attribute.

    NOT `args[0]': a ProxyError has NO `args' at all -- `OSError.__new__' fills
    that in only for subclasses that leave `__init__' alone, and this one does
    not -- so `args[0]' is an IndexError on every exception this module raises.
    `proxyerror-args-empty' asserts it."""
    try:
        fn(*a, **kw)
        return '<no-raise>'
    except Exception as exc:
        return '<%s>' % flat('%s' % (exc,))


try:
    import socks
except Exception as exc:                                        # pragma: no cover
    print('import_error=%s: %s' % (type(exc).__name__, exc))
    sys.exit(1)


class Buf(io.BytesIO):
    """A BytesIO the negotiators are free to close.

    Every negotiator closes its `makefile' handles in a `finally', and a closed
    BytesIO cannot be read back -- so the case that inspects what was written
    would raise instead of answering.  Dropping `close' is what keeps the
    transcript readable after the call returns."""
    def close(self):
        pass


class TextBuf(io.StringIO):
    """The same, for `_negotiate_HTTP', which reads its status line in TEXT
    mode -- `self.makefile()' with no arguments."""
    def close(self):
        pass


class Doubled(socks.socksocket):
    """A socksocket whose wire is memory.

    Overrides the whole of the module's I/O surface: `makefile' in both modes
    and `sendall'.  `_SOCKS5_request' takes its connection as a PARAMETER and
    `_negotiate_SOCKS5' passes `self', so overriding `makefile' here covers the
    SOCKS5 path as well as the SOCKS4 and HTTP ones.

    The object is still a real socksocket -- it holds a real unconnected fd,
    because `_SOCKS5_request' ends by calling `socket.socket.settimeout' through
    `super()' and that has to reach something.  No fd is ever connected, bound,
    read or written."""

    def wire(self, reply=b''):
        self.sent = Buf()
        self._rb = Buf(reply)
        self._rt = TextBuf(reply.decode('latin-1'))
        return self

    def makefile(self, mode='r', buffering=-1, *a, **kw):
        if 'b' in mode:
            return self.sent if ('w' in mode or 'a' in mode) else self._rb
        return self._rt

    def sendall(self, data, *a, **kw):
        self.sent.write(data)


_open = []


def dbl(reply=b'', **proxy):
    d = Doubled()
    _open.append(d)
    if proxy:
        d.set_proxy(**proxy)
    d.wire(reply)
    return d


def hexs(b):
    return '<%s>' % b.hex()


# The scripted SOCKS5 replies the cases below feed back, spelled once here so a
# case reads as the thing it is about rather than as a bytestring.
S5_METHOD_NONE = b'\x05\x00'          # version 5, method 0x00 (no auth)
S5_METHOD_USERPASS = b'\x05\x02'      # version 5, method 0x02 (username/password)
S5_AUTH_OK = b'\x01\x00'              # RFC1929 status 0
S5_OK_V4 = b'\x05\x00\x00' + b'\x01' + bytes([127, 0, 0, 1]) + b'\x1f\x90'


# ============================================================== identity =====
emit('is_flat_module', 'no' if hasattr(socks, '__path__') else 'yes')
emit('socks_module_file', '<%s>' % socks.__file__)
emit('version_attr', '<%s>' % socks.__version__)
emit('version_type', '<%s>' % type(socks.__version__).__name__)

# THE IMPORT NAME IS NOT THE DISTRIBUTION NAME.  The module is `socks' and the
# distribution is `PySocks', so asking importlib.metadata for the name you
# imported is a PackageNotFoundError.  The lowercase spelling IS accepted,
# because importlib.metadata normalises the name it is given -- so the pair
# below says which of the three spellings a caller may rely on.
for _name, _key in (('PySocks', 'dist_version_pysocks'),
                    ('pysocks', 'dist_version_lowercase'),
                    ('socks', 'dist_version_socks')):
    try:
        from importlib.metadata import version as _dv
        emit(_key, '<%s>' % _dv(_name))
    except Exception as exc:
        emit(_key, '!<%s>' % type(exc).__name__)

# THERE IS NO `__all__', so `from socks import *' exports every public name the
# module happens to hold -- including the modules it imported and the two
# variables its class-body loop left behind at module level.  `leaked_name' is
# the loop's index after the last iteration and `leaked_method' the descriptor
# it last looked up; neither is part of any API and both are exported.
emit('has_all', 'yes' if hasattr(socks, '__all__') else 'no')
emit('leaked_name', '<%s>' % getattr(socks, 'name', ''))
emit('leaked_method_type', '<%s>' % type(getattr(socks, 'method', None)).__name__)
emit('star_exports_socket',
     'yes' if getattr(socks, 'socket', None) is socket else 'no')

# THE PYTHON 2 DELEGATION MACHINERY IS INERT HERE.  `_BaseSocket' exists to let
# Python 2 override `send'/`recv', which on that version were slot descriptors
# rather than functions; the module decides at import time by testing each for
# callability.  On Python 3 all four ARE callable, so `_savenames' stays empty
# and `_BaseSocket.__init__' saves and deletes nothing.  A release that changed
# the test would show up here as a non-empty list.
emit('savenames', '<%s>' % ','.join(socks._BaseSocket._savenames))
emit('savenames_count', '%d' % len(socks._BaseSocket._savenames))


# ============================================================= constants =====
emit('socks4_const', '%d' % socks.SOCKS4)
emit('socks5_const', '%d' % socks.SOCKS5)
emit('http_const', '%d' % socks.HTTP)
emit('alias_socks4', 'yes' if socks.PROXY_TYPE_SOCKS4 == socks.SOCKS4 else 'no')
emit('alias_socks5', 'yes' if socks.PROXY_TYPE_SOCKS5 == socks.SOCKS5 else 'no')
emit('alias_http', 'yes' if socks.PROXY_TYPE_HTTP == socks.HTTP else 'no')
emit('proxy_types', '<%s>' % ','.join(
    '%s=%d' % (k, v) for k, v in sorted(socks.PROXY_TYPES.items())))
emit('printable_types', '<%s>' % ','.join(
    '%d=%s' % (k, v) for k, v in sorted(socks.PRINTABLE_PROXY_TYPES.items())))
emit('default_ports', '<%s>' % ','.join(
    '%d=%d' % (k, v) for k, v in sorted(socks.DEFAULT_PORTS.items())))
emit('socks4_error_codes', '<%s>' % ','.join(
    '%#04x' % k for k in sorted(socks.SOCKS4_ERRORS)))
emit('socks5_error_codes', '<%s>' % ','.join(
    '%#04x' % k for k in sorted(socks.SOCKS5_ERRORS)))
emit('socks4_error_5b', '<%s>' % socks.SOCKS4_ERRORS[0x5B])
emit('socks5_error_02', '<%s>' % socks.SOCKS5_ERRORS[0x02])
emit('orig_socket_is_socket',
     'yes' if socks._orig_socket is socket.socket else 'no')
emit('orgsocket_alias',
     'yes' if socks._orgsocket is socks._orig_socket else 'no')


# ======================================================= the exceptions ======
# EVERY ONE OF THEM IS AN OSError, so a caller who wrote `except OSError' around
# a plain socket already catches all six -- which is the drop-in claim's other
# half and is asserted rather than assumed.
emit('proxyerror_is_oserror',
     'yes' if issubclass(socks.ProxyError, OSError) else 'no')
emit('exception_bases', '<%s>' % ','.join(
    '%s:%s' % (c.__name__, c.__bases__[0].__name__)
    for c in (socks.GeneralProxyError, socks.ProxyConnectionError,
              socks.SOCKS5AuthError, socks.SOCKS5Error, socks.SOCKS4Error,
              socks.HTTPError)))

# EVERY EXCEPTION THIS MODULE RAISES IS AN OSError WITH NOTHING IN IT.
# `ProxyError.__init__' sets `self.msg' and never calls its base, and
# `OSError.__new__' populates `args'/`errno'/`strerror' only for a subclass that
# leaves `__init__' alone -- so `args' is the EMPTY TUPLE and `errno' is None.
# `except OSError as e: e.args[0]' is the idiom a caller reaches for after
# catching one, and here it is an IndexError; `e.errno' is the idiom
# `connect_ex' itself uses, and `connect-ex-ipv6' is where that costs something.
# The message is reachable only as `str(e)' or `e.msg'.
_pe = socks.ProxyError('outer', socket.error('inner'))
emit('proxyerror_str', '<%s>' % _pe)
emit('proxyerror_args', '<%s>' % (_pe.args,))
emit('proxyerror_args_len', '%d' % len(_pe.args))
emit('proxyerror_errno', '<%s>' % (_pe.errno,))
emit('proxyerror_strerror', '<%s>' % (_pe.strerror,))
emit('proxyerror_msg_attr', '<%s>' % _pe.msg)
emit('proxyerror_socket_err', '<%s>' % _pe.socket_err)
# The wrapped socket error is APPENDED TO `msg' by the constructor, so `msg' is
# not the string the caller passed -- it is that string plus ": " plus the cause.
emit('proxyerror_msg_is_appended',
     'yes' if _pe.msg == 'outer: inner' else 'no')
_pe2 = socks.ProxyError('bare')
emit('proxyerror_bare_str', '<%s>' % _pe2)
emit('proxyerror_bare_socket_err', '<%s>' % _pe2.socket_err)


# =================================================== socksocket's shape ======
emit('is_socket_subclass',
     'yes' if issubclass(socks.socksocket, socket.socket) else 'no')
emit('basesocket_is_socket_subclass',
     'yes' if issubclass(socks._BaseSocket, socket.socket) else 'no')
emit('mro', '<%s>' % ','.join(c.__name__ for c in socks.socksocket.__mro__))
emit('class_default_proxy', '<%s>' % (socks.socksocket.default_proxy,))

_s = dbl()
emit('instance_is_socket', 'yes' if isinstance(_s, socket.socket) else 'no')
emit('instance_proxy', '<%s>' % (_s.proxy,))
emit('instance_proxy_sockname', '<%s>' % (_s.proxy_sockname,))
emit('instance_proxy_peername', '<%s>' % (_s.proxy_peername,))
emit('instance_family', '%d' % _s.family)
emit('instance_type', '%d' % _s.type)
# THE ATTRIBUTES THE DROP-IN CLAIM RESTS ON.  A caller swaps the class and keeps
# calling the socket API, so the names have to be there; they are inherited, not
# redefined, which is why the list is asserted as a whole rather than one by one.
emit('socket_api_present', '<%s>' % ','.join(
    n for n in ('accept', 'bind', 'close', 'connect', 'connect_ex', 'detach',
                'dup', 'fileno', 'getpeername', 'getsockname', 'getsockopt',
                'listen', 'makefile', 'recv', 'recvfrom', 'send', 'sendall',
                'sendto', 'setblocking', 'setsockopt', 'settimeout', 'shutdown')
    if hasattr(_s, n)))

# THE TYPE CHECK IS THE ONE THING `__init__' DOES BEFORE ITS BASE.  A raw socket
# is refused with a ValueError naming the type it was handed.
emit('raw_refused', call(socks.socksocket, socket.AF_INET, socket.SOCK_RAW))
emit('raw_refused_msg', msg(socks.socksocket, socket.AF_INET, socket.SOCK_RAW))
_dg = socks.socksocket(socket.AF_INET, socket.SOCK_DGRAM)
_open.append(_dg)
emit('dgram_accepted', 'yes' if _dg.type == socket.SOCK_DGRAM else 'no')

# `settimeout' IS SHADOWED and keeps its own `_timeout', which is what
# `gettimeout' answers -- so the value survives even on a socket that was never
# connected, where the base class's own timeout was never applied.
emit('timeout_initial', '<%s>' % _s.gettimeout())
_s.settimeout(12.5)
emit('timeout_roundtrip', '<%s>' % _s.gettimeout())
_s.setblocking(False)
emit('setblocking_false', '<%s>' % _s.gettimeout())
_s.setblocking(True)
emit('setblocking_true', '<%s>' % _s.gettimeout())
_s.settimeout(None)


# ======================================================= set_proxy ===========
# THE CREDENTIALS ARE ENCODED AND THE REST IS NOT.  `set_proxy' stores the
# username and password as UTF-8 BYTES and everything else exactly as passed, so
# the tuple a caller reads back is not the tuple it wrote.
_p = dbl()
_p.set_proxy(socks.SOCKS5, 'proxy.example', 9050, True, 'user', 'pass')
emit('set_proxy_tuple', '<%s>' % (_p.proxy,))
emit('set_proxy_username_type', '<%s>' % type(_p.proxy[4]).__name__)
emit('set_proxy_addr_type', '<%s>' % type(_p.proxy[1]).__name__)
# ...and an EMPTY username is stored as None rather than as b"", because the
# encode is guarded by the string's own truthiness.  That is what makes the
# SOCKS4 `no user id' and SOCKS5 `no auth' branches fire for `username=""'.
_p.set_proxy(socks.SOCKS5, 'p', 1080, True, '', '')
emit('set_proxy_empty_username', '<%s>' % (_p.proxy[4],))
# The Python-2-era alias renames one keyword and forwards the rest.
_p.setproxy(proxytype=socks.SOCKS4, addr='a', port=1)
emit('setproxy_alias_type', '%d' % _p.proxy[0])
emit('setproxy_is_not_set_proxy',
     'yes' if socks.socksocket.setproxy is not socks.socksocket.set_proxy else 'no')

# `_proxy_addr' IS WHERE THE DEFAULT PORT COMES FROM, and an unknown proxy type
# with no port is the one input that turns it into a refusal.
def _paddr(t, port=None):
    d = dbl()
    d.set_proxy(t, 'h', port)
    return d._proxy_addr()

emit('default_port_socks4', '<%s>' % (_paddr(socks.SOCKS4),))
emit('default_port_socks5', '<%s>' % (_paddr(socks.SOCKS5),))
emit('default_port_http', '<%s>' % (_paddr(socks.HTTP),))
emit('explicit_port_wins', '<%s>' % (_paddr(socks.SOCKS5, 3128),))
emit('unknown_type_refused', call(_paddr, 99))
emit('unknown_type_msg', msg(_paddr, 99))
# ...and an unknown type WITH a port is accepted, because the guard is on the
# resolved port and not on the type.  The failure then lands later, in
# `_proxy_negotiators[proxy_type]', as a KeyError rather than a ProxyError.
emit('unknown_type_with_port', '<%s>' % (_paddr(99, 1080),))
emit('negotiator_keys', '<%s>' % ','.join(
    str(k) for k in sorted(socks.socksocket._proxy_negotiators)))


# ================================== the default proxy and wrap_module ========
# `set_default_proxy' WRITES A CLASS ATTRIBUTE, so it reaches every socksocket
# created afterwards and none created before.  Both halves are asserted, because
# a caller who sets it late is a caller whose earlier sockets are unproxied and
# silent about it.
_before = dbl()
emit('default_proxy_initially', '<%s>' % (socks.get_default_proxy(),))
socks.set_default_proxy(socks.SOCKS5, 'd.example', 1080, True, 'du', 'dp')
emit('default_proxy_set', '<%s>' % (socks.get_default_proxy(),))
emit('default_proxy_is_class_attr',
     'yes' if socks.get_default_proxy() is socks.socksocket.default_proxy else 'no')
_after = dbl()
emit('default_proxy_reaches_new', '<%s>' % (_after.proxy,))
emit('default_proxy_misses_old', '<%s>' % (_before.proxy,))
emit('getdefaultproxy_alias',
     'yes' if socks.getdefaultproxy is socks.get_default_proxy else 'no')
emit('setdefaultproxy_is_not_set_default',
     'yes' if socks.setdefaultproxy is not socks.set_default_proxy else 'no')

# `wrap_module' DOES NOT PATCH THE MODULE IT IS GIVEN -- IT PATCHES `socket'.
# The body is `module.socket.socket = socksocket', and `module.socket' is the one
# shared socket MODULE object, so the assignment rebinds `socket.socket' for the
# whole interpreter.  The docstring's "only work on modules that import socket
# directly into the namespace" describes how the module is REACHED, not how far
# the change goes: handed any module that has imported socket, it proxies every
# socket the process creates from then on, including ones in libraries the
# caller never named.
_orig_socket_class = socket.socket
emit('wrapped_module_is_socket_module',
     'yes' if sys.modules[__name__].socket is socket else 'no')
socks.wrap_module(sys.modules[__name__])
emit('wrap_patched_socket_module',
     'yes' if socket.socket is socks.socksocket else 'no')
emit('wrap_patched_globally',
     'yes' if sys.modules['socket'].socket is socks.socksocket else 'no')
emit('wrapmodule_alias', 'yes' if socks.wrapmodule is socks.wrap_module else 'no')
# THERE IS NO `unwrap_module'.  Restoring is the caller's job and the caller has
# to have kept the original class, which nothing in the API hands back.
emit('has_unwrap_module', 'yes' if hasattr(socks, 'unwrap_module') else 'no')
socket.socket = _orig_socket_class
emit('restored_cleanly', 'yes' if socket.socket is _orig_socket_class else 'no')
emit('restored_is_not_socksocket',
     'yes' if socket.socket is not socks.socksocket else 'no')
# ...and a socksocket created after the restore still behaves, which is what
# says the round trip left no residue on the class itself.
emit('post_restore_instance', '<%s>' % (dbl().proxy_sockname,))

socks.socksocket.default_proxy = None
emit('default_proxy_cleared', '<%s>' % (socks.get_default_proxy(),))
# ...and with no default set, `wrap_module' refuses rather than installing a
# socket that would connect direct while the caller believed it was proxied.
emit('wrap_without_default', call(socks.wrap_module, sys.modules[__name__]))
emit('wrap_without_default_msg', msg(socks.wrap_module, sys.modules[__name__]))
emit('wrap_left_socket_alone',
     'yes' if socket.socket is _orig_socket_class else 'no')


# ===================================== the peername pair, which is wrong =====
# `getpeername' IS REBOUND TO `get_peername' in the class body, and
# `get_proxy_peername' -- whose docstring says it returns the PROXY -- is
# `return self.getpeername()'.  That lookup finds the rebinding, so the two
# methods return the same thing: the DESTINATION.  There is no way through this
# API to ask a connected socksocket for the proxy's own address; the base
# class's `getpeername' is shadowed and nothing exposes it.
_pn = dbl()
_pn.proxy_peername = ('dest.example', 443)
_pn.proxy_sockname = ('10.0.0.1', 1080)
emit('get_peername', '<%s>' % (_pn.get_peername(),))
emit('getpeername_alias',
     'yes' if socks.socksocket.getpeername is socks.socksocket.get_peername else 'no')
emit('get_proxy_peername', '<%s>' % (_pn.get_proxy_peername(),))
emit('proxy_peername_equals_peername',
     'yes' if _pn.get_proxy_peername() == _pn.get_peername() else 'no')
emit('get_proxy_sockname', '<%s>' % (_pn.get_proxy_sockname(),))
emit('getproxysockname_alias',
     'yes' if socks.socksocket.getproxysockname is
     socks.socksocket.get_proxy_sockname else 'no')
emit('getproxypeername_alias',
     'yes' if socks.socksocket.getproxypeername is
     socks.socksocket.get_proxy_peername else 'no')


# ================================================ SOCKS5: the handshake ======
# THE GREETING SAYS WHICH METHODS THE CLIENT SUPPORTS, and it is decided by
# whether BOTH a username and a password were set: `05 01 00' offers only "no
# authentication", `05 02 00 02' offers that and username/password.
_d = dbl(S5_METHOD_NONE + S5_OK_V4, proxy_type=socks.SOCKS5, addr='p',
         port=1080)
_d._negotiate_SOCKS5('1.2.3.4', 80)
emit('s5_noauth_wire', hexs(_d.sent.getvalue()))
emit('s5_greeting_noauth', hexs(_d.sent.getvalue()[:3]))
emit('s5_request_ipv4', hexs(_d.sent.getvalue()[3:]))
emit('s5_ipv4_peername', '<%s>' % (_d.proxy_peername,))
emit('s5_ipv4_sockname', '<%s>' % (_d.proxy_sockname,))

# A username WITHOUT a password does not reach the username/password method: the
# offer is guarded by `username and password', so a half-configured proxy sends
# the no-auth greeting and fails later, at the server's choice, rather than here.
_d = dbl(S5_METHOD_NONE + S5_OK_V4, proxy_type=socks.SOCKS5, addr='p',
         port=1080, username='u')
_d._negotiate_SOCKS5('1.2.3.4', 80)
emit('s5_username_only_greeting', hexs(_d.sent.getvalue()[:3]))

_d = dbl(S5_METHOD_USERPASS + S5_AUTH_OK + S5_OK_V4, proxy_type=socks.SOCKS5,
         addr='p', port=1080, username='user', password='pass')
_d._negotiate_SOCKS5('1.2.3.4', 80)
_wire = _d.sent.getvalue()
emit('s5_userpass_greeting', hexs(_wire[:4]))
emit('s5_auth_packet', hexs(_wire[4:15]))
emit('s5_userpass_request', hexs(_wire[15:]))

# THE USERNAME LENGTH IS WRITTEN AS `chr(len(u)).encode()', WHICH IS UTF-8.
# RFC1929's ULEN is ONE byte and holds 1..255, but `chr(200).encode()' is TWO
# bytes -- so every credential of 128 characters or more produces a packet the
# server cannot parse, and it is the CLIENT that built it.  The length below is
# 212 where a correct encoder would write 211.
_lu, _lp = 'u' * 200, 'p' * 8
_d = dbl(S5_METHOD_USERPASS + S5_AUTH_OK + S5_OK_V4, proxy_type=socks.SOCKS5,
         addr='p', port=1080, username=_lu, password=_lp)
_d._negotiate_SOCKS5('1.2.3.4', 80)
_wire = _d.sent.getvalue()
emit('s5_long_user_auth_len', '%d' % (len(_wire) - 4 - 10))
emit('s5_long_user_ulen_bytes', hexs(_wire[5:7]))
emit('s5_long_user_ulen_width', '%d' % len(chr(len(_lu)).encode()))
emit('s5_short_user_ulen_width', '%d' % len(chr(4).encode()))
emit('s5_ulen_boundary', '%d' % min(
    n for n in range(1, 256) if len(chr(n).encode()) > 1))

# THE THREE ADDRESS FORMS.  IPv4 and IPv6 are detected by `inet_pton' and take
# the numeric forms; anything else is a name and takes the domain form, whose
# length prefix has the SAME UTF-8 defect as the username's.
def _s5_addr(host, port=80, **proxy):
    kw = dict(proxy_type=socks.SOCKS5, addr='p', port=1080)
    kw.update(proxy)
    d = dbl(S5_METHOD_NONE + S5_OK_V4, **kw)
    d._negotiate_SOCKS5(host, port)
    return d

_d = _s5_addr('example.com')
emit('s5_domain_request', hexs(_d.sent.getvalue()[3:]))
emit('s5_domain_peername', '<%s>' % (_d.proxy_peername,))
# `encode("idna")' IS A PASS-THROUGH FOR AN ASCII NAME, which is not what the
# call looks like it does.  Python's idna codec tries `input.encode("ascii")'
# first and, where that works, returns those bytes UNCHANGED after a per-label
# length check -- so the name is NOT lowercased, NOT normalised, and a trailing
# root dot is NOT stripped.  Whatever the caller wrote is what the proxy is
# asked for, byte for byte.
_d = _s5_addr('EXAMPLE.COM')
emit('s5_domain_case_preserved', hexs(_d.sent.getvalue()[3:]))
_d = _s5_addr('example.com.')
emit('s5_domain_trailing_dot', hexs(_d.sent.getvalue()[3:]))
# ...and the same fast path accepts an EMPTY name, because it length-checks only
# the labels BEFORE the last one.  The result is a domain request with a
# zero-length name -- `03 00' and a port -- built without a word of complaint
# from either the codec or this module.
emit('s5_domain_empty_host', hexs(_s5_addr('').sent.getvalue()[3:]))
# A NON-ASCII name is where the codec earns its name: it punycodes, so the
# request carries an `xn--' label the caller never wrote.  This is the one input
# for which the encode is not a no-op, which is why both halves are asserted.
emit('s5_domain_punycode', hexs(_s5_addr('b\xfccher.example').sent.getvalue()[3:]))
# A LABEL OVER 63 BYTES is refused by the idna codec, not by this module, so the
# caller sees a UnicodeError from an encode it never asked for.
emit('s5_domain_long_label', call(_s5_addr, 'a' * 64 + '.com'))
# A NAME OF 128 BYTES OR MORE trips the same UTF-8 length defect as the
# username: the prefix is two bytes and the packet is unparseable.
_long = ('a' * 60 + '.') * 2 + 'a' * 10 + '.com'
_d = _s5_addr(_long)
_body = _d.sent.getvalue()[3:]
emit('s5_long_host_len', '%d' % len(_long))
emit('s5_long_host_prefix', hexs(_body[3:6]))
emit('s5_long_host_request_len', '%d' % len(_body))

emit('s5_ipv6_request', call(
    lambda: _s5_addr('::1').sent.getvalue()[3:].hex()))
emit('s5_port_big_endian', hexs(_s5_addr('1.2.3.4', 8080).sent.getvalue()[-2:]))
emit('s5_port_max', hexs(_s5_addr('1.2.3.4', 65535).sent.getvalue()[-2:]))
# `struct.pack(">H", port)' is the only bound on the port, so a port outside
# 0..65535 is a struct.error from the packer rather than a checked argument.
emit('s5_port_out_of_range', call(_s5_addr, '1.2.3.4', 65536))
# rdns=False with a NUMERIC address never reaches the resolver: `inet_pton'
# answers first and returns before `rdns' is consulted.
emit('s5_rdns_false_numeric',
     hexs(_s5_addr('1.2.3.4', rdns=False).sent.getvalue()[3:]))
emit('rdns_false_is_untested', 'by-design')

# THE BOUND ADDRESS COMES BACK IN WHICHEVER FORM THE SERVER CHOSE, and the three
# forms do not agree on a TYPE: an IPv4 reply is parsed with `inet_ntoa' into a
# STR, a domain reply is returned as the raw BYTES `_readall' produced.  A caller
# storing `proxy_sockname[0]' gets str or bytes depending on the server.
_d = dbl(S5_METHOD_NONE + b'\x05\x00\x00\x03\x03abc\x00\x50',
         proxy_type=socks.SOCKS5, addr='p', port=1080)
_d._negotiate_SOCKS5('1.2.3.4', 80)
emit('s5_bnd_domain', '<%s>' % (_d.proxy_sockname,))
emit('s5_bnd_domain_type', '<%s>' % type(_d.proxy_sockname[0]).__name__)
emit('s5_bnd_ipv4_type', '<%s>' % type(_s5_addr('1.2.3.4').proxy_sockname[0]).__name__)

# UDP_ASSOCIATE is the other command this request builder serves, and it is
# reached through `bind' rather than `connect'.  Driving `_SOCKS5_request'
# directly is what lets the command byte be asserted without a relay.
_d = dbl(S5_METHOD_NONE + S5_OK_V4, proxy_type=socks.SOCKS5, addr='p', port=1080)
_resolved, _bnd = _d._SOCKS5_request(_d, b'\x03', ('0', 5000))
emit('s5_udp_cmd_byte', hexs(_d.sent.getvalue()[3:5]))
emit('s5_udp_resolved', '<%s>' % (_resolved,))
emit('s5_udp_bnd', '<%s>' % (_bnd,))


# ============================================ SOCKS5: the refusal paths ======
def _s5_fail(reply, **proxy):
    kw = dict(proxy_type=socks.SOCKS5, addr='p', port=1080)
    kw.update(proxy)
    d = dbl(reply, **kw)
    d._negotiate_SOCKS5('1.2.3.4', 80)

emit('s5_bad_version', call(_s5_fail, b'\x04\x00'))
emit('s5_bad_version_msg', msg(_s5_fail, b'\x04\x00'))
emit('s5_no_methods', call(_s5_fail, b'\x05\xff'))
emit('s5_no_methods_msg', msg(_s5_fail, b'\x05\xff'))
emit('s5_unknown_method', call(_s5_fail, b'\x05\x03'))
emit('s5_unknown_method_msg', msg(_s5_fail, b'\x05\x03'))
# THE SERVER MAY DEMAND CREDENTIALS THE CLIENT NEVER OFFERED, and the client
# says so with its own error rather than sending an empty auth packet.
emit('s5_userpass_unasked', call(_s5_fail, S5_METHOD_USERPASS))
emit('s5_userpass_unasked_msg', msg(_s5_fail, S5_METHOD_USERPASS))
emit('s5_auth_bad_version', call(
    _s5_fail, S5_METHOD_USERPASS + b'\x02\x00', username='u', password='p'))
emit('s5_auth_failed', call(
    _s5_fail, S5_METHOD_USERPASS + b'\x01\x01', username='u', password='p'))
emit('s5_auth_failed_msg', msg(
    _s5_fail, S5_METHOD_USERPASS + b'\x01\x01', username='u', password='p'))
# THE REPLY CODE IS CARRIED IN THE MESSAGE, formatted `{:#04x}', so `0x02'
# rather than `2' -- and an unrecognised code is "Unknown error" with the code
# still in front of it, which is what keeps a new server-side status legible.
emit('s5_status_02', call(_s5_fail, S5_METHOD_NONE + b'\x05\x02\x00'))
emit('s5_status_02_msg', msg(_s5_fail, S5_METHOD_NONE + b'\x05\x02\x00'))
emit('s5_status_05_msg', msg(_s5_fail, S5_METHOD_NONE + b'\x05\x05\x00'))
emit('s5_status_08_msg', msg(_s5_fail, S5_METHOD_NONE + b'\x05\x08\x00'))
emit('s5_status_unknown_msg', msg(_s5_fail, S5_METHOD_NONE + b'\x05\x7f\x00'))
emit('s5_reply_bad_version', call(_s5_fail, S5_METHOD_NONE + b'\x04\x00\x00'))
emit('s5_bad_atyp', call(
    _s5_fail, S5_METHOD_NONE + b'\x05\x00\x00\x09\x00\x00'))
emit('s5_bad_atyp_msg', msg(
    _s5_fail, S5_METHOD_NONE + b'\x05\x00\x00\x09\x00\x00'))
# A SHORT READ IS A GeneralProxyError, from `_readall', and it is the one
# refusal that names a closed connection rather than a protocol fault.
emit('s5_truncated', call(_s5_fail, b'\x05'))
emit('s5_truncated_msg', msg(_s5_fail, b'\x05'))
emit('s5_empty_reply', call(_s5_fail, b''))


# ================================================= SOCKS4 and SOCKS4a ========
def _s4(dest, reply=b'\x00\x5a\x1f\x90\x7f\x00\x00\x01', port=80, **proxy):
    kw = dict(proxy_type=socks.SOCKS4, addr='p', port=1080)
    kw.update(proxy)
    d = dbl(reply, **kw)
    d._negotiate_SOCKS4(dest, port)
    return d

# THE REQUEST IS VN=4, CD=1, a big-endian port, four address bytes, the user id
# and a NUL.  With no user id the field is empty and the packet is nine bytes.
_d = _s4('1.2.3.4')
emit('s4_request_no_userid', hexs(_d.sent.getvalue()))
emit('s4_request_len', '%d' % len(_d.sent.getvalue()))
emit('s4_sockname', '<%s>' % (_d.proxy_sockname,))
emit('s4_peername', '<%s>' % (_d.proxy_peername,))
_d = _s4('1.2.3.4', username='bob')
emit('s4_request_userid', hexs(_d.sent.getvalue()))
# THE USER ID IS WRITTEN RAW, with no length and no escaping, and the field is
# terminated by the NUL that follows it -- so a user id CONTAINING a NUL is
# truncated at it and the rest of the bytes are read by the server as whatever
# comes next.  Nothing in `set_proxy' or here rejects one.
_d = _s4('1.2.3.4', username='a\x00b')
emit('s4_userid_with_nul', hexs(_d.sent.getvalue()))
emit('s4_userid_nul_accepted', 'yes')

# SOCKS4a IS THE HOSTNAME EXTENSION: the address becomes the invalid 0.0.0.1,
# and the name follows the user id's NUL as a second NUL-terminated string.
_d = _s4('example.com')
emit('s4a_request', hexs(_d.sent.getvalue()))
emit('s4a_marker', hexs(_d.sent.getvalue()[4:8]))
# ...and the peername it records afterwards is that MARKER, not the hostname the
# caller asked for -- `_negotiate_SOCKS4' reports `inet_ntoa(addr_bytes)' on the
# remote-resolve path, so a caller reading `getpeername()' back sees 0.0.0.1.
emit('s4a_peername', '<%s>' % (_d.proxy_peername,))
emit('s4a_peername_is_marker',
     'yes' if _d.proxy_peername[0] == '0.0.0.1' else 'no')
# SOCKS4a takes the name through the same `encode("idna")' pass-through, so the
# case the caller wrote is the case on the wire here too.
_d = _s4('EXAMPLE.com')
emit('s4a_case_preserved', hexs(_d.sent.getvalue()[9:]))
emit('s4a_punycode', hexs(_s4('b\xfccher.example').sent.getvalue()[9:]))
emit('s4a_long_label', call(_s4, 'a' * 64 + '.com'))
emit('s4_port_big_endian', hexs(_s4('1.2.3.4', port=8080).sent.getvalue()[2:4]))
# `rdns' HAS NO EFFECT ON A NUMERIC DESTINATION -- `inet_aton' succeeds and the
# remote-resolve branch is never reached.  (A NAME with rdns=False would call
# `gethostbyname', which this probe never does.)
emit('s4_rdns_false_numeric',
     hexs(_s4('1.2.3.4', rdns=False).sent.getvalue()))

emit('s4_bad_first_byte', call(_s4, '1.2.3.4', b'\x04\x5a' + b'\x00' * 6))
emit('s4_bad_first_byte_msg', msg(_s4, '1.2.3.4', b'\x04\x5a' + b'\x00' * 6))
emit('s4_status_5b', call(_s4, '1.2.3.4', b'\x00\x5b' + b'\x00' * 6))
emit('s4_status_5b_msg', msg(_s4, '1.2.3.4', b'\x00\x5b' + b'\x00' * 6))
emit('s4_status_5c_msg', msg(_s4, '1.2.3.4', b'\x00\x5c' + b'\x00' * 6))
emit('s4_status_5d_msg', msg(_s4, '1.2.3.4', b'\x00\x5d' + b'\x00' * 6))
emit('s4_status_unknown_msg', msg(_s4, '1.2.3.4', b'\x00\x01' + b'\x00' * 6))
emit('s4_truncated', call(_s4, '1.2.3.4', b'\x00\x5a\x00'))
emit('s4_truncated_msg', msg(_s4, '1.2.3.4', b'\x00\x5a\x00'))


# ============================================================ HTTP ===========
def _http(dest='example.com', port=443,
          reply='HTTP/1.1 200 Connection established\r\n\r\n', **proxy):
    kw = dict(proxy_type=socks.HTTP, addr='p', port=8080)
    kw.update(proxy)
    d = dbl(reply.encode('latin-1'), **kw)
    d._negotiate_HTTP(dest, port)
    return d

def _wire_text(d):
    return d.sent.getvalue().decode('latin-1').replace('\r\n', '|')

_d = _http()
emit('http_request', '<%s>' % _wire_text(_d))
emit('http_request_hex', hexs(_d.sent.getvalue()))
emit('http_peername', '<%s>' % (_d.proxy_peername,))
# `proxy_sockname' IS SET TO A BYTES LITERAL HERE and to a str on both SOCKS
# paths, so the type of a field a caller reads back depends on which proxy it
# went through.  There is nothing in the API that normalises it.
emit('http_sockname', '<%s>' % (_d.proxy_sockname,))
emit('http_sockname_type', '<%s>' % type(_d.proxy_sockname[0]).__name__)
emit('http_sockname_is_bytes',
     'yes' if isinstance(_d.proxy_sockname[0], bytes) else 'no')
# THE AUTHORIZATION SCHEME IS SPELLED `basic', lower case, where RFC 7617 spells
# it `Basic'.  A proxy matching the scheme case-sensitively rejects this client.
_d = _http(username='u', password='p')
emit('http_auth_request', '<%s>' % _wire_text(_d))
emit('http_auth_scheme_lowercase',
     'yes' if b'basic ' in _d.sent.getvalue() else 'no')
# ...and the header is appended only when BOTH are set, so a username alone is
# silently dropped rather than sent as an empty-password credential.
emit('http_username_only', '<%s>' % _wire_text(_http(username='u')))
emit('http_no_proxy_connection_header',
     'yes' if b'Proxy-Connection' not in _http().sent.getvalue() else 'no')
# The CONNECT target goes through the same idna pass-through, so an ASCII name
# reaches the wire unchanged -- and a NON-ASCII one is punycoded in the request
# line while the `Host:' header, built by the same encode, is punycoded with it.
emit('http_case_preserved', '<%s>' % _wire_text(_http(dest='EXAMPLE.com')))
emit('http_punycode', '<%s>' % _wire_text(_http(dest='b\xfccher.example')))
# THE REASON PHRASE KEEPS ITS CRLF.  `status_line' is a whole line off the
# socket and `split(" ", 2)' puts everything after the code into `status_msg',
# terminator included, so the HTTPError's message ends in a bare CRLF that a log
# line then breaks across.
emit('http_reason_keeps_crlf',
     'yes' if msg(_http, reply='HTTP/1.1 407 Denied\r\n\r\n').endswith('\\r\\n>')
     else 'no')

emit('http_status_407', call(_http, reply='HTTP/1.1 407 Denied\r\n\r\n'))
emit('http_status_407_msg', msg(_http, reply='HTTP/1.1 407 Denied\r\n\r\n'))
# 400, 403 and 405 get an extra note appended, on the theory that a proxy
# answering them is not a CONNECT tunnel at all.
emit('http_status_403_msg',
     msg(_http, reply='HTTP/1.1 403 Forbidden\r\n\r\n'))
emit('http_status_403_has_note',
     'yes' if 'may not be' in msg(_http, reply='HTTP/1.1 403 F\r\n\r\n') else 'no')
emit('http_status_404_has_note',
     'yes' if 'may not be' in msg(_http, reply='HTTP/1.1 404 F\r\n\r\n') else 'no')
emit('http_status_non_numeric', call(_http, reply='HTTP/1.1 OK OK\r\n\r\n'))
emit('http_status_non_numeric_msg', msg(_http, reply='HTTP/1.1 OK OK\r\n\r\n'))
emit('http_bad_proto', call(_http, reply='ICY 200 OK\r\n\r\n'))
emit('http_bad_proto_msg', msg(_http, reply='ICY 200 OK\r\n\r\n'))
# A STATUS LINE WITH NO REASON PHRASE is a protocol-legal response that this
# client cannot parse: `split(" ", 2)' needs three fields, so `HTTP/1.1 200'
# unpacks short and raises.  RFC 9112 makes the reason phrase optional.
emit('http_two_field_status', call(_http, reply='HTTP/1.1 200\r\n\r\n'))
emit('http_two_field_status_msg', msg(_http, reply='HTTP/1.1 200\r\n\r\n'))
emit('http_empty_reply', call(_http, reply=''))
emit('http_empty_reply_msg', msg(_http, reply=''))


# ============================================ connect()'s input checks =======
# EVERY REFUSAL BELOW HAPPENS BEFORE ANY NETWORK CALL, which is what makes them
# assertable here at all -- `connect' validates its argument, then unpacks the
# proxy tuple, then connects.
def _connect(pair, **proxy):
    kw = dict(proxy_type=socks.SOCKS5, addr='p', port=1080)
    kw.update(proxy)
    d = dbl(**kw)
    d.connect(pair)

emit('connect_ipv6_bracket', call(_connect, ('[::1]', 80)))
emit('connect_ipv6_bracket_msg', msg(_connect, ('[::1]', 80)))
# A PAIR OF THE WRONG LENGTH IS REPORTED AS AN IPv6 REFUSAL.  The length test
# and the bracket test share one branch and one message, so a one-tuple -- a
# forgotten port -- comes back as "PySocks doesn't support IPv6", naming a
# feature the caller was not using.
emit('connect_short_pair', call(_connect, ('h',)))
emit('connect_short_pair_msg', msg(_connect, ('h',)))
emit('connect_three_tuple', call(_connect, ('h', 80, 0)))
# A NON-STRING HOST ESCAPES AS AN AttributeError.  `dest_pair[0].startswith'
# runs before the type check that would have raised GeneralProxyError, so the
# documented error for a bad pair is unreachable for this shape of bad pair.
emit('connect_non_string_host', call(_connect, (1234, 80)))
emit('connect_non_string_host_msg', msg(_connect, (1234, 80)))
emit('connect_empty_host', call(_connect, ('', 80)))
emit('connect_empty_host_msg', msg(_connect, ('', 80)))
emit('connect_string_port', call(_connect, ('h', '80')))
emit('connect_string_port_msg', msg(_connect, ('h', '80')))
# A BOOL PORT IS NOT ASSERTED HERE, and the omission is deliberate: `True' is an
# `int', so it passes every check above and the call goes on to CONNECT.  That is
# the one shape of bad pair this module accepts, and asserting it would mean
# opening a socket from a suite whose whole design is that it opens none.
emit('bool_port_is_untested', 'by-design')
# `connect_ex' IS DOCUMENTED TO RETURN AN ERROR INDICATOR rather than raise, and
# it does that only for errors carrying an `errno'.  The IPv6 refusal is a
# `socket.error' built from a string, so its errno is None and `connect_ex'
# re-raises -- a caller who chose `connect_ex' precisely to avoid exceptions
# gets one.
def _connect_ex(pair):
    d = dbl(proxy_type=socks.SOCKS5, addr='p', port=1080)
    return d.connect_ex(pair)

emit('connect_ex_ipv6', call(_connect_ex, ('[::1]', 80)))
emit('connect_ex_empty_host', call(_connect_ex, ('', 80)))
# `set_self_blocking' wraps both, and `functools.wraps' keeps the names, so a
# caller introspecting the API still sees `connect' rather than `wrapper'.
emit('connect_name', '<%s>' % socks.socksocket.connect.__name__)
emit('connect_ex_name', '<%s>' % socks.socksocket.connect_ex.__name__)
emit('connect_has_wrapped',
     'yes' if hasattr(socks.socksocket.connect, '__wrapped__') else 'no')


# ================================ create_connection and sockshandler =========
# `create_connection' IS THE ONLY ENTRY POINT THAT OPENS A SOCKET ITSELF, so it
# is not driven here -- what is asserted is the signature a caller writes
# against, which is the drop-in claim's other half.
import inspect
emit('create_connection_is_function',
     'yes' if inspect.isfunction(socks.create_connection) else 'no')
emit('create_connection_params', '<%s>' % ','.join(
    inspect.signature(socks.create_connection).parameters))
emit('create_connection_rdns_default', '<%s>' % (
    inspect.signature(socks.create_connection).parameters['proxy_rdns'].default,))
emit('set_proxy_rdns_default', '<%s>' % (
    inspect.signature(socks.socksocket.set_proxy).parameters['rdns'].default,))

# `sockshandler' IS THE SECOND MODULE THE .deb SHIPS and it is the one with real
# dependencies: `http.client' and `ssl' at module scope.  Its import is a case of
# its own because the recipe's RDEPENDS names `python3-netclient' and does not
# name an ssl package, so a failure here is a packaging answer and not a code one.
# EVERY KEY IN THIS BLOCK IS EMITTED ON BOTH PATHS, and that is not tidiness:
# `emitted_count' is what tells the suite a probe was truncated, so a block that
# emitted a different NUMBER of keys depending on whether an import worked would
# turn one packaging gap into a count mismatch -- and the count mismatch is the
# suite's only truncation detector.  On a failed import each case below reports
# `!<not-imported>', which is one red per case and a count that still adds up.
_sh = None
try:
    import sockshandler as _sh
    emit('sockshandler_import', 'yes')
    emit('sockshandler_import_msg', '<>')
except Exception as exc:
    emit('sockshandler_import', '!<%s>' % type(exc).__name__)
    emit('sockshandler_import_msg', '<%s>' % flat('%s' % (exc,)))


def _shv(fn):
    if _sh is None:
        return '!<not-imported>'
    try:
        return fn()
    except Exception as exc:
        return '!<%s>' % type(exc).__name__

emit('sockshandler_socks_is_socks',
     _shv(lambda: 'yes' if _sh.socks is socks else 'no'))
emit('sockshandler_classes', _shv(lambda: '<%s>' % ','.join(
    n for n in ('SocksiPyConnection', 'SocksiPyConnectionS',
                'SocksiPyHandler', 'merge_dict', 'is_ip')
    if hasattr(_sh, n))))
emit('sockshandler_no_rdns_empty',
     _shv(lambda: 'yes' if _sh.socks4_no_rdns == set() else 'no'))
emit('sockshandler_is_ip_v4', _shv(lambda: '<%s>' % _sh.is_ip('1.2.3.4')))
emit('sockshandler_is_ip_name', _shv(lambda: '<%s>' % _sh.is_ip('example.com')))
emit('sockshandler_is_ip_bare', _shv(lambda: '<%s>' % _sh.is_ip('localhost')))


# --------------------------------------------------------------- teardown ----
for _sk in _open:
    try:
        _sk.close()
    except Exception:
        pass
emit('closed_all', 'yes')

# The trailer.  `emitted_count' counts every key above INCLUDING `probe_done'
# and EXCLUDING itself, so a probe that died half way through is caught by the
# suite even though every line it did emit was correct.
emit('probe_done', 'yes')
print('emitted_count=%d' % len(emitted))