IPC

Unix 域套接字提供本地进程间通信，无须 IP 地址、TCP 端口或外部网络路由。它仍使用内核套接字设施、缓冲区、调度和应用协议。路径套接字具有文件系统名称，但 Unix 套接字也可以没有名称，Linux 还支持抽象名称。这些形式具有不同访问规则，因此“Unix 套接字就是一个受权限保护的文件”并不是完整解释。 坏掉的样子 应用找得到套接字路径，却可能因为监听已经退出而无法连接。关闭路径套接字，不会自动删除对应文件系统条目。重启时直接绑定相同路径，便可能得到 EADDRINUSE，即使那里已经没有服务接受连接。文件存在是有用诊断信息，但不能作为健康检查；要知道服务是否可用，仍需要真实通信与预期应用响应。 Linux 连接路径流套接字，需要对套接字拥有写权限，同时能够搜索沿途目录。套接字文件上的执行位，与目录遍历所需执行位并不是同一用途。宽松的套接字权限无法补偿不可访问的父目录。反过来，让敏感控制套接字对所有本地进程可写，也可能授予很强的应用能力，不能因为没有 TCP 端口就忽略其权限。 这些规则不会自动适用于所有名称形式与操作系统。Linux 抽象套接字不依赖文件路径，修改某个路径的权限无法保护抽象名称。可移植程序也不能假定每种 Unix 实现都同样执行套接字文件权限。对端凭据和应用授权，应按照实际平台与身份模型设计，不能只根据一个看起来像文件的名字推测安全属性。 快速验证 将脚本保存为 check-unix-socket.py，准备 Docker 后执行 python3 check-unix-socket.py。固定 Linux 环境使用 Python 3.11.16，服务器在一次性目录中建立路径流套接字。独立客户端进程能够降低到用户号和组号六万五千五百三十四；实验不会修改宿主账号、现有套接字或权限，所有改动随容器清理： 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) 父目录允许搜索、套接字权限为零六零零时，root 客户端能够连接，并收到服务器读取的对端凭据；降权客户端得到 EACCES。服务器通过 SO_PEERCRED 取得已经连接的进程身份，没有相信请求自行声明的用户字符串。脚本核对连接边界实际报告的有效用户号，以真实结果验证身份读取，而不只是检查选项名称出现过。 ...