Linux

Bash 的 here-string 会把一个词展开后的值送给命令的标准输入，并在末尾追加一个换行符。当变量里已经有少量文本，而目标程序从标准输入读取时，这种写法很方便。它省掉了额外的打印命令，但并不保证输入与变量逐字节相同，也不承诺内部一定使用内存、管道或文件。判断是否适用，应看接收方需要什么输入，而不是只比较命令行长度。 接收命令读取之前先完成展开 三个小于号之后的词可以包含变量展开、算术展开和命令替换，因此不一定是字面常量。单引号保留源码中的字面文本，双引号则允许通常的变量和命令展开。通过变量展开得到的字符串，即使包含美元符号和括号，也不会仅因出现在 here-string 中就被再次作为 shell 命令执行。源码中明确写出的替换语法与变量保存的数据必须区分。 当前 Bash 不会对 here-string 的词执行分词或文件名展开，而是把结果作为一个字符串交付。这里引用变量仍能清楚表达意图，但不能把这种重定向上下文和普通命令参数位置混为一谈。后者使用未引用变量时仍可能按空白拆分，或把星号变成文件名。因此，某个值通过 here-string 正常到达，并不能证明在别处省略引号也安全。 末尾追加换行是无条件发生的。原值没有换行时会多一个，原值已经有尾部换行时还会再多一个。空变量也因此产生包含一个换行字节的输入，而不是立即结束的零字节输入。对于按行处理的工具，这常常符合需求；对于严格长度、签名字节或特定结束字符的协议，则可能直接改变意义，不能把它当作原始字节透传方式。 坏掉的样子 Shell 形成输入后，read 还会按自己的规则解释。默认分词可能移除首尾 IFS 空白，反斜杠也可能作为转义符处理。读取需要保留空格和反斜杠的一行时，应使用 IFS= read -r name。它保护的是这一行的内容；单次 read 仍然在第一个换行处结束，并不会自动把任意多行字符串全部装进变量。输入构造规则与消费者解析规则是两个阶段。 带重定向的内建命令或循环可以在当前 shell 中运行，因此其中的赋值能够在后续代码中继续使用。相比把循环放进管道，这有助于避免 Bash 通常为管道阶段创建子 shell 导致的状态丢失。重定向本身不会包上一层子 shell，但如果表达式的其他地方明确使用子 shell 或命令替换，它们仍然具有通常的作用域与副作用，不能因为用了 here-string 就忽略。 ...

Case 语句把一个值依次与一组 shell 模式比较，适合命令分发、有限答案选择或文件名后缀分类。每个分支明确自己处理哪些输入，多个模式也可以共享同一个动作。分支顺序和默认行为都是接口的一部分：输入同时属于多个类别时谁先处理，完全不属于已知类别时是否报错，都需要明确规定。不能只看到成功输入能跑通，就认为整个分发器已经可靠。 模式匹配整个值 Case 使用的是 shell 模式，不是正则表达式。星号匹配任意长度字符串，问号匹配一个字符，方括号描述字符集合或范围。因此，[0-9]* 表示一个数字后面可以跟任意内容，并不表示整串都是数字。若要验证非空十进制文本，可以先拒绝空字符串，以及包含任意非数字字符的字符串，而不必把输入放进算术求值环境。 start|up|run) 这样的多个模式可以共用动作。*.log) 等后缀规则则只是在分类字符串，并不会打开文件或搜索目录。因此，匹配成功不能证明对应文件存在。与普通文件名展开不同，字符串匹配中的斜杠和开头的点也没有相同的特殊排除作用。分类、文件存在性和文件类型是不同判断，真正操作文件前仍应根据需求单独验证。 Bash 会对被测试的词执行文档规定的展开，但不会像普通命令参数那样对它分词或执行文件名展开。因此，这个位置的未引用空变量不会删掉语法上的词，进而把 case 变成语法错误。引用被测试值仍然是清楚的书写习惯。若开启 nounset，则要另外处理未提供的位置参数，或先验证参数数量；引号本身不能让未设置参数变得可用。 实际执行某个分支后，case 返回该分支最后执行命令的状态。因此，匹配成功与动作成功是两件事。下面的实验也让一个已经选中的分支返回六，并检查失败继续向外传播，避免把“找到了正确分支”误报成“分支里的操作已经成功”。 坏掉的样子 使用通常的双分号结束符时，首个匹配分支执行后，case 就结束。精确值应放在宽泛类别前，兜底星号通常放在最后，否则通用选项分支会吞掉后面的精确状态命令。语法检查只能证明结构可解析，不能证明顺序符合业务优先级。应该构造同时匹配两个规则的输入，并断言实际执行哪个动作，才能发现被遮蔽的分支。 如果没有任何模式匹配，case 不执行动作，却仍然返回零。因此，没有默认分支不等于会自动报错。命令分发器若要求拒绝未知命令，就必须在默认分支明确报告并返回失败；可选分类逻辑则可能有意允许不匹配。这个策略来自调用契约，不能假设 case 自带完整输入校验，更不能依赖自动退出选项补足业务含义。 模式上的引号也会改变含义。"*.log" 匹配字面字符，而不引用的写法匹配整个后缀类别。变量提供模式时同样需要区分：它究竟是受控的匹配表达式，还是应该精确比较的普通文本。不要让一个本来只是标识符的值意外获得通配能力，也不要为了“所有东西都加引号”而破坏本来需要的模式语义。 选项和版本还会改变行为。Nocasematch 控制是否忽略大小写；扩展模式要求在相关代码被解析之前开启 extglob。较新 Bash 还提供继续执行或继续匹配的其他分支结束符，它们不同于双分号，也不一定能在旧 shell 上运行。下面实验刻意只使用普通模式和双分号，保持 Bash 三点二兼容语法，本轮实际使用 Bash 五点二运行，避免把可选语法混入基本规则。 快速验证 下面完整代码包含精确命令、别名组、通配类别、空输入错误和未知输入错误。第二个函数验证十进制文本，不会把数字开头但后面含字母的值放过。断言同时检查正常输出与错误码，而不是仅观察整个脚本最后有没有非零退出。 ...

systemd 用户管理器代表一个账号运行服务。同一账号的多个登录会话通常共享这个管理器，而不是每个 shell 都拥有一套私有服务世界。用户单元可以放在 ~/.config/systemd/user，通过 systemctl --user 操作对应管理器。这提供的是以账号权限执行的生命周期管理，并不自动隔离所有共享系统资源。使用它之前，需要同时理解服务属于谁，以及管理器在什么条件下继续存在。 坏掉的样子 关闭一个终端，不一定停止用户服务，因为另一个登录会话可能仍然存在，管理器寿命也受到 linger 和停止延迟配置影响。即使最后一个会话结束，立即终止也不是普遍规则。必须跟随某一次图形会话结束的应用，需要明确的生命周期关系，不能假定任何用户单元天然具有这个范围。账号级后台任务与某个窗口的寿命，是不同需求。 选错管理器也是常见故障。普通 systemctl 操作系统管理器，systemctl --user 则需要连接正确用户的管理器及其运行时通信端点。通过另一身份执行命令，或者所在环境没有完成会话准备，都可能导致连接失败。仅仅设置一个环境变量，不会创建管理器，也不会授予控制其他账号的权限；诊断时应先确认执行身份与目标总线是否一致。 用户服务也不会自动继承发出控制命令的那个终端环境。所需路径、配置和图形会话变量，应通过有意选择的机制提供。使用普通账号会减少权限，但该账号仍可以按既有权限访问共享可写文件、使用网络套接字并消耗资源。因此“没有 root”并不等于“所有影响都绝对局限于自己的文件”，还要检查真实权限和资源限制。 快速验证 在具备 systemd、logind、PAM 集成、用户 D-Bus 会话与账号管理工具的一次性 Linux 系统中，以管理员身份运行下面的 Python 脚本。记录环境为 Debian 12、systemd 252.39。脚本创建临时账号、两个真实 PAM 会话和一个用户服务，不修改任何已有人员的单元或登录状态： from pathlib import Path from tempfile import TemporaryDirectory import os,pwd,subprocess,time,uuid name='xbcatm-user-'+uuid.uuid4().hex[:8];sessions=[name+'-a.service',name+'-b.service'];unit='field-worker.service' def run(*args): result=subprocess.run(args,capture_output=True,text=True,timeout=20) assert result.returncode==0,(args,result.stderr) return result.stdout.strip() def wait_for(check,seconds=5): deadline=time.monotonic()+seconds while not check(): assert time.monotonic()<deadline,'state transition timed out';time.sleep(0.05) with TemporaryDirectory(prefix=name+'-',dir='/run') as directory: root=Path(directory);root.chmod(0o755);home=root/'home';home.mkdir() run('useradd','--no-create-home','--user-group','--home-dir',str(home),'--shell','/bin/sh',name) account=pwd.getpwnam(name);uid=account.pw_uid;gid=account.pw_gid;manager=f'user@{uid}.service' def userctl(*args): return run('setpriv','--reuid='+str(uid),'--regid='+str(gid),'--init-groups', 'env',f'XDG_RUNTIME_DIR=/run/user/{uid}',f'DBUS_SESSION_BUS_ADDRESS=unix:path=/run/user/{uid}/bus', 'systemctl','--user',*args) def manager_value(key):return run('systemctl','show',manager,'-p',key,'--value') try: units=home/'.config/systemd/user';units.mkdir(parents=True) (home/'worker.py').write_text("from pathlib import Path\nimport os,time\np=Path(__file__).with_name('worker-pid');t=p.with_suffix('.tmp');t.write_text(str(os.getpid()));t.replace(p)\ntime.sleep(60)\n") (units/unit).write_text('[Service]\nType=exec\nExecStart=/usr/bin/python3 %h/worker.py\n[Install]\nWantedBy=default.target\n') for path in [home,*home.rglob('*')]:os.chown(path,uid,gid) for session in sessions: run('systemd-run','--quiet','--unit='+session,'--property=Type=simple', '--property=User='+name,'--property=PAMName=login','/usr/bin/sleep','60') wait_for(lambda:Path(f'/run/user/{uid}/bus').exists()) wait_for(lambda:len(run('loginctl','show-user',name,'-p','Sessions','--value').split())==2) assert run('loginctl','show-user',name,'-p','Linger','--value')=='no' initial=manager_value('MainPID');assert initial!='0' userctl('enable','--now',unit);wait_for(lambda:(home/'worker-pid').exists()) worker_pid=(home/'worker-pid').read_text() assert userctl('show',unit,'-p','MainPID','--value')==worker_pid run('systemctl','stop',sessions[0]) wait_for(lambda:len(run('loginctl','show-user',name,'-p','Sessions','--value').split())==1) assert manager_value('MainPID')==initial and userctl('show',unit,'-p','MainPID','--value')==worker_pid print('two PAM sessions share one user manager; closing one leaves its service unchanged') run('systemctl','stop',sessions[1]) wait_for(lambda:manager_value('ActiveState')=='inactive',seconds=20) assert not Path('/proc/'+worker_pid).exists() print('after the final session and configured stop delay, non-lingering manager and worker stop') print('controllers use setpriv and the actual user bus; they do not create extra PAM sessions') finally: for session in sessions: if run('systemctl','show',session,'-p','LoadState','--value')!='not-found':run('systemctl','stop',session) run('systemctl','stop',manager,f'user-runtime-dir@{uid}.service') run('userdel',name) 两个临时会话启动器使用相同账号，并设置 PAMName=login。脚本等待 logind 报告两个会话、且用户总线可用后，才继续控制服务。启动器使用 simple 类型，因此命令返回不会被当成 PAM 已完成的证据；会话数量和总线状态提供了明确的观察边界，避免把尚在建立的登录环境当成已经可用。 ...

trap 让 Shell 为某个信号或自身事件安排动作，适合删除临时文件、协调退出，但不会让中断变成原子操作，也无法保证任何终止方式都能完成清理。在 Bash 中，trap 是内建命令，EXIT 是 Shell 事件，而不是操作系统信号。可靠的设计需要明确处理哪些出口，并保留执行为什么结束的证据，不能仅凭注册了一次处理器就断言资源一定会被回收。 信号处理与退出清理分开 trap 'cleanup' INT TERM 会替换这些信号的默认动作。如果处理器只是返回，Shell 可能继续执行；安装处理器并不天然等于启动关停。一种明确的 Bash 写法是在信号处理器中主动退出，再让统一的退出处理器负责清理。正常执行到末尾、显式调用退出命令，也会经过这个出口，从而减少各业务分支之间重复的资源回收逻辑。 从函数或被点命令加载的文件返回，不等于结束整个 Shell。普通执行脚本里的顶层 return 也不是通用退出方式。此外，不能跨 Shell 假定未处理的致命信号一定会触发退出清理。应明确处理准备支持的终止信号，并使用服务真正采用的解释器测试，而不是在交互终端里试过一次便推广到所有启动环境。 trap - INT 恢复默认动作，trap '' INT 则忽略信号，两者策略不同。非交互 Shell 还可能从启动者继承被忽略的信号，而且无法重新捕获或重置，所以启动路径也是行为的一部分。需要确认实际安装的动作时，可以检查 trap -p 输出；配置中写过处理器，与当前 Shell 最终持有什么处理器，仍然需要区分。 保留最初的退出原因 进入退出处理器后，应先保存 $?，避免后面的命令覆盖它。清理末尾固定执行 exit 1，会把成功改成失败，也会抹掉原来不同的错误码。使用保存的状态退出，才能保留触发收尾的原始结果。清理本身失败时仍应输出可见诊断，并制定明确策略，尤其要说明主工作成功而清理失败时最终如何报告，不能把这两种结果混成一句“脚本结束”。 ...

常规 Linux 容器通过守护进程所在宿主的 Linux 内核运行进程，并利用命名空间提供选定的隔离视图。镜像文件提供用户态环境，不会给每个容器单独启动一个内核。在 macOS 的 Docker Desktop 中，守护进程和 Linux 容器位于 Linux 虚拟机里，所以容器共享的是该虚拟机内核，而不是 macOS 内核。讨论“宿主”时必须说清楚这一层。 坏掉的样子 假定每个容器都有独立操作系统内核，会导致错误诊断。如果守护进程所在 Linux 环境缺少某个内核能力，仅换成包含另一发行版用户态文件的镜像，并不会让该能力出现。反过来，升级镜像内的软件包，也不会升级宿主内核。这个区别影响系统调用、内核模块、安全边界和资源统计，不能用容器里显示的发行版名字代替判断。 命名空间隔离可以配置，不是每次部署都完全相同的固定组合。容器可以加入其他网络命名空间、使用宿主网络、共享卷，或者采用不同的用户命名空间映射。看到容器内 UID 为零，也不能直接判断容器外权限；必须结合能力集合、用户映射、安全策略和暴露资源分析。一个数字或名称不足以描述完整安全边界。 localhost 也是常见误区。在独立网络命名空间下，127.0.0.1 指向调用容器自己的回环空间，不会自动到达兄弟服务或开发者电脑。同样，在一个普通容器的可写层创建文件，不会自动出现在使用同一镜像的另一个容器里。排查访问失败时，应先分清地址和文件究竟属于哪一个视图。 快速验证 取得固定的 Python 镜像后，使用宿主 Python 和 Docker 守护进程运行脚本。实测 Docker 为 29.3.1，Docker Desktop 虚拟机内 Linux 为 6.12.76，镜像内 Python 为 arm64 的 3.11.16。两个容器使用相同镜像、不同主机名，并设置无外部网络。结束时只删除这两个临时容器： ...

文件锁协调的是一组参与者：大家约定锁哪个对象、何时获取锁，以及哪些操作必须放在临界区内。Linux 提供多种锁接口，它们的所有权和释放规则并不相同。选一个文件名，再调用名称里带 lock 的函数，并不足以构成正确协议；还必须明确底层对象、锁的种类，以及相关每份引用的寿命，否则看起来都成功加锁的程序仍可能同时修改同一份逻辑数据。 坏掉的样子 在本实验使用的 Linux 本地文件系统上，flock 提供建议式互斥。另一个请求冲突锁的进程会等待或失败，但完全没有请求锁的普通写入仍然可以进行。因此，文件锁不会赋予内容隐私，也不会自动限制所有能够打开文件的程序。只有参与者实际遵循共同协议，互斥才能保护需要保持一致的数据操作。 如果更新通过新 inode 替换原路径，直接锁数据文件就会出现对象分离。已有锁仍关联旧的打开文件描述，后续打开者却访问替换后的文件，并可能成功获取另一把锁。两个写者都以为自己独占同一份逻辑文档，实际锁住的却是不同对象。路径名称稳定，并不意味着路径背后的文件身份稳定。 传统 POSIX 记录锁还具有容易忽略的寿命规则：锁属于进程，该进程关闭任何一个指向该文件的描述符，都会释放它在该文件上持有的记录锁。因此，某个库独立打开并关闭同一 inode，也可能使程序其他位置持有的锁失效。这与通过打开文件描述拥有锁的 flock 不同，不能把两套释放规则混用。还要注意，通过任意复制引用显式解锁，也会释放同一把 flock；其他复制描述符仍然打开，并不能抵消显式解锁。 快速验证 把脚本保存为 locking.py，在一次性 Linux 环境中运行 python3 locking.py。实测环境为 Linux 6.12.76、Python 3.11.2 和本地 tmpfs。每个竞争锁的请求都在独立子进程中执行，并使用非阻塞获取，因此预期冲突会产生可检查结果，不会让演示一直挂起： from pathlib import Path from tempfile import TemporaryDirectory import fcntl,os,subprocess,sys probe_code=r''' import fcntl,sys with open(sys.argv[1],'r+') as handle: try: if sys.argv[2]=='flock':fcntl.flock(handle,fcntl.LOCK_EX|fcntl.LOCK_NB) else:fcntl.lockf(handle,fcntl.LOCK_EX|fcntl.LOCK_NB,1,int(sys.argv[3]),0) except BlockingIOError:print('blocked') else:print('acquired') ''' def probe(path,kind='flock',start=0): result=subprocess.run([sys.executable,'-c',probe_code,str(path),kind,str(start)],capture_output=True,text=True,timeout=5) assert result.returncode==0,result.stderr return result.stdout.strip() with TemporaryDirectory(prefix='xbcatm-locks-') as directory: root=Path(directory);target=root/'data';target.write_text('original') handle=target.open('r+');fcntl.flock(handle,fcntl.LOCK_EX) assert probe(target)=='blocked' subprocess.run([sys.executable,'-c',"from pathlib import Path;import sys;Path(sys.argv[1]).write_text('uncooperative')",str(target)],capture_output=True,text=True,check=True) handle.seek(0);assert handle.read()=='uncooperative' duplicate=os.dup(handle.fileno());handle.close() try:assert probe(target)=='blocked' finally:os.close(duplicate) assert probe(target)=='acquired' print('flock blocks cooperating contenders but not ordinary writes; dup retains the same lock') with target.open('r+') as handle: fcntl.flock(handle,fcntl.LOCK_EX) replacement=root/'replacement';replacement.write_text('new inode');replacement.replace(target) assert os.fstat(handle.fileno()).st_ino!=target.stat().st_ino assert probe(target)=='acquired' lockfile=root/'stable.lock';lockfile.touch() with lockfile.open('r+') as handle: fcntl.flock(handle,fcntl.LOCK_EX) replacement.write_text('next inode');replacement.replace(target) assert probe(lockfile)=='blocked' assert probe(lockfile)=='acquired' print('replacing locked data changes the inode; a separate stable lock file protects that replacement protocol') first=os.open(target,os.O_RDWR);second=os.open(target,os.O_RDWR) try: fcntl.lockf(first,fcntl.LOCK_EX|fcntl.LOCK_NB,1,0,0) assert probe(target,'record',0)=='blocked' and probe(target,'record',1)=='acquired' os.close(second);second=None assert probe(target,'record',0)=='acquired' print('POSIX record locks permit disjoint bytes; closing another descriptor to the inode releases process-owned locks') finally: if second is not None:os.close(second) os.close(first) 首先，父进程持有排他 flock，遵守协议的子进程报告 blocked。另一个子进程完全不请求锁就写入文件，父进程随后读到修改后的内容。同一个文件、同一组权限下的两个结果，分别证明互斥确实存在，以及它的建议式边界。不能只展示一个竞争者失败，就推断所有普通读写也会被锁阻止。 ...

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 取得已经连接的进程身份，没有相信请求自行声明的用户字符串。脚本核对连接边界实际报告的有效用户号，以真实结果验证身份读取，而不只是检查选项名称出现过。 ...

普通 write(2) 成功，可以让实际接受的字节通过内核文件状态被访问，但本身不能建立崩溃持久性。fsync 请求同步文件数据及相关元数据，并等待存储栈报告完成。它是更新协议中的一个操作，仍需要正确顺序、错误检查、目录项处理和底层真实保证。不能只在代码里出现一次 fsync，就推断整个更新已经具有完整事务与恢复语义。 坏掉的样子 用户态缓冲还形成更早一层边界。语言缓冲区尚未把字节提交给内核时，对描述符调用 fsync 无法同步这些字节。应先 flush 缓冲，并检查两个阶段的错误。普通 write 也可能只完成部分写入，延迟存储错误则可能在同步时才出现，而不是在最初写入时报告，因此每层返回结果都需要按契约处理。 同步新建文件，不一定同步命名它的目录项。在 Linux 上，持久创建或替换协议还需要适当的目录同步。原子 rename 提供运行期间的命名空间可见性保证，却不会独立证明目录变化能够经历崩溃保留。当前打开新路径能读到数据，与重启后仍能通过该名字找到预期对象，是不同问题。 fdatasync 不意味着完全不处理元数据。它同步数据，以及后续正确读取数据所需的元数据，包括变化后的文件长度；对于读取不必需的其他元数据更新，则可以省略。把它理解为与 fsync 在所有方面相同，或者完全无视元数据，都会掩盖选择两者的真正依据，应根据应用需要保持的状态理解同步范围。 快速验证 把脚本保存为 syncfile.py，在装有 Python 和 strace 的一次性 Linux 系统中运行 python3 syncfile.py。临时目录必须支持普通文件和目录同步。实测使用 Linux 6.12.76、Python 3.11.2、strace 6.1 和 ext4 卷。样本检查真实调用与运行结果，不会切断存储电源： from pathlib import Path from tempfile import TemporaryDirectory import subprocess,sys program=r''' from pathlib import Path import errno,os,sys root=Path(sys.argv[1]);current=root/'current';candidate=root/'candidate' current.write_bytes(b'old');old=current.open('rb') try: with candidate.open('wb') as handle: handle.write(b'new');handle.flush();os.fsync(handle.fileno()) os.replace(candidate,current) directory=os.open(root,os.O_RDONLY|os.O_DIRECTORY) try:os.fsync(directory) finally:os.close(directory) assert current.read_bytes()==b'new' and old.read()==b'old' with current.open('ab') as handle: handle.write(b'-extended');handle.flush();os.fdatasync(handle.fileno()) assert current.read_bytes()==b'new-extended' and current.stat().st_size==12 read_end,write_end=os.pipe() try: try:os.fsync(write_end) except OSError as error:assert error.errno==errno.EINVAL else:raise AssertionError('fsync on a pipe unexpectedly succeeded') finally:os.close(read_end);os.close(write_end) finally:old.close() print('replacement content, old open reference, appended length, and unsupported-pipe error verified') ''' with TemporaryDirectory(prefix='xbcatm-fsync-') as directory: root=Path(directory);trace=root/'trace' result=subprocess.run(['strace','-yy','-s','200','-e','trace=%file,write,fsync,fdatasync','-o',str(trace),'--',sys.executable,'-c',program,str(root)],capture_output=True,text=True,timeout=10) assert result.returncode==0,result.stderr lines=trace.read_text().splitlines() file_sync=next(i for i,line in enumerate(lines) if 'fsync(' in line and str(root/'candidate') in line and '= 0' in line) rename=next(i for i,line in enumerate(lines) if line.startswith('rename') and str(root/'current') in line and '= 0' in line) directory_sync=next(i for i,line in enumerate(lines) if 'fsync(' in line and '<'+str(root)+'>' in line and '= 0' in line) assert file_sync<rename<directory_sync assert any('fdatasync(' in line and str(root/'current') in line and '= 0' in line for line in lines) print(result.stdout,end='') for line in lines: if line.startswith(('fsync(','fdatasync(','rename')):print(line) print('syscall order verified: flush data, fsync candidate, replace name, fsync parent directory') 子进程创建内容为 old 的当前文件，并保留一份打开读取引用。然后把 new 写入独立候选文件，flush Python 缓冲，再对候选描述符调用 fsync。候选文件位于同一个目录，也使此实验的后续 rename 保持在同一文件系统内，不依赖跨挂载复制等不同操作来模拟替换。 ...