浏览器拿一个一次性 code,再由服务端把它兑成访问令牌。

OAuth 授权码用码换令牌

让令牌只留在后端、只把短命的一次性 code 交给浏览器,能从很大程度上缩掉令牌在 URL 与日志里被偷走的地盘。 浏览器只拿 code access token 不该出现在浏览器里,这正是授权码流程存在的理由。浏览器跳转到授权端点、看到确认页,然后带着一个短命的一次性 code 回到回调地址——它从头到尾拿不到 token 本身。因为真正拿 token 的是后端:后端拿着 code,配上自己的 client secret,去 token 端点换。这个分工——浏览器只搬运 code,后端才掌握 secret——就是整个流程的意义所在。 后端负责兑换 交换发生在服务端,意味着 access token 只落在后端手里:浏览器既不会把它记进历史,客户端一次脚本 bug 也不会把它泄漏出去。后端构造 code + client_secret + redirect_uri 发给 token 端点,换回一个 JSON 的 access token(可选地还有 refresh)。这也解释了为什么没有 client secret 的单页应用(SPA)与原生应用,在这里总是敏感目标:它们没有 secret,所以需要哪「authorization code + PKCE」的组合,而不是普通 code flow。code flow 用一次后端的往返,换来 token 不经过浏览器这份安全。 ...

十月 20, 2027 · 2 分钟 · XB Field Notes
把一条请求藏进另一条里、再由后端放出的攻击。

请求走私把代理链路带偏

一条 HTTP/1.1 请求并不会自己说明它在哪里结束。它的边界是算出来的——靠 Content-Length,或者靠 chunked 传输编码——而链路上的每一跳都在各自独立地算。请求走私,就是两跳从同一串字节里算出了不同答案时发生的事。 RFC 9112 后来把规则写明了:Transfer-Encoding 压过 Content-Length;服务端可以拒绝同时带两者的报文;无论怎么选,之后都必须关闭连接。这在纸面上堵住了缝。真实的链路里仍然有早于这条规则的一跳、下游只会说 HTTP/1.0 的一跳,以及一个在入口终结 HTTP/2、再向内重新按 HTTP/1.1 组帧的边缘层。攻击就住在这些分歧里。 两个解析器,一条消息边界 组帧规则是有优先级的。若存在 Transfer-Encoding 且 chunked 是最后一个编码,报文体一直读到那个长度为 0 的终止块为止;否则一个合法的 Content-Length 给出字节数;两者都没有的请求没有报文体。当两个头同时出现,规范说 Transfer-Encoding 胜出——但只读 Content-Length 的那一跳会提前停下、把余下的字节转发出去,而那些余下的字节就是攻击。 实验把两者放在真实的 socket 上:前端按 Content-Length 组帧、原样转发字节,后端按 Transfer-Encoding 组帧。攻击者的 POST 同时声明 Content-Length: 57 与 Transfer-Encoding: chunked,报文体是 0\r\n\r\n 再加一条完整的请求。对前端来说那是一个 57 字节的报文体;对后端来说,chunked 报文体是零字节,后面那条请求是一条新报文。一条请求变成了两条,而前端从未见过后一条。 ...

十月 19, 2027 · 1 分钟 · XB Field Notes
将头、声明与签名压缩并以时间限定的 URL 安全令牌。

JWT 带着签名与寿命

把载荷当成公开信息,用签名保证其完整,一旦看见过期时间已过去就果断拒绝,别让一枚重放令牌活得比它该活的更久。 编码不加密 JWT 不是靠藏来保证安全的;它由 header、claims 与签名三段用 base64url 编码,用点连起来。任何拿到令牌的人都能反解出编码,逐条读出全部 claims。这是刻意为之,却也带来一条硬规矩:里面绝不允许放任何秘密——不能放密钥、不能放只有服务端才该看的会话数据。凡是你不会写进公开日志的东西,就别塞进 claim;需要回查的值用 jti 去查表,而不是直接放在载荷里。编码只是为了便于携带,不构成任何层面的保密。 签名先行为先 令牌只有在消费方先验证后才可信。正确的顺序是:先用信任、且认得出是签发者在握的那把钥匙去验签名,然后再去读 claims、核 audience。跳过签名、或是信 header 里 alg 的验签器,是最经典的失守姿势,因为伪造者只要把 alg 改成 none,或者把自己的公钥交出去,就能被当成真货。所以应把算法白名单钉死,核对签发者,不接受陌生方,绝不让令牌自己告诉你要用哪把钥匙。密钥与链路是信任的根基,要由可配置的密钥管理来决定,而不是由 token 自己说了算。 过期即契约 签名证明的是谁写了这枚令牌,而不是它多活到什么时候。exp 是它的寿命边界,必须强制执行:哪怕签名还新鲜,exp 一旦过去也该被拒绝。真实系统会留一点有界的时钟偏差,让刚签发出来的令牌别被 NTP 的几秒时钟偏差误杀,但绝不能盖过令牌本身的生命期。凡涉及敏感数据,都要配合基于 iat 的短租期,因为否则这枚状态无关、无法单点吊销的令牌会一路活到结束。过期即契约,也是唯一能让一枚被偷或重放的令牌不活过自己窗口的保障。 坏掉的样子 最典型的失守是验签器信任令牌自带的算法头。攻击者复制一份合法载荷,把头部改成无签名算法,去掉第三段签名,跳过验签的服务端就会直接放行。这不是载荷写错,而是验签路径被令牌劫持。对策是服务端钉死算法白名单,只接受一种算法家族,直接拒绝无签名算法,不根据头部切换逻辑,可接受算法写在配置里。 第二类经典故障是公钥与对称密钥之间的混淆。当同一个验签器同时接受非对称与对称算法,攻击者就能把本就公开的公钥拿来当作对称密钥,用它签发伪造令牌。验签器用同一份公开值去计算校验值,结果恰好能对上,伪造令牌就被当成真货。防范办法是每个签发者只允许一种算法家族,两类密钥分开存放,标识只做本地查表键,不指向文件路径或外部地址。 缺时间戳的故障最安静。没有过期时间的令牌永远不会过期,偷走之后可以一直用到密钥轮换;没有签发时间就没有锚点,时钟偏差检查与寿命审计都无从谈起。即使两者都有,若寿命动辄数天,一次泄露就等于长期会话。对策是签发时必带过期与签发时间,验签时强制检查,敏感接口只给几分钟租期,时钟容忍只给几十秒。 无状态带来的注销难题必须提前承认。服务端不存会话,就无法单独吊销某一枚令牌,登出时要么等它自然过期,要么维护标识黑名单。生产做法是短过期配合黑名单,登出即拉黑,敏感操作再回查服务端。同时牢记载荷只是编码,人人可读,不放口令与敏感信息,保密值只放引用标识,内容留在服务端按需查询。 快速验证 把下面代码存为验证脚本,用命令运行。全程只用标准库手写签发与验签,让签名比对过期检查与篡改拒绝都看得见。不依赖第三方包,不联网,用负寿命模拟过期,用改动载荷首字符模拟篡改。 import base64 import hashlib import hmac import json import time SECRET = b"demo-secret-key-change-me" def b64url_encode(raw: bytes) -> str: return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") def b64url_decode(data: str) -> bytes: padding = "=" * (-len(data) % 4) return base64.urlsafe_b64decode(data + padding) def issue(subject: str, lifetime: int = 600) -> str: header = b64url_encode( json.dumps({"alg": "HS256", "typ": "JWT"}, separators=(",", ":")).encode() ) now = int(time.time()) payload = b64url_encode( json.dumps( {"sub": subject, "iat": now, "exp": now + lifetime}, separators=(",", ":"), ).encode() ) signature = b64url_encode( hmac.new(SECRET, f"{header}.{payload}".encode(), hashlib.sha256).digest() ) return f"{header}.{payload}.{signature}" def verify(token: str) -> dict: header_b64, payload_b64, signature_b64 = token.split(".") header = json.loads(b64url_decode(header_b64)) assert header.get("alg") == "HS256", f"unexpected alg: {header.get('alg')}" expected = b64url_encode( hmac.new(SECRET, f"{header_b64}.{payload_b64}".encode(), hashlib.sha256).digest() ) assert hmac.compare_digest(expected, signature_b64), "bad signature" claims = json.loads(b64url_decode(payload_b64)) assert "exp" in claims and "iat" in claims, "missing exp/iat" assert int(time.time()) < int(claims["exp"]), "token expired" return claims token = issue("demo-user") claims = verify(token) assert claims["sub"] == "demo-user" print(f"valid: sub={claims['sub']}") stale = issue("demo-user", lifetime=-10) try: verify(stale) raise AssertionError("expired token must not verify") except AssertionError as exc: assert str(exc) == "token expired", exc print("expired correctly rejected: token expired") head, pay, sig = token.split(".") tampered_pay = ("A" if pay[0] != "A" else "B") + pay[1:] try: verify(f"{head}.{tampered_pay}.{sig}") raise AssertionError("tampered token must not verify") except AssertionError as exc: assert str(exc) == "bad signature", exc print("tampered correctly rejected: bad signature") print("PASS: valid accepted; expired and tampered rejected") 脚本先签发正常令牌并验签通过,证明正常路径畅通。再用负寿命签发过期令牌,确认以过期原因拒绝。最后篡改载荷首字符,确认以签名不匹配拒绝。三组断言覆盖正常过期与伪造三种行为。预期输出共四行,分别是主体一行两行拒绝原因与一行汇总。 ...

十月 18, 2027 · 2 分钟 · XB Field Notes
固定 SQL 结构接收绑定值,数据库仍可选择不同执行计划。

预备语句把取值与 SQL 结构分开

参数绑定为数据库中的取值提供一个明确位置。姓名里即使带有引号或分号,也仍然只是值,不会因此变成另一段查询条件。复用命名预备语句则是另一项优化:服务端能够在同一个会话内,跨多次执行保留语句信息。安全地绑定数据,并不要求应用为每一条查询都显式建立一个命名语句,也不要求先证明查询会频繁重复。 坏掉的样子 在 Psycopg 中,占位符 %s 属于驱动接口,不是让程序使用 Python 字符串格式化。值应作为 execute 的第二个参数传入,占位符不要加引号,只有一个位置参数时也需要单元素元组。如果先把值插入字符串,再调用一个支持参数化的接口,原本希望保护的边界早就已经被越过;接口名称本身不会替程序补救这次拼接。 占位符代表值,不代表表名、列名或关键字。动态标识符需要明确的结构选择:先用白名单限制支持的列,再用驱动的标识符组合功能正确引用选中的列名。升降序关键字也应由应用控制。分页的数量与偏移量可以直接绑定,同时仍须按产品要求验证范围。能在语法上接收一个整数,不代表允许请求任意大的页面或无限制消耗资源。 预备语句不会把第一次的执行计划永久钉死。PostgreSQL 区分根据本次参数生成的定制计划,以及跨不同参数复用的通用计划;自动选择时会考虑规划开销与估算执行成本。模式或统计信息变化也可能触发重新规划。因此,语句保持参数化时,访问路径仍可以变化。是否复用某种计划,与输入能否变成 SQL 语法,是不同的性质,不能混为一谈。 快速验证 准备安装了 Psycopg 3 的 Python 环境,把 PG_DSN 设置为可丢弃的 PostgreSQL 数据库连接。连接配置沿用已有安全提供方式。实测版本为 Psycopg 3.3.5 与 PostgreSQL 18.6。将下面代码保存为 Python 文件,在已设置该环境变量的终端执行。使用的是临时表,连接关闭后会自动消失: import os import psycopg from psycopg import sql with psycopg.connect(os.environ['PG_DSN'], autocommit=True, prepare_threshold=5) as conn: conn.execute('CREATE TEMP TABLE people (id integer PRIMARY KEY, label text NOT NULL)') conn.execute("INSERT INTO people VALUES (1,'alice'),(2,'bob')") query = 'SELECT id FROM people WHERE label=%s ORDER BY id LIMIT %s' hostile = "' OR true --" assert conn.execute(query, (hostile, 10), prepare=False).fetchall() == [] count = conn.execute('SELECT count(*) FROM pg_prepared_statements').fetchone()[0] assert count == 0 print('bound without named preparation:', count, 'named statements; no match') for value in ("O'Reilly", r'a\b', 'x; DROP TABLE people; --'): assert conn.execute('SELECT %s::text', (value,), prepare=False).fetchone() == (value,) for label, expected in (('alice', [(1,)]), ('bob', [(2,)])): assert conn.execute(query, (label, 10), prepare=True).fetchall() == expected assert conn.execute('SELECT count(*) FROM pg_prepared_statements').fetchone() == (1,) print('named preparation reused: alice=1, bob=2') def ordered(column, limit): if column not in {'id', 'label'}: raise ValueError('unsupported sort column') if type(limit) is not int or not 1 <= limit <= 100: raise ValueError('limit must be an integer from 1 through 100') statement = sql.SQL('SELECT id FROM people ORDER BY {} LIMIT %s').format( sql.Identifier(column)) return conn.execute(statement, (limit,), prepare=False).fetchall() assert ordered('label', 1) == [(1,)] for column, limit in [('label; DROP TABLE people', 1), ('id', 0), ('id', 101)]: try: ordered(column, limit) except ValueError: pass else: raise AssertionError('invalid request accepted') print('identifier allow-list and bound LIMIT checked') conn.execute('PREPARE plan_lab(text) AS SELECT id FROM people WHERE label=$1') conn.execute('SET plan_cache_mode=force_custom_plan') assert conn.execute("EXECUTE plan_lab('alice')").fetchall() == [(1,)] counters = "SELECT generic_plans,custom_plans FROM pg_prepared_statements WHERE name='plan_lab'" assert conn.execute(counters).fetchone() == (0, 1) conn.execute('SET plan_cache_mode=force_generic_plan') assert conn.execute("EXECUTE plan_lab('bob')").fetchall() == [(2,)] assert conn.execute(counters).fetchone() == (1, 1) print('same prepared statement: custom=1, generic=1') conn.execute('RESET plan_cache_mode') conn.execute('DEALLOCATE plan_lab') 第一项输出显示命名语句数量为零,而且看起来像注入的输入没有匹配任何记录。这次调用明确关闭命名预备,却保留参数绑定,证明安全边界在语句复用之前就存在。随后包含引号、反斜杠、分号的文本都原样返回,表仍可被后面的查询继续使用,字符串里写着删除命令也没有获得执行机会。 ...

十月 16, 2027 · 2 分钟 · XB Field Notes
一个小工具带着属主的身份运行,换来一点集中的能力。

Setuid 已属主身份运行

setuid 二进制以文件属主的身份运行,而不是以启动它的人的身份。内核在 exec 时刻替换凭据,passwd 因此能在普通用户并不拥有 /etc/shadow 写权限的情况下改写它。这个位属于文件,而它交出去的是属主的整个身份,不是某一项窄权限。 故意做这种交换是件奇怪的事:你等于把一整个账户托付给一个程序,于是程序里任何一处缺陷都会变成提权缺陷。何况这套机制会在好几种有据可查的情况下静默失效,而 capabilities、用户命名空间与 socket 激活,今天已经覆盖了 setuid 原本的大部分用途。真正值得问的,往往不是「怎么把这个程序写对」,而是「这件事能不能不要 setuid」。 身份跟着文件走 execve(2) 把规则写得很直白:若程序文件上设置了 set-user-ID 位,调用进程的有效用户 ID 就变为该文件属主的 ID。真实 UID 与真实 GID 保持不变,新的有效 UID 随后被复制进 saved set-user-ID。真实、有效、保存这三份凭据,就是这套机制的全部。 这些位在程序跑起来之前就能看到:stat 显示 /usr/bin/sudo 的模式是 4511(-r-s--x--x)、/usr/bin/su 是 4755,两者属主都是 uid 0,而 /usr/bin/passwd 只是普通的 0755。哪些二进制带这个位,是每台机器各自的部署事实,只能去读,不能假设。 ...

十月 2, 2027 · 2 分钟 · XB Field Notes
一处谁都能写的临时空间,文件名可能相撞。

临时目录是公用空间

/tmp 是名字的共享命名空间,不是私有的暂存区。任何本地用户都能创建你的进程即将使用的那个名字,所以名字并不构成对文件的占有;而 hier(7) 明确说,里面的内容可能在任何时候被删除,而且不会通知你。 粘滞位让这个目录可以安全地被共享,却没有让它能被可预测地使用。因此修法不是改权限,而是改系统调用的形状:用一次调用同时生成名字并创建文件,中间不留任何空隙给别人抢占。 1777 才让共享成为可能 模式 1777 是「人人可写」加上粘滞位,而在这台主机上这不是假设:/tmp 是指向 private/tmp 的符号链接,后者是 drwxrwxrwt。没有粘滞位,目录里任何能写的人都能删除或重命名别人的文件;有了它,只有文件属主、目录属主或 root 可以。正是这一个位,让 /tmp 能被许多互不信任的用户同时使用。 粘滞位做不到的事,是在文件存在之前保护那个名字。open("/tmp/svc.log", "w") 会跟随一个已经存在的符号链接,于是写入落在链接指向的地方,而不是一个新文件里。实验用真实的 open(2) 验证了这一点,随后验证了两次拒绝:O_CREAT|O_EXCL 因名字已被占用而返回 EEXIST,O_CREAT|O_NOFOLLOW 因最后一段是符号链接而返回 ELOOP。 不是你创建的文件,权限不属于你 同一个缺陷的另一半更隐蔽:既没有符号链接,也没有可见的竞争。O_CREAT 只在它真正创建文件时才应用模式参数,所以一个被预先创建、人人可读的文件,在你用模式 0600 打开它之后仍然保持 0666,你写进去的秘密也就一直人人可读。实验用真实的 stat(2) 复现了这一点。 修法是让一次调用同时完成命名与创建。mkstemp(3) 把契约写得很清楚:模板的最后六个字符是 XXXXXX,文件以权限 0600 创建,并以 O_EXCL 打开,从而保证创建者就是调用者。mktemp(1) 在 shell 里做同样的事,会遵循 TMPDIR,默认模板为 tmp.XXXXXXXXXX。它的 -u 只打印名字而不创建任何东西,手册直接把它标为 “(unsafe)";实验确认那个被打印出来的名字之后仍然空闲,而这正是 mkstemp 要关掉的那扇窗。如果工作单元是目录而不是文件,mkdtemp(3) 应用同一条规则:以模式 0700 创建它,别人进不去。这一步同样是一次调用完成命名与创建,没有留给别人的窗口。 ...

九月 26, 2027 · 1 分钟 · XB Field Notes
三个相连的方块,依次标注观察、决策与行动。

框架策略明确谁可以嵌入当前页面

点击劫持利用用户对界面的误判,让一个动作落到并非其真实意图的控件上。恶意父页面可以把被嵌入的应用放在误导性视觉层后面,使用户点击实际进入该应用中的真实按钮。父页面不必能够读取子页面的文档结构,用户仍然可以与框架里的内容交互。因此,同源读取限制不能替代应用是否允许被嵌入的独立决定,也不能仅靠看不到对方文档就判断点击过程安全。 被嵌入的响应可以通过内容安全策略禁止任何祖先嵌入,或者明确列出允许的祖先来源。这里写的应当是获准包含当前文档的页面来源。假设商户页面嵌入支付页面,应由支付页面的策略允许商户来源;仅列出支付组件自己的来源,并不能表达商户有权嵌入它。配置之前先明确哪一份响应属于子页面、哪个来源是父页面,可以避免把允许列表方向写反。 把限制放到正确的响应上 嵌入祖先指令回答谁能嵌入当前文档,框架来源指令则回答当前文档自己可以加载哪些框架。给子页面设置严格的 frame-src,并不会阻止另一个父页面嵌入它。默认来源全部拒绝,也不会作为 frame-ancestors 的回退值。这些指令控制的是关系中的不同方向,不能仅因为名字都涉及框架,就把其中一个当成另一个的替代品。 嵌入祖先指令不能通过 CSP 的 meta 元素生效,必须通过响应头交付。遇到多层嵌套时,每一级祖先都需要符合策略,而不只是直接父级。某个组件即使可以合法放在指定应用中,当该应用本身又被放进未经许可的外层框架时,仍可能被阻止。因此制定允许列表时,要审阅整个预期嵌入链,不能只凭最内层组件和它紧邻的容器作判断。 坏掉的样子 一次部署可能只给容器页面增加了策略,却让真正敏感的子页面响应保持不受限制;另一次部署可能把策略写入 HTML 元数据,看到配置文字存在就误以为浏览器会执行。应检查实际被框架加载的响应,并验证浏览器是否渲染。仅收到二百状态,既不能证明嵌入成功,也不能证明没有框架防护,因为浏览器可以收到文档后,拒绝在当前祖先环境中显示它。 互相冲突的响应头也可能误导审阅者。在支持这套规则的现代浏览器里,强制执行的嵌入祖先策略优先于该响应的 X-Frame-Options。给允许某个父来源的策略再附加一个 DENY,并不必然得到期望中的更严格结果。如果确实需要保留多个响应头,应让它们与实际受众和浏览器支持约定一致,不能假定它们都像多份 CSP 那样自然取限制交集。 快速验证 将代码保存为 frame-policy.py,执行 python3 frame-policy.py –serve,并打开打印出的父页面地址。本次实际使用 Python 3.14.4 与 Chrome 152。父服务和子服务位于不同的本机回环端口,六份子响应只有合成说明文字,以及表示成功渲染的消息。父页面只接受预期子来源发来的消息并展示结果,结束后按控制键加 C 关闭两个临时服务,不会操作真实账号或业务控件。 import json import sys from http.server import BaseHTTPRequestHandler, HTTPServer from threading import Event, Thread from urllib.request import urlopen paths = ["open", "csp-block", "xfo-block", "csp-allow", "meta", "frame-src"] requests = [] class Child(BaseHTTPRequestHandler): def log_message(self, *args): pass def do_GET(self): name = self.path.lstrip("/") assert name in paths requests.append(name) meta = ('<meta http-equiv="Content-Security-Policy" content="frame-ancestors \'none\'">' if name == "meta" else "") body = ("<!doctype html><meta charset=utf-8>" + meta + "<p>Rendered: " + name + "</p>" "<script>parent.postMessage({frame:" + json.dumps(name) + "}," + json.dumps(app_origin) + ");</script>").encode() self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") if name == "csp-block": self.send_header("Content-Security-Policy", "frame-ancestors 'none'") elif name == "xfo-block": self.send_header("X-Frame-Options", "DENY") elif name == "csp-allow": self.send_header("Content-Security-Policy", "frame-ancestors " + app_origin) self.send_header("X-Frame-Options", "DENY") elif name == "frame-src": self.send_header("Content-Security-Policy", "frame-src 'none'") self.end_headers() self.wfile.write(body) class Parent(BaseHTTPRequestHandler): def log_message(self, *args): pass def do_GET(self): if self.path == "/events": body = json.dumps(requests).encode() kind = "application/json" else: body = ("""<!doctype html><meta charset="utf-8"><title>Frame policy lab</title> <h1>Which child responses can this parent embed?</h1><pre id="result">Waiting for frame messages...</pre> <button id="check">Show server requests</button><div id="frames"></div><script> const child=""" + json.dumps(child_origin) + ";const names=" + json.dumps(paths) + """; window.received=[]; window.snapshot=async()=>{ const result={rendered:[...window.received].sort(),serverRequests:await (await fetch('/events')).json()}; document.getElementById('result').textContent=JSON.stringify(result,null,2);return result; }; addEventListener('message',event=>{ if(event.origin===child && names.includes(event.data?.frame)) { window.received.push(event.data.frame); document.getElementById('result').textContent=JSON.stringify(window.received,null,2); } }); document.getElementById('check').onclick=()=>window.snapshot(); for(const name of names) { const frame=document.createElement('iframe');frame.src=child+'/'+name;frame.title=name; frame.width=240;frame.height=100;document.getElementById('frames').append(frame); } </script>""").encode() kind = "text/html; charset=utf-8" self.send_response(200) self.send_header("Content-Type", kind) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) child = HTTPServer(("127.0.0.1", 0), Child) app = HTTPServer(("127.0.0.1", 0), Parent) child_origin = f"http://127.0.0.1:{child.server_port}" app_origin = f"http://127.0.0.1:{app.server_port}" threads = [Thread(target=server.serve_forever, daemon=True) for server in (child, app)] for thread in threads: thread.start() try: for name in paths: with urlopen(child_origin + "/" + name, timeout=5) as response: assert response.status == 200 if name == "csp-allow": assert response.headers["Content-Security-Policy"] == "frame-ancestors " + app_origin assert response.headers["X-Frame-Options"] == "DENY" response.read() requests.clear() print(json.dumps({"app_url": app_origin, "child_origin": child_origin, "wire_responses": 6}), flush=True) if "--serve" in sys.argv: Event().wait() except KeyboardInterrupt: pass finally: for server in (child, app): server.shutdown() server.server_close() for thread in threads: thread.join(timeout=5) 开放子页面应正常渲染;csp-block 被禁止祖先嵌入策略拒绝,xfo-block 被 DENY 拒绝。csp-allow 虽然同时发送 DENY,但强制祖先策略明确允许当前父来源,因此正常渲染。meta 页面也会渲染,因为该交付方式不能执行嵌入祖先指令;frame-src 页面同样渲染,因为它限制的是子页面接下来自己加载的框架,而不是正在验证的父子嵌入关系。 ...

九月 23, 2027 · 2 分钟 · XB Field Notes
XML 解析器被实体引用诱导去读了一个本地文件。

XXE 展开实体的代价

XML 外部实体处理可能通过解析器去读本地文件、访问原本够不到的主机,而且这一切都是服务器解析时替你完成的。应当在解析器里把它禁掉。 实体是解析器的功能 XML 实体是解析器自带的一个功能,让文档可以引用别处的内容;也正是同一套机制,在得不到防护时把一条请求变成一次可无限扩展的实体炸弹。一份文档可以定义这样一个实体:它指向本机某个文件、某个网络地址,或另一个实体;解析器一旦去解析它,就会把文件内容读进来、塞进文档中「占位」的位置。便利与危险常常是双胞胎——允许多引用的解析器,自然也就允许去翻 /etc/passwd,或者把 URL 递到网络的另一端。 外部指针会泄文件 外部实体之所以会「泄」,是因为解析器替你去读了它。一次典型的 XXE 攻击,会发送一条指向文件路径或 URL 的实体声明;而一个过于宽松的解析器,就会把文件读出来、在响应里把其中一部分当作「占位符」的回显——于是一台内部文件或内网地址,就在攻击者只发送了一段 XML 帧的情况下被看见。伤害到底多大,取决于解析器允许访问什么;它的边界不是请求的体积,而是解析这段文档的进程权限。 干脆把这个功能关掉 最直接也最强的补救,是把这部分功能整个关掉:禁用外部普通实体与外部参数实体,决定解析器「是否允许翻网络」,并避开那些默认就去解析的外部解析库。在确实绕不开实体时,用 schema 对文档设一道闸,再用恰好够用的最小权限去跑解析。把 XXE 当做一个可开关的配置特性,而不是某段代码里的漏洞,才能让那些要处理历史数据的低层系统依旧安全——会决定「啃不啃得动一台机器」的,从来不是这份 XML 有没有内容,而是解析器有没有钥匙。 坏掉的样子 最典型的坏法是读文件:文档里声明一个指向本地路径的外部实体,应用又把展开后的内容回显到报错信息或预览字段里。请求本身只有几行,真正读磁盘的是解析器进程,所以访问日志看起来人畜无害,泄露只出现在响应正文。能读到什么不取决于攻击者,而取决于跑解析器的账号权限,服务账号能看到的秘密都可能变成远程可读。 同样的形状把网址放进系统标识符,就会变成服务端请求伪造。攻击者让解析器去访问云厂商的元数据地址拿凭证,或者去敲只有内网才能到达的管理后台。每次重试都是服务器替攻击者再跑一次,所以外部实体会放大伪造请求的伤害:一行很短的字符串,就能换来一次带内网身份的抓取。 还有一种不需要网络的坏法,叫实体炸弹。几个互相嵌套的内部实体,每层膨胀约十倍,一千字节的文档就能展开成几吉字节的文本,模式校验还没跑起来,内存就先被耗尽。容易被忽略的是图片上传:浏览器把可缩放矢量图形当作标记语言解析,于是头像通道里夹带文档类型声明,就把同一套机制从图片管线带了进来。防御常见的裂缝是不一致,主接口关掉了文档类型,订阅源、安全断言或随后重新解析文件的缩略图程序却还开着,攻击者总能找到那扇没关的门。 快速验证 把下面的程序保存到临时工作目录的xxe_probe.py,运行python3 xxe_probe.py。它只用标准库,不访问网络也不读写磁盘,对比宽松基线与加固入口的差别。基线会把内部实体炸弹展开到一千个字符,加固入口则拒绝一切声明了实体的文档类型。 """XXE guard demo: stdlib only, no network, no disk reads.""" import io import re import xml.sax import xml.sax.handler BENIGN = '<?xml version="1.0"?><note><to>ops</to><body>hello</body></note>' FILE_READ = ('<?xml version="1.0"?><!DOCTYPE r [<!ENTITY xxe SYSTEM "file:///etc/passwd">]>' '<r>&xxe;</r>') SSRF = ('<?xml version="1.0"?><!DOCTYPE r [<!ENTITY xxe SYSTEM ' '"http://169.254.169.254/latest/meta-data/">]><r>&xxe;</r>') BILLION_LAUGHS = ('<?xml version="1.0"?><!DOCTYPE l [<!ENTITY a "0123456789">' '<!ENTITY b "&a;&a;&a;&a;&a;&a;&a;&a;&a;&a;">' '<!ENTITY c "&b;&b;&b;&b;&b;&b;&b;&b;&b;&b;">]>' '<l>&c;</l>') SVG_UPLOAD = ('<?xml version="1.0"?><!DOCTYPE svg [<!ENTITY xxe SYSTEM "file:///etc/hostname">]>' '<svg xmlns="http://www.w3.org/2000/svg"><text>&xxe;</text></svg>') class _Text(xml.sax.handler.ContentHandler): def __init__(self): self.chunks = [] def characters(self, content): self.chunks.append(content) class _RejectResolver(xml.sax.handler.EntityResolver): def resolveEntity(self, publicId, systemId): raise xml.sax.SAXException(f"blocked external entity: {systemId!r}") def naive_parse(xml_text): """Permissive baseline: plain stdlib parse with no guard.""" parser = xml.sax.make_parser() handler = _Text() parser.setContentHandler(handler) parser.parse(io.StringIO(xml_text)) return "".join(handler.chunks) def safe_parse(xml_text): """Hardened entry point: refuse entity machinery, then lock the parser.""" if re.search(r"<!DOCTYPE|<!ENTITY", xml_text, re.IGNORECASE): raise ValueError("rejected: DOCTYPE/entity declarations are not accepted") parser = xml.sax.make_parser() parser.setFeature(xml.sax.handler.feature_external_ges, False) try: parser.setFeature(xml.sax.handler.feature_external_pes, False) except xml.sax.SAXNotSupportedException: pass # expat never fetches parameter entities; keep going parser.setEntityResolver(_RejectResolver()) handler = _Text() parser.setContentHandler(handler) parser.parse(io.StringIO(xml_text)) return "".join(handler.chunks) def check_rejected(name, payload): try: safe_parse(payload) except ValueError as exc: print(f"blocked {name}: {exc}") return raise AssertionError(f"safe_parse accepted {name}; expected ValueError") def main(): text = safe_parse(BENIGN) assert "hello" in text, f"benign parse lost content: {text!r}" print(f"benign ok: {text.strip()!r}") expanded = naive_parse(BILLION_LAUGHS) assert len(expanded) >= 1000, f"baseline did not expand: {len(expanded)} chars" print(f"baseline expands billion-laughs to {len(expanded)} chars (unsafe)") for name, payload in [("file-read", FILE_READ), ("ssrf", SSRF), ("billion-laughs", BILLION_LAUGHS), ("svg-upload", SVG_UPLOAD)]: check_rejected(name, payload) print("PASS: 1 benign accepted, 1 baseline expansion shown, 4 malicious inputs rejected") if __name__ == "__main__": main() 运行成功会先打印正常文档的文本,再报告基线展开的长度并标记为不安全,随后列出文件读取、服务端伪造、实体炸弹和图形上传四种输入被拦截,最后以通过行收尾。加固函数在解析前拒绝任何文档类型或实体声明,再关闭外部普通实体与外部参数实体,并装上遇到残留引用就抛异常的解析器。这些断言已在苹果电脑的 Python 3.14 上运行通过。演示只覆盖解析器配置,生产环境还要配合白名单、模式校验,以及在文件落盘前就拒绝文档类型的上传管线。 ...

九月 22, 2027 · 2 分钟 · XB Field Notes