三个相连方块,依次标有 Observe、Decide 和 Act。

systemd 管理进程的启动、失败与停止

服务管理器把进程启动、失败和停止变成可以观察的状态变化。使用 systemd 时,单元可以明确前台命令、重启策略与输出目的地,不必再靠一个只会反复拉起程序、却说不清为什么停止的 shell 循环。但这并不能证明运行中的应用已经能响应请求,也不能证明重新启动一定修好了原来的故障。管理生命周期与验证业务可用性,需要各自成立。 坏掉的样子 服务类型必须匹配程序的启动契约。Type=simple 很早就认为启动成功,此时可执行文件甚至还不一定已经成功执行。Type=exec 则等待执行程序成功,因此缺少可执行文件时,启动任务可以直接失败。不过,这两种类型都不会自动等待应用连接数据库、加载状态或完成业务初始化。看到启动命令返回成功,还不能据此开始向它发送必须成功的业务请求。 如果程序自行转入后台,就可能与预期前台主进程的单元不一致。应使用程序的前台模式,让管理器掌握生命周期,或者按照程序真实的守护化协议选择有文档依据的类型。不要靠进程名称搜索来猜主进程编号:名字可能重复,包装脚本也可能把错误的进程留作单元主进程,导致观察、停止和重启的对象都不符合预期。 重启策略也要有清楚边界。Restart=on-failure 覆盖的是规定的失败退出,并不是所有终止信号,更不是所有不健康状态。一个卡住却没有退出的进程,不会自动产生重启事件。应该结合有意义的应用健康观察与有限重试,而不是把单元状态为活动,直接等同于服务已经恢复对外工作。 快速验证 在 PID 1 为 systemd 的一次性 Linux 环境中,以管理员身份运行下面的 Python 脚本。它只写入明确命名的临时运行时单元和一个临时工作目录。记录中的环境是隔离容器内的 Debian 12、systemd 252.39。执行前确认两个单元文件名尚未被使用,避免与已有服务混淆: from pathlib import Path from tempfile import TemporaryDirectory import subprocess,time unit='xbcatm-supervise-lab.service';bad='xbcatm-exec-failure-lab.service' unit_path=Path('/run/systemd/system')/unit;bad_path=unit_path.with_name(bad) assert not unit_path.exists() and not bad_path.exists() def run(args,success=True): result=subprocess.run(args,text=True,capture_output=True,timeout=10) assert (result.returncode==0)==success,(args,result.returncode,result.stderr) return result.stdout.strip() def value(name,key):return run(['systemctl','show',name,'-p',key,'--value']) def wait_for(check): deadline=time.monotonic()+5 while not check(): assert time.monotonic()<deadline,'state transition timed out' time.sleep(0.02) with TemporaryDirectory(prefix='xbcatm-supervise-',dir='/run') as directory: root=Path(directory);worker=root/'worker.py';ready=root/'ready' worker.write_text("""from pathlib import Path import os,signal,sys signal.signal(signal.SIGTERM,lambda *_: sys.exit(0)) Path(__file__).with_name('ready').write_text(str(os.getpid())) print('worker ready',flush=True) while True: signal.pause() """) unit_path.write_text(f'''[Unit] Description=Disposable supervision exercise StartLimitIntervalSec=10s StartLimitBurst=3 [Service] Type=exec ExecStart=/usr/bin/python3 -u {worker} Restart=on-failure RestartSec=200ms TimeoutStopSec=2s StandardOutput=journal StandardError=journal ''') bad_path.write_text(f'[Service]\nType=exec\nExecStart={root}/missing-program\n') try: run(['systemd-analyze','verify',str(unit_path)]) run(['systemctl','daemon-reload']);run(['systemctl','start',unit]) wait_for(lambda: ready.exists() and ready.read_text()==value(unit,'MainPID')) first=value(unit,'MainPID');assert first!='0' first_invocation=value(unit,'InvocationID') run(['systemctl','kill','--kill-who=main','--signal=KILL',unit]) wait_for(lambda: value(unit,'MainPID') not in ('0',first) and ready.read_text()==value(unit,'MainPID')) assert int(value(unit,'NRestarts'))==1 print('SIGKILL: replacement PID confirmed; NRestarts=1') run(['journalctl','--sync']) for invocation in (first_invocation,value(unit,'InvocationID')): logs=run(['journalctl','_SYSTEMD_INVOCATION_ID='+invocation,'--no-pager','-o','cat']) assert 'worker ready' in logs print('journal contains both application startups') run(['systemctl','stop',unit]);time.sleep(0.3) assert value(unit,'ActiveState')=='inactive' and value(unit,'MainPID')=='0' print('manager stop leaves service inactive') run(['systemctl','daemon-reload']);run(['systemctl','start',bad],success=False) assert value(bad,'Result')=='exit-code' and value(bad,'ExecMainStatus')=='203' print('Type=exec reports missing executable: status 203/EXEC') finally: run(['systemctl','stop',unit,bad]) for name in (unit,bad): if value(name,'ActiveState')=='failed': run(['systemctl','reset-failed',name]) unit_path.unlink();bad_path.unlink(missing_ok=True) run(['systemctl','daemon-reload']) 工作进程始终留在前台,并写出自己的启动标记。脚本等待该标记与管理器报告的主进程编号一致,再只向受管理的主进程发送 SIGKILL。随后确认编号发生变化,且自动重启次数为一,从而证明管理器确实替换了被终止的那个进程,并非只是原进程仍然存活,或者恰巧找到了同名程序。 ...

十一月 1, 2026 · 2 分钟 · XB Field Notes
一个维护任务连接一枚时钟、运行时间界与日志结果。

维护定时器应让其最近结果可见

调度只是工作的一半,运维者还需要看到任务上次何时运行、如何结束。

八月 4, 2026 · 1 分钟 · XB Field Notes
一个服务处在一个有界且带速率限制的重启循环里。

重启策略需要一个速率限制

使用显式的服务类型、有界的重启行为,以及在日志里留下证据的启动速率限制。

八月 3, 2026 · 1 分钟 · XB Field Notes
浅色网格上显示文章标题,旁侧是依次标注 Observe、Decide、Act 的三个相连方框。

用实际拒绝结果验证 systemd 服务边界

服务加固应该把一种明确不需要的操作变为被拒绝的操作,同时保留应用必须完成的工作。systemd 为文件系统视图、命名空间、权限和其他资源提供执行设置,但这些是边界各不相同的独立控制。添加四条熟悉的指令,并不意味着所有可读文件、可接触进程、继承的文件描述符或允许的系统调用都会消失。判断依据应是配置的真实语义与进程的实际行为。 坏掉的样子 ProtectSystem=strict 让服务挂载命名空间内的大部分文件系统层级变为只读。只读不代表不可读取,而且该设置有文档明确的例外,例如还需要单独控制的接口文件系统。服务仍然可能泄露它能够读取的信息,也仍然可以破坏可写目录中的数据。因此应该先使用专用服务身份,列清必要访问,再选择对应限制,而不是把一个只读开关当成覆盖所有资源的隔离墙。 PrivateTmp=yes 给服务提供私有临时目录。如果应用原本依赖另一个服务通过普通 /tmp 交换文件,这项设置可能直接打断协作。PrivateNetwork=yes 创建拥有独立回环接口的网络命名空间,因此宿主回环上的监听器不再是里面的同一个端点。需要直接连接数据库或公共接口的程序,若没有有意安排所需连接,就不适合直接套用这一设置。应先查真实调用路径,不能假定网络隔离不会影响正常功能。 NoNewPrivileges=yes 阻止执行程序时通过设置用户身份的二进制文件、文件能力等机制取得新的特权,并不删除进程已经拥有的权限。同样,单独写一条 IPAddressAllow,不意味着其他地址全部被拒绝:地址过滤有允许与拒绝的优先关系,还依赖内核功能支持。选项名称不是完整安全策略,每项控制都需要按其契约理解,尤其不能把未验证的限制写成已经成立的保证。 快速验证 在一次性的 Linux 系统中以管理员身份运行下面的 Python 脚本,要求 PID 1 为 systemd,并具备 Python 和标准用户管理工具。记录中的环境是 Debian 12、systemd 252.39。脚本创建唯一命名的系统账号、运行时服务、状态目录与无害样本文件,并在结束时删除。前后两阶段使用同一个非 root 身份和相同工作负载: from pathlib import Path from tempfile import TemporaryDirectory import json,shutil,socket,subprocess,uuid suffix=uuid.uuid4().hex[:8];name='xbcatm-hardening-'+suffix unit=name+'.service';unit_path=Path('/run/systemd/system')/unit state=Path('/var/lib')/name;marker=Path('/tmp')/(name+'-marker');scratch=Path('/tmp')/(name+'-scratch') assert not unit_path.exists() and not state.exists() and not marker.exists() def ctl(*args): result=subprocess.run(['systemctl',*args],capture_output=True,text=True,timeout=10) assert result.returncode==0,(args,result.stderr) return result.stdout.strip() with TemporaryDirectory(prefix=name+'-',dir='/run') as directory, socket.socket() as listener: root=Path(directory);root.chmod(0o755);outside=root/'outside';outside.write_text('initial\n');outside.chmod(0o666) marker.write_text('host namespace marker\n');listener.bind(('127.0.0.1',0));listener.listen(2) worker=root/'worker.py' worker.write_text(r"""from pathlib import Path import json,socket,sys root=Path(__file__).parent;state=Path(sys.argv[1]);marker=Path(sys.argv[2]);scratch=Path(sys.argv[3]);port=int(sys.argv[4]);phase=sys.argv[5] result={} try: with (root/'outside').open('a') as output:output.write(phase+'\n') result['outside_write']=True except OSError as error:result['outside_write']=False;result['outside_errno']=error.errno result['host_tmp_visible']=marker.exists() scratch.write_text(phase);result['tmp_write']=scratch.read_text()==phase try: with socket.create_connection(('127.0.0.1',port),timeout=0.5):pass result['network_connect']=True except OSError as error:result['network_connect']=False;result['network_errno']=error.errno status=dict(line.split(':',1) for line in Path('/proc/self/status').read_text().splitlines() if ':' in line) result['no_new_privs']=status['NoNewPrivs'].strip() (state/'result.json').write_text(json.dumps(result)) """) definition=f'''[Service] Type=oneshot User={name} Group={name} StateDirectory={name} StateDirectoryMode=0700 ExecStart=/usr/bin/python3 {worker} {state} {marker} {scratch} {listener.getsockname()[1]} PHASE ''' subprocess.run(['useradd','--system','--no-create-home','--user-group','--shell','/usr/sbin/nologin',name],check=True) try: unit_path.write_text(definition.replace('PHASE','baseline')) ctl('daemon-reload');ctl('start',unit) baseline=json.loads((state/'result.json').read_text()) assert baseline=={'outside_write':True,'host_tmp_visible':True,'tmp_write':True,'network_connect':True,'no_new_privs':'0'},baseline connection,_=listener.accept();connection.close() original=outside.read_text() settings='ProtectSystem=strict\nPrivateTmp=yes\nPrivateNetwork=yes\nNoNewPrivileges=yes\n' unit_path.write_text(definition.replace('PHASE','hardened')+settings) subprocess.run(['systemd-analyze','verify',str(unit_path)],check=True) ctl('daemon-reload');ctl('start',unit) hardened=json.loads((state/'result.json').read_text()) assert hardened['outside_write'] is False and hardened['outside_errno']==30,hardened assert hardened['host_tmp_visible'] is False and hardened['tmp_write'] is True assert hardened['network_connect'] is False and hardened['network_errno']==111,hardened assert hardened['no_new_privs']=='1' assert outside.read_text()==original and scratch.read_text()=='baseline' print('baseline: outside write, host tmp visibility, loopback connection permitted; NoNewPrivs=0') print('hardened: outside write=EROFS; host tmp hidden; host listener refused; NoNewPrivs=1') print('managed state directory and private temporary writes still succeed') finally: ctl('stop',unit) if ctl('show',unit,'-p','ActiveState','--value')=='failed':ctl('reset-failed',unit) unit_path.unlink(missing_ok=True);ctl('daemon-reload') marker.unlink(missing_ok=True);scratch.unlink(missing_ok=True) if state.exists():shutil.rmtree(state) subprocess.run(['userdel',name],check=True) 基线阶段故意让该账号能够写入一个样本文件,同时在管理器看到的临时目录里创建标记,并在其回环接口启动监听器。尚未加限制时,工作进程能够追加文件、看到标记、连接监听器,且读取到零值的 NoNewPrivs 标志。这一步排除了普通文件权限本来就禁止写入、或者监听器根本不存在等混淆因素,之后的拒绝才有明确对照。 ...

六月 27, 2026 · 2 分钟 · XB Field Notes