回环地址返回调用方自己的网络命名空间

import json,subprocess,time,uuid

image='python@sha256:528257d48c1da0dcecc2e725d1ae34498d60c965f1241e39cd6a85a8859bdf84'
names=['xbcatm-loop-'+uuid.uuid4().hex[:10] for _ in range(2)]
server=r'''
import json,os,selectors,socket,sys
selector=selectors.DefaultSelector()
for family,address in [(socket.AF_INET,'127.0.0.1'),(socket.AF_INET6,'::1')]:
    listener=socket.socket(family,socket.SOCK_STREAM)
    if family==socket.AF_INET6:listener.setsockopt(socket.IPPROTO_IPV6,socket.IPV6_V6ONLY,1)
    listener.bind((address,19091));listener.listen(8);selector.register(listener,selectors.EVENT_READ)
print('ready',flush=True)
while True:
    for key,events in selector.select():
        connection,peer=key.fileobj.accept()
        with connection:
            connection.settimeout(1)
            connection.sendall(json.dumps({'service':sys.argv[1],'netns':os.readlink('/proc/self/ns/net')}).encode())
'''
probe=r'''
import json,os,socket
answers={}
for address in ['127.0.0.1','::1']:
    try:
        with socket.create_connection((address,19091),timeout=1) as stream:
            pieces=[]
            while True:
                piece=stream.recv(1024)
                if not piece:break
                pieces.append(piece)
            answers[address]=json.loads(b''.join(pieces))
    except ConnectionRefusedError:answers[address]={'error':'ECONNREFUSED'}
print(json.dumps({'netns':os.readlink('/proc/self/ns/net'),'interfaces':socket.if_nameindex(),'answers':answers}))
'''
def docker(*args):
    result=subprocess.run(['docker',*args],capture_output=True,text=True,timeout=30)
    assert result.returncode==0,result.stderr
    return result.stdout
def inspect(name):return json.loads(docker('exec',name,'python3','-c',probe))
def wait_for(name,identity):
    deadline=time.monotonic()+5
    while time.monotonic()<deadline:
        result=inspect(name)
        if all(answer.get('service')==identity for answer in result['answers'].values()):return result
        time.sleep(.05)
    raise AssertionError('listener did not become ready')
try:
    for name,command in [(names[0],['python3','-u','-c',server,'A']),
                         (names[1],['python3','-c','import time;time.sleep(120)'])]:
        docker('run','-d','--rm','--name',name,'--network','none','--memory','128m','--pids-limit','64',image,*command)
    a=wait_for(names[0],'A');b_before=inspect(names[1])
    assert a['netns']!=b_before['netns']
    assert all(v=={'error':'ECONNREFUSED'} for v in b_before['answers'].values())
    docker('exec','-d',names[1],'python3','-u','-c',server,'B')
    b_after=wait_for(names[1],'B');a_after=inspect(names[0])
    assert all(v.get('service')=='A' for v in a_after['answers'].values())
    for result in [a,b_after,a_after]:
        assert all(answer['netns']==result['netns'] for answer in result['answers'].values())
    assert all(any(interface[1]=='lo' for interface in result['interfaces']) for result in [a,b_before,b_after])
    print(json.dumps({'A':a,'B_before_listener':b_before,'B_after_listener':b_after,'A_still_A':a_after},sort_keys=True))
finally:
    for name in names:subprocess.run(['docker','rm','-f',name],capture_output=True,text=True,timeout=10)