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

import json,re,subprocess,sys,tempfile,time
from pathlib import Path

image='python@sha256:528257d48c1da0dcecc2e725d1ae34498d60c965f1241e39cd6a85a8859bdf84'
server=r'''
import json,os,socket,sys,time
deadline=time.monotonic()+120
with socket.socket(socket.AF_INET,socket.SOCK_STREAM) as listener:
    listener.bind(('127.0.0.1',19091));listener.listen(8);listener.settimeout(.25)
    while time.monotonic()<deadline:
        try:connection,peer=listener.accept()
        except TimeoutError:continue
        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,time
try:
    deadline=time.monotonic()+2
    with socket.create_connection(('127.0.0.1',19091),timeout=1) as stream:
        payload=b''
        while True:
            remaining=deadline-time.monotonic()
            if remaining<=0:raise TimeoutError('response deadline')
            stream.settimeout(min(1,remaining))
            piece=stream.recv(1024)
            if not piece:break
            payload+=piece;assert len(payload)<=4096,'oversize response'
        answer=json.loads(payload)
except ConnectionRefusedError:answer={'error':'ECONNREFUSED'}
print(json.dumps({'netns':os.readlink('/proc/self/ns/net'),'interfaces':socket.if_nameindex(),
    'answers':{'127.0.0.1':answer}}))
'''
def docker(*args,timeout=30):
    result=subprocess.run(['docker',*args],capture_output=True,text=True,timeout=timeout)
    assert result.returncode==0,result.stderr
    return result.stdout
def owned_id(path):
    value=path.read_text().strip()
    assert re.fullmatch('[0-9a-f]{64}',value),'missing or invalid owned container ID'
    return value
def remove_owned(paths):
    errors=[]
    for path in paths:
        try:
            cid=owned_id(path)
            print('Cleanup target: '+cid,file=sys.stderr)
            docker('rm','-f',cid,timeout=10)
        except Exception as error:
            errors.append(f'{path.name}: {error}')
    if errors:raise RuntimeError('Cleanup not confirmed; inspect the recorded IDs: '+'; '.join(errors))
def inspect(cid,timeout=5):
    result=json.loads(docker('exec',cid,'python3','-c',probe,timeout=timeout))
    assert set(result['answers'])=={'127.0.0.1'},'missing or extra probe address'
    assert re.fullmatch(r'net:\[\d+\]',result['netns']),'invalid namespace ID'
    assert any(interface[1]=='lo' for interface in result['interfaces']),'loopback missing'
    return result
def expected(result,identity):
    return {'127.0.0.1':{'service':identity,'netns':result['netns']}}
def wait_for(cid,identity):
    deadline=time.monotonic()+5
    while True:
        remaining=deadline-time.monotonic()
        if remaining<=0:raise AssertionError('listener did not become ready')
        result=inspect(cid,timeout=remaining)
        if time.monotonic()>deadline:raise AssertionError('readiness deadline exceeded')
        if result['answers']==expected(result,identity):return result
        assert result['answers']=={'127.0.0.1':{'error':'ECONNREFUSED'}},'unexpected service response'
        time.sleep(min(.05,max(0,deadline-time.monotonic())))
with tempfile.TemporaryDirectory(prefix='xbcatm-loop-') as work:
    paths=[];ids=[]
    try:
        for identity,command in [('A',['python3','-u','-c',server,'A']),
                                 ('B',['python3','-c','import time;time.sleep(120)'])]:
            path=Path(work)/(identity+'.cid');paths.append(path)
            docker('create','--cidfile',str(path),'--pull','never','--network','none',
                '--memory','128m','--pids-limit','64',image,*command)
            cid=owned_id(path);ids.append(cid)
            print('Owned container: '+cid,file=sys.stderr)
            docker('start',cid)
        a=wait_for(ids[0],'A');b_before=inspect(ids[1])
        assert a['netns']!=b_before['netns']
        assert b_before['answers']=={'127.0.0.1':{'error':'ECONNREFUSED'}}
        docker('exec','-d',ids[1],'python3','-u','-c',server,'B')
        b_after=wait_for(ids[1],'B');a_after=inspect(ids[0])
        assert b_after['netns']==b_before['netns']
        assert a_after['netns']==a['netns']
        assert a_after['answers']==expected(a_after,'A')
        report={'A':a,'B_before_listener':b_before,'B_after_listener':b_after,'A_still_A':a_after}
    finally:
        remove_owned(paths)
print(json.dumps(report,sort_keys=True))