← TCP/IP: fundamentals and diagnosis
11 / 12 · 60 MIN

IPv4/IPv6 endpoints and flow identity

Observe addresses, ports and families on real sockets and build a diagnostic worksheet that distinguishes each connection.

Execute and record scope

Save the complete code below as run.py and run python3 run.py --output evidence.json. You need a system with IPv6 loopback and permission to create local sockets. Recorded execution used Python 3.13.1 on macOS; consulted documentation from the 3.13 line displayed 3.13.16. The script binds only to 127.0.0.1 and::1, without wildcard listeners or global network changes. Ports are temporary. If IPv6 is unavailable, record the limitation instead of presenting the group as passed or silently disabling checks.

Preparing an address is not connecting

The first group uses getaddrinfo with numeric addresses and TCP transport. Its result includes family, type and sockaddr for a later attempt. It does not mean port 443 is open, HTTPS exists or DNS answered. The rejection group uses AI_NUMERICHOST and AI_NUMERICSERV: a hostname and a service name are rejected before any connect. In a fictional adapter, classifying either as a firewall fault would send the ticket to the wrong team. Record phase and flags together with the error.

Normalize value and retain context

The::1 form and expanded 0:0:0:0:0:0:0:1 form produce the same 16 bytes. The experiment compares those bytes and renders::1 again. This observation avoids false incidents caused by literal string comparison. For link-local addresses, also retain zone information when needed: fe80::42%7 and fe80::42%9 should not be merged merely because their address bits match. This second example is document-based analysis using RFC 4007; no link-local traffic was executed. Zone numbering is local to the node and should not be copied across hosts as a universal identifier.

Compare listeners sharing a port number

The lab creates an IPv4 TCP listener on 127.0.0.1 and an IPv6 TCP listener on::1, both with the same numeric port. For IPv6 it sets and reads IPV6_V6ONLY=1. The v4-endpoint and v6-endpoint responses identify separate teaching endpoints inside the script. This is not one dual-stack listener. Python documentation describes IPv4-mapped addresses in another configuration, but that configuration was not executed here. In a fictional rollout, different versions by family may result from distinct listeners. Identify socket, process and configuration before blaming caches or changing clients.

Reverse perspective without losing the flow

For each connection, compare client getsockname and getpeername with the server’s accepted socket address. Without NAT or proxy in this experiment, the client’s local source matches the accepted peer. Two simultaneous connections to the same listener retain different source ports. The explicit-bind group requests port zero and records the assigned port before and after connecting. Zero is a selection policy, not the effective port to find in logs. With production intermediaries, record each hop; this direct correspondence should not be assumed across NAT or proxies.

Add transport to the worksheet

Another group binds a UDP socket to the same IPv4 address and port number as the TCP listener. The UDP-ONLY datagram reaches the UDP socket while the TCP listener remains open. Use this observation to complete the worksheet: protocol, family, source, source port, destination, destination port, context and time. Port alone is insufficient for attributing traffic to an application. The experiment neither converts protocols nor measures datagram loss. During operational collection, also distinguish the configured address from the selected candidate and observed peer so the next team can reproduce the attempt.

"""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']}))
IN PRACTICE

Exercise: compare client local=127.0.0.1:51000, peer=127.0.0.1:42000 with the accepted socket. Then add a second connection and identify the distinguishing field.

Common pitfalls

Keeping only the port, comparing IPv6 as strings, deleting a scoped address’s zone or copying an ephemeral port as universal configuration.

Related topics: Addresses, prefixes, and scope · Routes and next-hop resolution · Diagnosis in the application context

Take this idea with you

Evidence should identify the observed flow: transport, family, both endpoints and context. Configuration and observation are different fields.

Create account

Reference: Python socket interface · BigSavant TCP/IP 2026-09; TCP RFC 9293; IPv6 RFC 8200 with RFC 9673 update; Linux socket and iproute2 guidance