Unix 套接字提供本地通信，名称与访问规则仍须明确

import json,subprocess,uuid
image="python@sha256:528257d48c1da0dcecc2e725d1ae34498d60c965f1241e39cd6a85a8859bdf84"
program=r"""
from pathlib import Path
from tempfile import TemporaryDirectory
import errno,json,socket,socketserver,stat,struct,subprocess,sys,threading

class Handler(socketserver.StreamRequestHandler):
    def handle(self):
        self.request.settimeout(1);assert self.rfile.readline()==b'ping\n'
        pid,uid,gid=struct.unpack('3i',self.request.getsockopt(socket.SOL_SOCKET,socket.SO_PEERCRED,struct.calcsize('3i')))
        self.wfile.write(json.dumps({'uid':uid,'gid':gid,'pid_positive':pid>0}).encode()+b'\n')
client_code=r'''
import json,os,socket,sys
if sys.argv[2]=='drop':os.setgroups([]);os.setgid(65534);os.setuid(65534)
try:
    with socket.socket(socket.AF_UNIX,socket.SOCK_STREAM) as s:
        s.settimeout(1);s.connect(sys.argv[1]);s.sendall(b'ping\n');out=s.makefile('rb')
        with out:print(out.readline().decode().strip())
except PermissionError:print(json.dumps({'error':'EACCES','uid':os.geteuid()}))
'''
def client(path,drop=False):
    p=subprocess.run([sys.executable,'-c',client_code,str(path),'drop' if drop else 'keep'],capture_output=True,text=True,timeout=3)
    assert p.returncode==0,p.stderr
    return json.loads(p.stdout)
with TemporaryDirectory(prefix='unix-socket-') as directory:
    root=Path(directory);root.chmod(0o755);path=root/'service.sock'
    server=socketserver.UnixStreamServer(str(path),Handler)
    path.chmod(0o600);thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start()
    try:
        owner=client(path);denied=client(path,True)
        assert owner['uid']==0 and denied=={'error':'EACCES','uid':65534}
        path.chmod(0o666);allowed=client(path,True);assert allowed['uid']==65534 and allowed['gid']==65534
        root.chmod(0o700);directory_denied=client(path,True);assert directory_denied==denied
        root.chmod(0o755)
    finally:server.shutdown();server.server_close();thread.join(timeout=2);assert not thread.is_alive()
    assert stat.S_ISSOCK(path.stat().st_mode)
    with socket.socket(socket.AF_UNIX,socket.SOCK_STREAM) as probe:
        try:probe.connect(str(path));raise AssertionError('stale socket accepted')
        except ConnectionRefusedError:stale='ECONNREFUSED'
    with socket.socket(socket.AF_UNIX,socket.SOCK_STREAM) as duplicate:
        try:duplicate.bind(str(path));raise AssertionError('stale path reused')
        except OSError as error:assert error.errno==errno.EADDRINUSE
    path.unlink()
    with socket.socket(socket.AF_UNIX,socket.SOCK_STREAM) as replacement,socket.socket(socket.AF_UNIX,socket.SOCK_STREAM) as probe:
        replacement.bind(str(path));replacement.listen(1);replacement.settimeout(1);probe.settimeout(1);probe.connect(str(path))
        peer,_=replacement.accept()
        with peer,probe.makefile('rb') as incoming:
            peer.sendall(b'restarted');assert incoming.read(9)==b'restarted'
    print(json.dumps({'owner_peer':owner,'mode_0600_denied':denied,'mode_0666_peer':allowed,
        'directory_0700_denied':directory_denied,'closed_path_remains_socket':True,'stale_connect':stale,
        'stale_bind':'EADDRINUSE','restarted_after_owned_path_unlink':True}))
"""
name='xbcatm-socket-'+uuid.uuid4().hex[:10]
try:
    result=subprocess.run(['docker','run','--rm','-i','--name',name,'--network','none',
        '--memory','128m','--pids-limit','64',image,'python3','-'],input=program,
        capture_output=True,text=True,timeout=60)
    assert result.returncode==0,result.stderr
    print(json.dumps(json.loads(result.stdout),ensure_ascii=False,sort_keys=True))
finally:
    subprocess.run(['docker','rm','-f',name],capture_output=True,text=True,timeout=10)