diff --git a/governance/elevation.py b/governance/elevation.py new file mode 100644 index 0000000..854e097 --- /dev/null +++ b/governance/elevation.py @@ -0,0 +1,257 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""elevation.py —— JIT 提权 v0 裁决/收回引擎(IR-0006 W2-C4 / BEH-05 / 卡 #415) + +/elevate 评论 → 本表裁决(governance/policy/elevation.yaml,默认拒绝)→ +批准/拒绝记录按 schema v1 落 elevation-ledger(governance/elevation/ +shadow-evidence.jsonl,kind=approval,evidence-query 第 4 源——subject.card +可查询,AC-9c);TTL 到期 sweep 收回(AC-9d:sweep 后零"过期未收回"grant= +无长期驻留提权断言,open-check 机器锚点)。 + +评论格式(单行命令 + kv 参数,乱序允许): + + /elevate capability=org-variable-write ttl=30 reason=复位熔断前的根因排查 spec=specs/IR-0006/spec.md#AC-9 + +- capability=NAME(词法 token);ttl=分钟数(可缺省→policy defaults); + spec=<引用>(词法 token);reason=自由文本(到下一个 " key=" 或行尾)。 +- HO 场景 3:reason 或 spec 缺失 → deny(policy request.required 执法)。 + +子命令: + parse --comment-file F --card owner/repo#n --requester L --delivery-id ID + → stdout 请求 JSON(capability/ttl/reason/spec_ref/delivery_id...) + adjudicate --request-file F --role R [--policy P] [--now ISO] + → stdout 裁决 JSON(verdict=grant|deny + reason/effective_ttl/expires_at) + sweep --ledger-dir D [--policy P] [--now ISO] + → stdout 到期未收回 grant 列表 JSON(供 workflow 补 revoke 事件) + open-check --ledger-dir D [--now ISO] + → 无过期未收回 grant 断言(exit 0=通过 / 3=有驻留——sweep 后必须 0) + +退出码:0=成功 | 2=参数/环境/策略非法(fail-closed——裁决输入不可信不判) +""" +import argparse +import datetime +import glob +import json +import os +import re +import sys + +try: + import yaml +except ImportError: # pragma: no cover + yaml = None + +DEFAULT_POLICY = os.path.join(os.path.dirname(os.path.abspath(__file__)), + "policy", "elevation.yaml") +CAP_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$") + + +def die(code, msg): + print(msg, file=sys.stderr) + sys.exit(code) + + +def now_iso(now_arg): + if now_arg: + try: + return datetime.datetime.strptime(now_arg, "%Y-%m-%dT%H:%M:%SZ") \ + .replace(tzinfo=datetime.timezone.utc) + except ValueError: + die(2, f"--now 非法 ISO8601Z: {now_arg!r}") + return datetime.datetime.now(datetime.timezone.utc) + + +def iso(dt): + return dt.strftime("%Y-%m-%dT%H:%M:%SZ") + + +def parse_comment(body: str, card: str, requester: str, delivery_id: str) -> dict: + """提取 /elevate 命令与 kv 参数(首行;reason 为到下一 kv token 或行尾的自由文本)。""" + first = (body or "").splitlines()[0].strip() if body else "" + tokens = first.split() + if not tokens or tokens[0] != "/elevate": + die(2, "首 token 非 /elevate(调用方应先过滤)") + rest = " ".join(tokens[1:]) + # kv token 定界:前有空白且 key ∈ {capability, ttl, spec, reason} + kv_re = re.compile(r"(?:^|\s)(capability|ttl|spec|reason)\s*=\s*", re.IGNORECASE) + req = {"kind": "elevate", "card": card, "requester": requester, + "delivery_id": delivery_id} + for m in kv_re.finditer(rest): + key = m.group(1).lower() + start = m.end() + nxt = kv_re.search(rest, start) + val = rest[start:nxt.start() if nxt else len(rest)].strip() + if key == "ttl": + if not re.match(r"^[0-9]+$", val): + die(2, f"ttl 非法(须为正整数分钟): {val!r}") + req["ttl"] = int(val) + elif key == "spec": # 评论用 spec=,请求 JSON 规范化为 spec_ref + req["spec_ref"] = val + else: + req[key] = val + if "capability" in req and not CAP_RE.match(req["capability"]): + die(2, f"capability 非法: {req['capability']!r}") + return req + + +def load_policy(path: str) -> dict: + if yaml is None: + die(2, "缺 PyYAML——策略表无法加载(fail-closed)") + try: + pol = yaml.safe_load(open(path, encoding="utf-8")) + except (OSError, yaml.YAMLError) as e: + die(2, f"策略表加载失败: {e}") + caps = pol.get("capabilities") or {} + d = pol.get("defaults") or {} + req = pol.get("request") or {} + if not isinstance(caps, dict) or not caps: + die(2, "策略表缺非空 capabilities") + if not isinstance(d.get("ttl_minutes"), int) or d["ttl_minutes"] <= 0: + die(2, "策略表 defaults.ttl_minutes 须为正整数") + required = req.get("required") or ["reason", "spec_ref"] + if not isinstance(required, list): + die(2, "策略表 request.required 须为列表") + for name, c in caps.items(): + if not CAP_RE.match(name): + die(2, f"能力名非法: {name!r}") + if not isinstance(c.get("allowed_roles"), list) or not c["allowed_roles"]: + die(2, f"capabilities.{name}.allowed_roles 须为非空列表") + mt = c.get("max_ttl_minutes") + if not isinstance(mt, int) or mt <= 0: + die(2, f"capabilities.{name}.max_ttl_minutes 须为正整数") + return pol + + +def adjudicate(req: dict, role: str, policy: dict, now: datetime.datetime) -> dict: + """策略裁决(纯函数)。返回 {verdict: grant|deny, reason, ...}。""" + caps = policy["capabilities"] + required = (policy.get("request") or {}).get("required") or ["reason", "spec_ref"] + out = {"verdict": "deny", "requester": req.get("requester"), + "card": req.get("card"), "delivery_id": req.get("delivery_id")} + # HO 场景 3:缺理由/缺 spec 引用必拒(policy request.required 执法) + for field in required: + key = "spec_ref" if field == "spec_ref" else field + if not str(req.get(key) or "").strip(): + out["reason"] = f"missing-{key}(HO 场景 3:{'理由' if key == 'reason' else 'spec 引用'}必附,fail-closed)" + return out + cap = req.get("capability") + if cap not in caps: + out["reason"] = f"capability 未在策略表声明(默认拒绝): {cap!r}" + return out + c = caps[cap] + if role not in c["allowed_roles"]: + out["reason"] = f"role 不匹配(capability={cap} 允许 {c['allowed_roles']},请求方 role={role!r})" + return out + ttl = req.get("ttl") + if ttl is None: + ttl = policy["defaults"]["ttl_minutes"] + if ttl <= 0 or ttl > c["max_ttl_minutes"]: + out["reason"] = (f"ttl 越界(capability={cap} 上限 {c['max_ttl_minutes']} 分钟," + f"请求 {ttl})") + return out + expires = now + datetime.timedelta(minutes=ttl) + out.update({"verdict": "grant", "capability": cap, "reason": "", + "effective_ttl_minutes": ttl, "granted_at": iso(now), + "expires_at": iso(expires), + "elevation_id": "elev-" + iso(now).replace("-", "").replace(":", "") + .replace("T", "-").replace("Z", "") + "-" + cap, + "request_reason": req.get("reason"), "spec_ref": req.get("spec_ref")}) + return out + + +def load_ledger(ledger_dir: str) -> list: + recs = [] + for f in sorted(glob.glob(os.path.join(ledger_dir, "shadow-evidence*.jsonl"))): + with open(f, encoding="utf-8") as fh: + for ln in (l.strip() for l in fh): + if ln: + recs.append(json.loads(ln)) + return recs + + +def payload_of(rec: dict) -> dict: + p = rec.get("payload") + if isinstance(p, str): + try: + return json.loads(p) + except json.JSONDecodeError: + return {} + return p if isinstance(p, dict) else {} + + +def open_grants(ledger_dir: str, now: datetime.datetime) -> list: + """到期未收回的 grant(action=elevation.grant 且无同 elevation_id 的 revoke)。""" + revoked = {payload_of(r).get("elevation_id") + for r in load_ledger(ledger_dir) + if r.get("action") == "elevation.revoke"} + out = [] + for r in load_ledger(ledger_dir): + if r.get("action") != "elevation.grant": + continue + p = payload_of(r) + eid = p.get("elevation_id") + if eid in revoked: + continue + exp = p.get("expires_at") + if exp and exp < iso(now): # 字典序即时间序(同 ISO 格式) + out.append({"elevation_id": eid, "card": (r.get("subject") or {}).get("card"), + "capability": p.get("capability"), + "requester": (r.get("actor") or {}).get("identity"), + "expires_at": exp, "grant_record": r}) + return out + + +def main(): + ap = argparse.ArgumentParser(prog="elevation.py", + description="JIT 提权 v0 裁决/收回引擎(IR-0006 W2-C4)") + sub = ap.add_subparsers(dest="cmd", required=True) + + p = sub.add_parser("parse") + p.add_argument("--comment-file", required=True) + p.add_argument("--card", required=True) + p.add_argument("--requester", required=True) + p.add_argument("--delivery-id", required=True) + + p = sub.add_parser("adjudicate") + p.add_argument("--request-file", required=True) + p.add_argument("--role", required=True, help="owner/agent/none(调用方 API 判定)") + p.add_argument("--policy", default=DEFAULT_POLICY) + p.add_argument("--now", default=None, help="ISO8601Z(测试注入;缺省=当前)") + + p = sub.add_parser("sweep") + p.add_argument("--ledger-dir", required=True) + p.add_argument("--now", default=None) + + p = sub.add_parser("open-check") + p.add_argument("--ledger-dir", required=True) + p.add_argument("--now", default=None) + + a = ap.parse_args() + if a.cmd == "parse": + with open(a.comment_file, encoding="utf-8") as f: + body = f.read() + req = parse_comment(body, a.card, a.requester, a.delivery_id) + print(json.dumps(req, ensure_ascii=False, sort_keys=True)) + elif a.cmd == "adjudicate": + with open(a.request_file, encoding="utf-8") as f: + req = json.load(f) + pol = load_policy(a.policy) + v = adjudicate(req, a.role, pol, now_iso(a.now)) + print(json.dumps(v, ensure_ascii=False, sort_keys=True)) + elif a.cmd == "sweep": + expired = open_grants(a.ledger_dir, now_iso(a.now)) + slim = [{k: g[k] for k in ("elevation_id", "card", "capability", + "requester", "expires_at")} for g in expired] + print(json.dumps(slim, ensure_ascii=False, sort_keys=True)) + elif a.cmd == "open-check": + expired = open_grants(a.ledger_dir, now_iso(a.now)) + if expired: + for g in expired: + print(f"ELEVATION-STALE {g['elevation_id']} card={g['card']} " + f"expired={g['expires_at']}", file=sys.stderr) + die(3, f"存在 {len(expired)} 条过期未收回 grant——JIT 驻留断言失败(AC-9d)") + print("OK 零过期未收回 grant(无长期驻留提权断言通过,AC-9d)") + + +if __name__ == "__main__": + main() diff --git a/governance/policy/elevation.yaml b/governance/policy/elevation.yaml new file mode 100644 index 0000000..822389b --- /dev/null +++ b/governance/policy/elevation.yaml @@ -0,0 +1,34 @@ +schema_version: 1 +# ============================================================================ +# elevation.yaml —— JIT 提权策略表(唯一授权真源) +# 卡: IR-0006 W2-C4(.github#415)/ BEH-05 / HO 场景 3 +# 执行: governance/elevation.py adjudicate(转 .github/workflows/elevation.yml 消费) +# +# 铁律(默认拒绝):能力未声明/角色不匹配/超 TTL/缺理由或 spec 引用 = deny。 +# 提权是瞬时能力(JIT):grant 记录带 expires_at,elevation.yml 每小时 sweep +# 到期未收回的 grant 补 revoke 记录(AC-9d:无长期驻留提权断言=open-check)。 +# 批准/拒绝/收回记录统一按 schema v1 落 elevation-ledger(kind=approval, +# subject.card 可查询——evidence-query 第 4 源)。 +# ============================================================================ +defaults: + # 未声明 ttl 的缺省(BEH-05:TTL≤波次——波次租约上限 240 分钟,对齐 + # arbiter capabilities.yaml defaults.ttl_minutes) + ttl_minutes: 240 +request: + # HO 场景 3(凭证失效面):无理由或无 spec 引用必拒——fail-closed 方向 + required: [reason, spec_ref] +capabilities: + # 受控能力档(allowlist;未知=拒绝)。v0 声明三档(写面分级): + org-variable-write: + # org Actions 变量写(例:熔断复位 AUTO_MERGE_DISABLED PATCH——ADR-0040 + # 复位路径人工面,提权代签走本档;短 TTL 严控) + allowed_roles: [agent, owner] + max_ttl_minutes: 60 + label-write: + # 治理标签写(state:* 切换补偿等) + allowed_roles: [agent, owner] + max_ttl_minutes: 240 + ledger-append: + # 台账面写(elevation 自身账本外) + allowed_roles: [agent, owner] + max_ttl_minutes: 120 diff --git a/governance/tests/test-elevation.sh b/governance/tests/test-elevation.sh new file mode 100644 index 0000000..ffdb2ce --- /dev/null +++ b/governance/tests/test-elevation.sh @@ -0,0 +1,166 @@ +#!/usr/bin/env bash +# test-elevation.sh —— JIT 提权 v0 裁决/收回引擎自测(IR-0006 W2-C4 / 卡 #415) +# +# 覆盖(卡 AC 对应): +# AC-9c parse(kv 乱序/reason 含空格/ttl 非法拒绝) +# adjudicate:HO 场景 3 负向(无 reason / 无 spec 必拒); +# 未知 capability 拒;角色不匹配拒;ttl 超档拒;缺省 ttl+grant 全字段 +# AC-9d sweep(到期未收回列表)+ open-check 断言(过期驻留=exit 3; +# 已 revoke 不再计;未到期不计) +# 策略表加载 fail-closed(缺 capabilities / max_ttl 非法 → exit 2) +# 用法: bash governance/tests/test-elevation.sh(gate.yml 自动纳入) +set -uo pipefail +DIR="$(cd "$(dirname "$0")/../.." && pwd)" +FAILS=0 +pass() { echo "PASS $1"; } +fail() { echo "FAIL $1"; FAILS=$((FAILS+1)); } + +TMP=$(mktemp -d); trap 'rm -rf "$TMP"' EXIT +ELEV="$DIR/governance/elevation.py" +POLICY="$DIR/governance/policy/elevation.yaml" +NOW="2026-08-29T10:00:00Z" + +req() { # req + printf '%s\n' "$2" >"$1" +} + +# ---- parse:kv 乱序 + reason 含空格 ---- +req "$TMP/c1" "/elevate ttl=30 spec=specs/IR-0006/spec.md#AC-9 capability=org-variable-write reason=复位熔断前的根因排查与处置" +python3 "$ELEV" parse --comment-file "$TMP/c1" --card Cloudbird-Software/.github#415 \ + --requester agent-x --delivery-id node-1 >"$TMP/r1.json" \ + && pass "parse 乱序 kv" || fail "parse 乱序 kv" +python3 - "$TMP/r1.json" <<'PYEOF' && pass "parse 字段断言(reason 含空格完整保留)" || fail "parse 字段断言" +import json, sys +r = json.load(open(sys.argv[1], encoding="utf-8")) +assert r["capability"] == "org-variable-write", r +assert r["ttl"] == 30, r +assert r["spec_ref"] == "specs/IR-0006/spec.md#AC-9", r +assert r["reason"] == "复位熔断前的根因排查与处置", r +assert r["delivery_id"] == "node-1" and r["requester"] == "agent-x", r +print("OK") +PYEOF + +# ---- parse:ttl 非法 → exit 2(fail-closed) ---- +req "$TMP/c2" "/elevate capability=label-write ttl=abc reason=x spec=y" +python3 "$ELEV" parse --comment-file "$TMP/c2" --card a/b#1 --requester t --delivery-id d >/dev/null 2>&1 +[[ $? -eq 2 ]] && pass "parse ttl 非法 → exit 2" || fail "parse ttl 非法未拒" + +# ---- adjudicate:HO 场景 3 负向(AC-9c 核心) ---- +mk() { # mk + cat >"$1" <"$TMP/v1.json" && pass "无 reason 裁决出结果" || fail "无 reason 裁决失败" +python3 - "$TMP/v1.json" <<'PYEOF' && pass "HO 场景 3:无 reason 必拒(fail-closed)" || fail "HO 场景 3 无 reason 未拒" +import json, sys +v = json.load(open(sys.argv[1], encoding="utf-8")) +assert v["verdict"] == "deny" and "missing-reason" in v["reason"], v +print("OK") +PYEOF + +mk "$TMP/req-nospec" "org-variable-write" '"有理由"' '""' +python3 "$ELEV" adjudicate --request-file "$TMP/req-nospec" --role agent \ + --policy "$POLICY" --now "$NOW" >"$TMP/v2.json" +python3 - "$TMP/v2.json" <<'PYEOF' && pass "HO 场景 3:无 spec 引用必拒" || fail "HO 场景 3 无 spec 未拒" +import json, sys +v = json.load(open(sys.argv[1], encoding="utf-8")) +assert v["verdict"] == "deny" and "missing-spec_ref" in v["reason"], v +print("OK") +PYEOF + +# ---- adjudicate:未知 capability / 角色不匹配 / ttl 超档 ---- +mk "$TMP/req-badcap" "root-shell" '"r"' '"s"' +python3 "$ELEV" adjudicate --request-file "$TMP/req-badcap" --role agent --policy "$POLICY" --now "$NOW" \ + | grep -q '"verdict": "deny"' && pass "未知 capability 拒(默认拒绝)" || fail "未知 capability 未拒" + +mk "$TMP/req-role" "org-variable-write" '"r"' '"s"' +python3 "$ELEV" adjudicate --request-file "$TMP/req-role" --role none --policy "$POLICY" --now "$NOW" \ + | grep -q '"verdict": "deny"' && pass "角色不匹配拒(none)" || fail "角色不匹配未拒" + +cat >"$TMP/req-ttl" <<'EOF' +{"kind":"elevate","card":"a/b#1","requester":"t","delivery_id":"d", + "capability":"org-variable-write","reason":"r","spec_ref":"s","ttl":61} +EOF +python3 "$ELEV" adjudicate --request-file "$TMP/req-ttl" --role agent --policy "$POLICY" --now "$NOW" \ + | grep -q '"verdict": "deny"' && pass "ttl 超档拒(org-variable-write 上限 60)" || fail "ttl 超档未拒" + +# ---- adjudicate:正向 grant(缺省 ttl 生效 + 全字段 + role=owner) ---- +cat >"$TMP/req-ok" <<'EOF' +{"kind":"elevate","card":"Cloudbird-Software/.github#415","requester":"agent-x", + "delivery_id":"node-ok","capability":"label-write","reason":"状态机补偿写", + "spec_ref":"specs/IR-0006/spec.md#AC-9"} +EOF +python3 "$ELEV" adjudicate --request-file "$TMP/req-ok" --role agent --policy "$POLICY" --now "$NOW" >"$TMP/vg.json" +python3 - "$TMP/vg.json" <<'PYEOF' && pass "grant 正向(缺省 ttl=240 + expires_at/elevation_id/request_reason)" || fail "grant 正向断言" +import json, sys +v = json.load(open(sys.argv[1], encoding="utf-8")) +assert v["verdict"] == "grant", v +assert v["effective_ttl_minutes"] == 240, v +assert v["expires_at"] == "2026-08-29T14:00:00Z", v +assert v["elevation_id"].startswith("elev-20260829-10"), v +assert v["request_reason"] == "状态机补偿写" and v["spec_ref"].endswith("#AC-9"), v +print("OK") +PYEOF +python3 "$ELEV" adjudicate --request-file "$TMP/req-ok" --role owner --policy "$POLICY" --now "$NOW" \ + | grep -q '"verdict": "grant"' && pass "owner 亦过档" || fail "owner 被误拒" + +# ---- 策略表 fail-closed ---- +sed 's/^capabilities:/capabilities_broken:/' "$POLICY" >"$TMP/badpol.yaml" +python3 "$ELEV" adjudicate --request-file "$TMP/req-ok" --role agent --policy "$TMP/badpol.yaml" >/dev/null 2>&1 +[[ $? -eq 2 ]] && pass "策略缺 capabilities → exit 2(fail-closed)" || fail "策略非法未拒" +sed 's/max_ttl_minutes: 60/max_ttl_minutes: -1/' "$POLICY" >"$TMP/badpol2.yaml" +python3 "$ELEV" adjudicate --request-file "$TMP/req-ok" --role agent --policy "$TMP/badpol2.yaml" >/dev/null 2>&1 +[[ $? -eq 2 ]] && pass "max_ttl 非法 → exit 2(fail-closed)" || fail "max_ttl 非法未拒" + +# ---- sweep + open-check(AC-9d) ---- +LED="$TMP/elev"; mkdir -p "$LED" +# 账本:grant A(过期未收回)、grant B(未到期)、grant C(过期已收回)、非 grant 记录 +python3 - "$LED" <<'PYEOF' +import datetime, json, os, sys +led = sys.argv[1] +def rec(action, eid, exp, cap="label-write", card="Cloudbird-Software/.github#415", actor="agent-x"): + return json.dumps({ + "ts": "2026-08-29T09:00:00Z", "kind": "approval", "action": action, + "verdict": "pass", "subject": {"card": card, "tenant": "cloudbird-internal"}, + "actor": {"identity": actor, "role": "agent"}, + "payload": json.dumps({"elevation_id": eid, "capability": cap, + "expires_at": exp, "request_reason": "r", "spec_ref": "s"}), + "seq": 0, "prev_hash": None, "hash": "h"}, ensure_ascii=False, sort_keys=True) +rows = [ + rec("elevation.grant", "elev-A", "2026-08-29T09:30:00Z"), + rec("elevation.grant", "elev-B", "2026-08-29T23:00:00Z"), + rec("elevation.grant", "elev-C", "2026-08-29T08:00:00Z"), + rec("elevation.revoke", "elev-C", None), + rec("elevation.deny", None, None), +] +open(os.path.join(led, "shadow-evidence.jsonl"), "w", encoding="utf-8").write("\n".join(rows) + "\n") +PYEOF +python3 "$ELEV" sweep --ledger-dir "$LED" --now "$NOW" >"$TMP/sw.json" +python3 - "$TMP/sw.json" <<'PYEOF' && pass "sweep:仅过期未收回的 elev-A 入列(B 未到期/C 已收回/deny 不计)" || fail "sweep 列表断言" +import json, sys +rows = json.load(open(sys.argv[1], encoding="utf-8")) +assert [r["elevation_id"] for r in rows] == ["elev-A"], rows +assert rows[0]["card"].endswith("#415") and rows[0]["capability"] == "label-write", rows +print("OK") +PYEOF +python3 "$ELEV" open-check --ledger-dir "$LED" --now "$NOW" >/dev/null 2>&1 +[[ $? -eq 3 ]] && pass "open-check:过期驻留 → exit 3(断言红)" || fail "open-check 未红" +# 收回 elev-A 后断言转绿 +echo "$(cat "$LED/shadow-evidence.jsonl")" > "$LED/shadow-evidence.jsonl" +python3 - "$LED/shadow-evidence.jsonl" <<'PYEOF' +import json, sys +rows = [json.loads(l) for l in open(sys.argv[1], encoding="utf-8") if l.strip()] +rows.append({**rows[0], "action": "elevation.revoke", "payload": json.dumps({ + "elevation_id": "elev-A", "cause": "ttl-expired"})}) +open(sys.argv[1], "w", encoding="utf-8").write("\n".join(json.dumps(r, ensure_ascii=False, sort_keys=True) for r in rows) + "\n") +PYEOF +python3 "$ELEV" open-check --ledger-dir "$LED" --now "$NOW" >/dev/null 2>&1 +[[ $? -eq 0 ]] && pass "AC-9d:收回后 open-check 绿(无长期驻留提权)" || fail "收回后断言仍红" + +echo "----------------------------------------" +if [[ $FAILS -eq 0 ]]; then echo "test-elevation: PASS"; exit 0; fi +echo "test-elevation: $FAILS 处失败"; exit 1