Endpoints IPv4/IPv6 e identidade do fluxo

"""Original endpoint observations using only numeric IPv4/IPv6 loopback addresses.
python3 run.py --output evidence.json
No DNS traffic, wildcard binds, packet capture or global network changes.
"""
import argparse,errno,hashlib,json,pathlib,platform,socket,time
from contextlib import ExitStack
from datetime import datetime,timezone


def run():
    checks={};details={};tracked=[]
    flags=socket.AI_NUMERICHOST|socket.AI_NUMERICSERV
    def check(name,values,ok):
        checks[name]={'passed':bool(ok),**values}
        if not ok:raise AssertionError(name+': '+json.dumps(values))
    def candidates(host,port):return socket.getaddrinfo(host,str(port),socket.AF_UNSPEC,socket.SOCK_STREAM,socket.IPPROTO_TCP,flags)
    def receive(sock,count):
        data=bytearray()
        while len(data)<count:
            part=sock.recv(count-len(data))
            if not part:raise EOFError('fixture response incomplete')
            data.extend(part)
        return bytes(data)
    with ExitStack() as stack:
        def make(family,kind=socket.SOCK_STREAM):
            s=stack.enter_context(socket.socket(family,kind));s.settimeout(2);tracked.append(s);return s
        def accepted(listener):
            s,peer=listener.accept();stack.enter_context(s);tracked.append(s);s.settimeout(2);return s,peer
        v4=candidates('127.0.0.1',443);v6=candidates('::1',443)
        check('numeric-candidates-are-family-specific',{'ipv4Families':sorted(set(socket.AddressFamily(x[0]).name for x in v4)),'ipv6Families':sorted(set(socket.AddressFamily(x[0]).name for x in v6)),'allTCP':all(x[1]==socket.SOCK_STREAM and x[2]==socket.IPPROTO_TCP for x in v4+v6),'socketConnectCalls':0},all(x[0]==socket.AF_INET for x in v4) and all(x[0]==socket.AF_INET6 for x in v6) and len(v4)>0 and len(v6)>0)
        expanded='0:0:0:0:0:0:0:1';packed=socket.inet_pton(socket.AF_INET6,expanded);short=socket.inet_ntop(socket.AF_INET6,packed)
        check('equivalent-ipv6-text-has-identical-address-bytes',{'inputStringsEqual':expanded=='::1','packedAddressesEqual':packed==socket.inet_pton(socket.AF_INET6,'::1'),'normalized':short,'bytes':len(packed)},short=='::1' and packed==socket.inet_pton(socket.AF_INET6,'::1'))
        rejected={}
        for name,host,service in [('host','fixture.invalid','443'),('service','127.0.0.1','https')]:
            try:candidates(host,service)
            except socket.gaierror as exc:rejected[name]={'type':type(exc).__name__,'code':exc.errno}
            else:raise AssertionError('numeric-only input accepted a name')
        details['numericRejections']=rejected
        check('numeric-only-flags-reject-host-and-service-names',{'hostRejected':True,'serviceRejected':True,'socketConnectCalls':0,'numericHostFlag':True,'numericServiceFlag':True},len(rejected)==2)
        listen4=make(socket.AF_INET);listen4.bind(('127.0.0.1',0));listen4.listen(4);port=listen4.getsockname()[1]
        listen6=make(socket.AF_INET6);listen6.setsockopt(socket.IPPROTO_IPV6,socket.IPV6_V6ONLY,1);listen6.bind(('::1',port,0,0));listen6.listen(4)
        observed={}
        for label,family,listener,target in [('v4',socket.AF_INET,listen4,('127.0.0.1',port)),('v6',socket.AF_INET6,listen6,('::1',port,0,0))]:
            client=make(family);client.connect(target);server,peer=accepted(listener);tag=(label+'-endpoint').encode();server.sendall(tag);body=receive(client,len(tag));observed[label]={'clientLocal':client.getsockname(),'clientPeer':client.getpeername(),'acceptedPeer':peer,'serverLocal':server.getsockname(),'body':body.decode()}
        details['familyEndpoints']=observed
        check('same-port-number-can-identify-separate-family-listeners',{'samePort':listen4.getsockname()[1]==listen6.getsockname()[1],'ipv6Only':listen6.getsockopt(socket.IPPROTO_IPV6,socket.IPV6_V6ONLY),'ipv4Body':observed['v4']['body'],'ipv6Body':observed['v6']['body'],'wildcardBound':False},observed['v4']['body']=='v4-endpoint' and observed['v6']['body']=='v6-endpoint')
        flows=[]
        for n in range(2):
            client=make(socket.AF_INET);client.connect(listen4.getsockname());server,peer=accepted(listen4);flows.append({'source':client.getsockname(),'destination':client.getpeername(),'acceptedPeer':peer})
        details['simultaneousFlows']=flows
        check('source-port-distinguishes-simultaneous-flows',{'sameDestination':flows[0]['destination']==flows[1]['destination'],'differentSourcePorts':flows[0]['source'][1]!=flows[1]['source'][1],'acceptedPeersMatch':all(x['source']==x['acceptedPeer'] for x in flows)},flows[0]['destination']==flows[1]['destination'] and flows[0]['source'][1]!=flows[1]['source'][1] and all(x['source']==x['acceptedPeer'] for x in flows))
        client=make(socket.AF_INET6);client.connect(listen6.getsockname());before={'connected':True,'acceptCallsForThisConnection':0,'applicationReads':0,'applicationAuthentication':False};server,peer=accepted(listen6)
        check('connect-can-complete-before-application-accept',before,client.getpeername()[:2]==listen6.getsockname()[:2])
        client=make(socket.AF_INET6);client.bind(('::1',0,0,0));before=client.getsockname();client.connect(listen6.getsockname());after=client.getsockname();server,peer=accepted(listen6)
        details['boundSource']={'beforeConnect':before,'afterConnect':after,'acceptedPeer':peer}
        check('explicit-source-bind-is-visible-at-the-peer',{'sourcePortPreserved':before[1]==after[1],'sourceAddressPreserved':before[0]==after[0],'acceptedPeerMatches':after==peer,'scopeId':after[3]},before==after==peer)
        udp=make(socket.AF_INET,socket.SOCK_DGRAM);udp.bind(('127.0.0.1',port));sender=make(socket.AF_INET,socket.SOCK_DGRAM);sender.bind(('127.0.0.1',0));sender.sendto(b'UDP-ONLY',udp.getsockname());body,peer=udp.recvfrom(32)
        details['udp']={'source':sender.getsockname(),'destination':udp.getsockname(),'receivedFrom':peer}
        check('transport-protocol-separates-the-same-address-and-port',{'sameAddressPort':udp.getsockname()==listen4.getsockname(),'udpBody':body.decode(),'tcpListenerStillOpen':listen4.fileno()!=-1},udp.getsockname()==listen4.getsockname() and body==b'UDP-ONLY' and peer==sender.getsockname())
        reserve=make(socket.AF_INET6);reserve.setsockopt(socket.IPPROTO_IPV6,socket.IPV6_V6ONLY,1);reserve.bind(('::1',0,0,0));unused=reserve.getsockname();reserve.close()
        attempts=[];selected=None;started=time.monotonic();deadline=started+2
        for family,target in [(socket.AF_INET6,unused),(socket.AF_INET,listen4.getsockname())]:
            remaining=deadline-time.monotonic()
            if remaining<=0:raise TimeoutError('candidate budget expired')
            candidate=make(family);candidate.settimeout(remaining)
            try:candidate.connect(target)
            except OSError as exc:
                attempts.append({'family':int(family),'address':target,'outcome':type(exc).__name__,'errno':exc.errno});candidate.close()
            else:
                selected=candidate;attempts.append({'family':int(family),'address':target,'outcome':'connected'});break
        assert selected is not None
        server,peer=accepted(listen4);server.sendall(b'FALLBACK-OK');body=receive(selected,11)
        details['candidateAttempts']={'attempts':attempts,'elapsedSeconds':time.monotonic()-started,'releasedIPv6Port':unused,'portReuseRacePossible':True}
        check('failed-candidate-does-not-prove-service-unavailable',{'attemptCount':len(attempts),'firstErrnoIsConnectionRefused':attempts[0].get('errno')==errno.ECONNREFUSED,'secondConnected':attempts[-1]['outcome']=='connected','body':body.decode(),'parallelRacingImplemented':False},len(attempts)==2 and attempts[0].get('errno')==errno.ECONNREFUSED and body==b'FALLBACK-OK')
        host,service=socket.getnameinfo(listen6.getsockname(),socket.NI_NUMERICHOST|socket.NI_NUMERICSERV)
        check('numeric-endpoint-logging-keeps-host-and-port-separate',{'host':host,'serviceMatchesPort':service==str(port),'serviceIsNumeric':service.isdecimal(),'reverseLookupRequested':False},host=='::1' and service==str(port))
    return {'executedAt':datetime.now(timezone.utc).isoformat(),'pythonVersion':platform.python_version(),'platform':platform.platform(),'checks':checks,'details':details,'passed':sum(x['passed'] for x in checks.values()),'failed':sum(not x['passed'] for x in checks.values()),'allSocketsClosed':all(x.fileno()==-1 for x in tracked),'scriptSha256':hashlib.sha256(pathlib.Path(__file__).read_bytes()).hexdigest(),'scope':'Actual IPv4 and IPv6 TCP plus IPv4 UDP loopback sockets; numeric address conversion and lookup APIs. No DNS queries, wildcard listeners, NAT, proxy, link-local interface traffic, routing changes, packet capture, PMTU measurement, TLS, external traffic, load or business persistence. Sequential supplied-candidate fallback is not a Happy Eyeballs implementation. The released test port can be reused by another process; explicit refusal is checked rather than assumed.'}

if __name__=='__main__':
    p=argparse.ArgumentParser();p.add_argument('--output',required=True);args=p.parse_args();result=run();pathlib.Path(args.output).write_text(json.dumps(result,indent=2)+'\n');print(json.dumps({k:result[k] for k in ['passed','failed','allSocketsClosed']}))

Tópicos relacionados: Endereços, prefixos e âmbito · Rotas e resolução do próximo salto · Diagnóstico no contexto da aplicação

Referência: Python socket interface · DR TCP/IP 2026-09; TCP RFC 9293; IPv6 RFC 8200 with RFC 9673 update; Linux socket and iproute2 guidance