Operations

在选择重定向或错误状态前，先检查方法、Location 头与缓存行为。

两个小的只读 helper 把公共端点变成发布前和发布后都能检查的证据。

复盘是一张从检测到症状到验证修复的顺序图，由产生每个事实的命令支撑。

服务加固应该把一种明确不需要的操作变为被拒绝的操作，同时保留应用必须完成的工作。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 标志。这一步排除了普通文件权限本来就禁止写入、或者监听器根本不存在等混淆因素，之后的拒绝才有明确对照。 ...

提前准备 TTL，观察权威与递归应答，并让两个端点在整个缓存窗口内都有效。

DNS TTL 表示一条资源记录通常可以在缓存中保留多久，之后需要刷新。修改权威记录会改变未来回答，却不会向所有曾经存储该数据的解析器发送失效通知。因此，一次迁移既包含权威更新，也包含一段客户端可能合法发现不同地址的时期。规划服务可用性时，需要同时考虑新答案和仍然存在的旧答案，而不是只看控制台保存成功。 坏掉的样子 在修改地址的同一时刻降低 TTL，并不会缩短此前缓存回答的寿命。保存了旧长 TTL 的解析器尚未获知新短 TTL。此时立即移除旧服务，可能让那些客户端失败，即使所有权威查询都已返回新地址。问题来自服务重叠期与尚未消失的发现状态不匹配；反复检查权威端的新值正确，并不能解决使用旧缓存的客户端请求。 回滚也具有同样的不对称性。把区域改回去，不会从已经学到新地址的缓存里撤回该地址。短暂正常的新目的地，可能在回滚命令完成后继续收到请求。应让回滚目标与刚发布过的目标在相关重叠期内都可用，或者明确客户端访问任何一个时如何获得可接受结果。DNS 回滚不是立即停止新地址流量的开关。 也不要把每层缓存的原始 TTL 简单相加，作为传播时间。普通转发缓存返回剩余寿命，下游再对收到的值继续计时；不同解析器群体则可能在不同时间填充缓存。应用自有缓存、长连接、权威同步延迟以及配置的陈旧回答策略，还会让单个 TTL 无法成为端到端完成保证。应把这些行为分别调查，不能用一个“传播中”笼统覆盖。 快速验证 将脚本保存为 check-dns-move.py，准备 Docker 后执行 python3 check-dns-move.py。固定环境以 BIND 9.18.49 为权威，以 dnsmasq 2.90 为一个转发缓存，二者运行在隔离容器回环接口上。程序只修改自己的临时区域，每次递增序列号并确认权威重载完成，再观察缓存。整个过程不改变公共 DNS，也不调整系统时间： from pathlib import Path from tempfile import TemporaryDirectory import json,subprocess,uuid recipe='FROM python@sha256:528257d48c1da0dcecc2e725d1ae34498d60c965f1241e39cd6a85a8859bdf84\nRUN apt-get update && apt-get install -y --no-install-recommends bind9=1:9.18.49-1~deb12u2 bind9-dnsutils=1:9.18.49-1~deb12u2 dnsmasq-base=2.90-4~deb12u2 python3-dnspython=2.3.0-1 && rm -rf /var/lib/apt/lists/*\nCMD ["/usr/bin/python3"]\n' program=r""" from contextlib import ExitStack,contextmanager from pathlib import Path from tempfile import TemporaryDirectory import json,re,signal,subprocess,time import dns.exception,dns.message,dns.query,dns.rcode,dns.rdatatype def run(*args,check=True): result=subprocess.run(args,capture_output=True,text=True,timeout=15) if check:assert result.returncode==0,(args,result.returncode,result.stdout,result.stderr) return result def query(name,kind='A',port=1053): return dns.query.udp(dns.message.make_query(name,kind),'127.0.0.1',port=port,timeout=1) def values(response,kind='A'): wanted=dns.rdatatype.from_text(kind) return [item.to_text() for rrset in response.answer if rrset.rdtype==wanted for item in rrset] def show(response): def section(records):return [{'name':str(rr.name),'type':dns.rdatatype.to_text(rr.rdtype),'ttl':rr.ttl,'data':[item.to_text() for item in rr]} for rr in records] return {'rcode':dns.rcode.to_text(response.rcode()),'answer':section(response.answer),'authority':section(response.authority)} def wait(predicate,timeout=15): deadline=time.monotonic()+timeout while time.monotonic()<deadline: try: if predicate():return except (OSError,dns.exception.Timeout):pass time.sleep(.05) raise AssertionError('DNS fixture condition did not become true') def stop(process): if process.poll() is None: process.terminate() try:process.wait(timeout=5) except subprocess.TimeoutExpired:process.kill();process.wait(timeout=5) @contextmanager def lab(records,soa_ttl=8,minimum=4,extra=''): with TemporaryDirectory(prefix='xbcatm-dns-') as directory,ExitStack() as stack: root=Path(directory);zone=root/'zone';conf=root/'named.conf';log=root/'named.log';serial=1 def save(body): zone.write_text(f'$ORIGIN lab.test.\n$TTL 30\n@ {soa_ttl} IN SOA ns.lab.test. hostmaster.lab.test. ({serial} 60 60 3600 {minimum})\n@ IN NS ns.lab.test.\nns IN A 127.0.0.1\n'+body+'\n') run('named-checkzone','lab.test',str(zone)) save(records) conf.write_text(f'''options {{ directory "{root}"; listen-on port 1053 {{127.0.0.1;}}; listen-on-v6 {{none;}}; recursion no; dnssec-validation no; notify no; querylog yes; max-cache-size 8m; pid-file "{root}/named.pid"; session-keyfile none; {extra} }}; controls {{}}; zone "lab.test" {{ type primary; file "{zone}"; }}; ''') run('named-checkconf',str(conf)) output=stack.enter_context(log.open('w')) authority=subprocess.Popen(['named','-g','-c',str(conf),'-n','1','-U','1'],stdout=output,stderr=subprocess.STDOUT) stack.callback(stop,authority) def ready(): assert authority.poll() is None,log.read_text() response=query('lab.test.','SOA');return response.rcode()==0 and response.answer[0][0].serial==serial wait(ready) def update(body): nonlocal serial serial+=1;save(body);authority.send_signal(signal.SIGHUP);wait(ready) def cache(port=1054,upstream=1053,size=150,options=()): logfile=root/('cache-'+str(port)+'.log');stream=stack.enter_context(logfile.open('w')) command=['dnsmasq','--no-daemon','--conf-file=/dev/null','--no-resolv','--no-hosts','--bind-interfaces', '--listen-address=127.0.0.1','--port='+str(port),'--server=127.0.0.1#'+str(upstream), '--cache-size='+str(size),'--log-queries','--log-facility=-',*options] process=subprocess.Popen(command,stdout=stream,stderr=subprocess.STDOUT);stack.callback(stop,process) def cache_ready(): assert process.poll() is None,logfile.read_text() return query('lab.test.','SOA',port).rcode()==0 wait(cache_ready);return process,logfile def queries(name,kind='A'): return sum(bool(re.search(r'query: '+re.escape(name.rstrip('.'))+r' IN '+kind+r'\b',line)) for line in log.read_text().splitlines()) yield root,update,cache,queries def record(address,ttl):return f'move {ttl} IN A {address}' with lab(record('127.0.0.2',8)) as (_,update,cache,queries): cache();before=queries('move.lab.test');started=time.monotonic();initial=query('move.lab.test.',port=1054) wait(lambda: queries('move.lab.test')==before+1) assert values(initial)==['127.0.0.2'] and initial.answer[0].ttl==8 update(record('127.0.0.2',3));lowered=query('move.lab.test.');cached_after_lower=query('move.lab.test.',port=1054) assert lowered.answer[0].ttl==3 and cached_after_lower.answer[0].ttl>3 update(record('127.0.0.3',3));new_authority=query('move.lab.test.');old_cache=query('move.lab.test.',port=1054) assert values(new_authority)==['127.0.0.3'] and values(old_cache)==['127.0.0.2'] time.sleep(3.3);after_new_ttl=query('move.lab.test.',port=1054) assert values(after_new_ttl)==['127.0.0.2'] observations=[] def changed(): answer=query('move.lab.test.',port=1054);observations.append(show(answer));return values(answer)==['127.0.0.3'] wait(changed,timeout=8);migration_delay=round(time.monotonic()-started,3) update(record('127.0.0.2',3));rollback_authority=query('move.lab.test.');rollback_cache=query('move.lab.test.',port=1054) assert values(rollback_authority)==['127.0.0.2'] and values(rollback_cache)==['127.0.0.3'] wait(lambda: values(query('move.lab.test.',port=1054))==['127.0.0.2'],timeout=6) print(json.dumps({'initial':show(initial),'lowered_authority':show(lowered),'cache_after_lower':show(cached_after_lower), 'after_address_change':{'authority':show(new_authority),'cache':show(old_cache)},'after_new_TTL_elapsed':show(after_new_ttl), 'new_address_observed_after_seconds':migration_delay,'rollback':{'authority':show(rollback_authority),'cache':show(rollback_cache)}, 'rollback_eventually_observed':True},sort_keys=True)) """ name='xbcatm-dns-'+uuid.uuid4().hex[:10] def docker(*args,**kwargs):return subprocess.run(['docker',*args],capture_output=True,text=True,timeout=300,**kwargs) try: with TemporaryDirectory(prefix=name+'-') as directory: root=Path(directory);(root/'Dockerfile').write_text(recipe) built=docker('build','-t',name,str(root));assert built.returncode==0,built.stderr result=docker('run','--rm','-i','--name',name,'--network','none','--memory','384m','--pids-limit','128',name, '/usr/bin/python3','-',input=program) assert result.returncode==0,result.stderr print(json.dumps(json.loads(result.stdout),sort_keys=True)) finally: docker('rm','-f',name);docker('image','rm',name) 开始时，move.lab.test 指向 127.0.0.2，TTL 为八秒。经 dnsmasq 查询后存入该回答，权威日志确认发生一次上游查询。程序随后保持地址不变，把权威 TTL 改为三秒。直接查询看到三秒，先前填充的缓存却仍显示大于三秒的剩余 TTL。这说明修改源数据没有重写现有缓存状态，旧回答仍按照收到时的信息计时。 ...

状态码需要放在请求方法的上下文里理解，它告诉客户端怎样解释本次 HTTP 响应。首位数字只能缩小类别，不能说明完整业务结果。例如，任务已经接受，仍可能没有完成；三开头的响应，也未必表示页面搬家。服务器应根据实际完成的事情选择具体状态，并提供相应响应头和内容，避免让一个看起来合理的数字承担它从未承诺的含义。 成功不只有一种含义 二百表示请求按照相应方法的语义成功。成功读取任务状态资源，可以如实返回“任务仍在等待”的表示；这不代表任务本身已经结束。二零一表示创建了一个或多个资源，提供 Location 时，它用于识别主要的新建资源。接口需要说明客户端接下来应操作哪个资源，而不是把所有二开头的响应都解释成“用户最初想要的整个业务已经完成”。 二零二表示请求已经接受处理，但尚未完成，也不会对最终一定执行成功作出保证。应提供观察进度的途径，并定义最终成功与失败，否则客户端很难区分任务缓慢与任务丢失。二零四表示请求成功完成且没有响应内容；不能因为框架平常总是序列化 JSON，就给这种响应也附加正文。具体状态对应的消息约束需要贯穿最终输出。 三开头也可能是在验证已有表示 三零一与三零八表达永久重定向，三零二与三零七表达临时重定向，并具有不同的方法保留规则。三零三可以让调用方在处理步骤之后转而读取另一个资源。请求携带正文或具有副作用时，更要明确选择：可能改变方法的跳转，与必须保留方法的跳转，不能随意互换。客户端是否跟随，也应依据它实际要完成的动作判断。 三零四是另一种情况：它回应条件 GET 或 HEAD，表示无需重新传输选中的表示。接收方按缓存规则使用合适的已存内容，并更新相关元数据。三零四没有响应内容，也不是要求客户端沿 Location 导航。此时显示为空，是协议预期，并不能证明服务器把文件弄丢了。如果调用方没有相应缓存表示，就不能把这个空正文直接当作完整资源交给上层。 坏掉的样子 把所有四开头都归为“不能重试”，会误处理限流等情况；把所有五开头都归为“立即重试”，又可能重复非幂等操作或加剧过载。五零三覆盖暂时的服务不可用，包括过载或维护，并不只用于计划停机。重试决策还要结合方法、服务器提示和客户端预算。状态类别提供线索，不是能够无条件套用的完整重试算法。 四零四没有保证缺失只是暂时的：它可以表示找不到当前表示，也可以表示服务器不愿披露该表示。四一零才传达可能永久不可用的含义。同样，四零三并不普遍证明调用者已通过身份验证。业务代码不能在这些数字上附加比协议更强的假设；响应正文也应符合接口已声明的契约，否则下游即使正确识别了状态，仍可能误解实际处理结果。 快速验证 保存为 status-contract.py，运行 python3 status-contract.py。实验通过真实回环 HTTP 请求操作一个很小的内存任务状态和固定资源。这里使用的 Python HTTP 客户端不会自动跟随 Location，也没有实现缓存，所以脚本中每一步后续动作都明确写出。它不需要真实队列或数据库，避免把未验证的后台处理能力混进状态语义实验。 from http.client import HTTPConnection from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from threading import Thread import json job = {'state': 'absent'} class Handler(BaseHTTPRequestHandler): def reply(self, status, data=b'', headers=None): self.send_response(status) if status not in (204, 304): self.send_header('Content-Length', str(len(data))) for key, value in (headers or {}).items(): self.send_header(key, value) self.end_headers() if status not in (204, 304): self.wfile.write(data) def do_POST(self): if self.path != '/jobs': self.reply(404) return job['state'] = 'pending' self.reply(202, b'{"state":"pending"}', {'Content-Type': 'application/json', 'Location': '/jobs/42'}) def do_GET(self): if self.path == '/jobs/42': if job['state'] == 'complete': self.reply(303, headers={'Location': '/results/42'}) elif job['state'] == 'pending': self.reply(200, b'{"state":"pending"}', {'Content-Type': 'application/json'}) else: self.reply(404) elif self.path == '/results/42' and job['state'] == 'complete': self.reply(200, b'{"value":7}', {'Content-Type': 'application/json'}) elif self.path == '/asset': if self.headers.get('If-None-Match') == '"v1"': self.reply(304, headers={'ETag': '"v1"'}) else: self.reply(200, b'artifact', {'ETag': '"v1"'}) else: self.reply(404) def log_message(self, *args): pass server = ThreadingHTTPServer(('127.0.0.1', 0), Handler) thread = Thread(target=server.serve_forever) thread.start() try: responses = [] def request(method, path, headers=None): connection = HTTPConnection('127.0.0.1', server.server_port, timeout=5) try: connection.request(method, path, headers=headers or {}) response = connection.getresponse() value = {'status': response.status, 'body': response.read().decode(), 'location': response.getheader('Location'), 'etag': response.getheader('ETag')} responses.append(value) return value finally: connection.close() accepted = request('POST', '/jobs') assert accepted['status'] == 202 and job['state'] == 'pending' assert request('GET', '/results/42')['status'] == 404 assert json.loads(request('GET', accepted['location'])['body']) == {'state': 'pending'} job['state'] = 'complete' # Explicit fixture transition, not a background worker. redirect = request('GET', '/jobs/42') assert redirect['status'] == 303 and redirect['location'] == '/results/42' assert json.loads(request('GET', redirect['location'])['body']) == {'value': 7} cached = request('GET', '/asset') validation = request('GET', '/asset', {'If-None-Match': cached['etag']}) assert validation['status'] == 304 and validation['body'] == '' and validation['location'] is None assert cached['body'] == 'artifact' print(json.dumps({'responses': responses, 'reuse_cached_body': cached['body']})) finally: server.shutdown() thread.join() server.server_close() POST 首先返回二零二，任务此时仍为等待状态；结果地址仍返回四零四，读取状态资源则返回二百和 pending 表示。之后脚本显式把内存状态改成完成，这不是已经测试过的后台工作进程。状态端点随即返回三零三，客户端再读取 Location 指向的位置，取得数值为七的结果。每个数字描述的都是当前响应，并没有提前承诺后面的步骤一定成功。 ...

静态源是一道你能命名、能缓存、能回滚的边界。