凌晨救火还要等三个人开权限：生产应急访问 break-glass 与操作审计的 6 个工程决策

#!/usr/bin/env bash
# file: seal-check.sh
# 作用：巡检 break-glass 封存凭证的完整性、有效期与轮换时间
# 依赖：gpg, jq, awscli(或 s3cmd, 视存储后端而定), curl(用于告警 webhook)
# 输入：环境变量 SEAL_BUCKET / SEAL_KEY / ALERT_WEBHOOK
# 输出：标准输出打印巡检结果；异常时退码 1 并推送到告警 webhook
# 建议频率：每日一次。为什么每日：凭证轮换本身不频繁，但"被封存对象被误删/被改动"这类
# 事故需要快速发现，日巡检能在低成本下把最大发现延迟压到 24 小时内。

set -euo pipefail

SEAL_BUCKET="${SEAL_BUCKET:-sre-breakglass-seal}"
SEAL_KEY="${SEAL_KEY:-prod/emergency/envelope.gpg}"
ALERT_WEBHOOK="${ALERT_WEBHOOK:-}"
STALE_DAYS=180   # 超过 180 天未轮换即判为异常。为什么是 180：等保要求密钥类凭据定期轮换，
                 # 而应急凭据使用频率极低，半年是一个既能覆盖常见合规口径、
                 # 又不至于把轮换成本抬得过高的折中值。
EXIT=0

alert() {
  echo "[ALERT] $1"
  if [[ -n "${ALERT_WEBHOOK}" ]]; then
    # 用 --data-urlencode 避免消息里的特殊字符破坏 JSON
    curl -sS -X POST -H 'Content-Type: application/json' \
      --data-urlencode "text=break-glass 封存巡检异常: $1" \
      "${ALERT_WEBHOOK}" >/dev/null || true
  fi
  EXIT=1
}

# 1) 封存对象是否存在
if ! aws s3api head-object --bucket "${SEAL_BUCKET}" --key "${SEAL_KEY}" >/dev/null 2>&1; then
  alert "封存对象不存在: s3://${SEAL_BUCKET}/${SEAL_KEY}"
  exit "${EXIT}"
fi

# 2) 读取对象元数据：最后修改时间 + 是否被标记为最新版本
META=$(aws s3api head-object --bucket "${SEAL_BUCKET}" --key "${SEAL_KEY}" \
        --query '{lm:LastModified,ver:VersionId}' --output json)
LAST_MODIFIED=$(echo "${META}" | jq -r '.lm')
echo "封存对象最后修改时间: ${LAST_MODIFIED}"

# 3) 距上次轮换的天数
LAST_TS=$(date -u -jf "%Y-%m-%dT%H:%M:%S" "${LAST_MODIFIED%%.*}" +%s 2>/dev/null || \
          date -u -d "${LAST_MODIFIED}" +%s)
NOW_TS=$(date -u +%s)
AGE_DAYS=$(( (NOW_TS - LAST_TS) / 86400 ))
echo "距上次轮换: ${AGE_DAYS} 天"
if (( AGE_DAYS > STALE_DAYS )); then
  alert "封存凭证已 ${AGE_DAYS} 天未轮换，超过阈值 ${STALE_DAYS} 天"
fi

# 4) 私钥分片是否齐全（分片存于不同成员的独立存储，这里只校验对象存在）
for i in 1 2 3; do
  if ! aws s3api head-object --bucket "${SEAL_BUCKET}" --key "shares/shard-${i}.gpg" >/dev/null 2>&1; then
    alert "私钥分片缺失: shard-${i}"
  fi
done

# 5) 校验密文可解密性（仅用公钥试解，不落盘明文）
# 为什么只做“可解密性”校验：完整解密会产生明文材料，巡检不应持有明文。
# 这里用 gpg 的 --list-packets 检查密文结构完整，不输出明文。
if ! gpg --list-packets <(aws s3 cp "s3://${SEAL_BUCKET}/${SEAL_KEY}" - 2>/dev/null) >/dev/null 2>&1; then
  alert "封存密文结构异常，可能已损坏"
fi

echo "巡检完成，退码 ${EXIT}"
exit "${EXIT}"

#!/usr/bin/env python3
# file: jit_grantd.py
"""
jit-grantd —— 最小可运行的限时提权（JIT）与自动回收服务。

【依赖】仅 Python 3.9+ 标准库，无需 pip 安装任何包。
        生产环境建议替换 http.server 为 FastAPI/uvicorn，并在前置网关做 mTLS，
        但核心的“签发-到期-回收-审计”状态机与本文完全一致。

【运行】
    python3 jit_grantd.py            # 监听 127.0.0.1:8787
【输入】
    POST /grant   {"subject":"alice","resource":"ns/payment","verbs":["patch"],
                   "ttl_seconds":900,"reason":"INC-20261027-0317","severity":"SEV1"}
    POST /revoke  {"grant_id":"g-xxxx"}
    GET  /grants  列出当前有效授权
【输出】
    JSON。每次操作都写入 audit_log 表，包含 who/what/when/why，供事后审计。
"""

import json
import sqlite3
import secrets
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse

DB_PATH = "jit_grants.db"
LISTEN = ("127.0.0.1", 8787)

# TTL 上下限。为什么在服务端强制夹取：客户端参数不可信，如果允许任意 TTL，
# 一次“手滑”填了 86400 就退化成常备权限，所以边界必须在服务端收口。
TTL_MIN, TTL_MAX = 300, 1800   # 5 分钟 ~ 30 分钟


def init_db():
    con = sqlite3.connect(DB_PATH)
    con.executescript(
        """
        -- grants：当前有效的临时授权
        CREATE TABLE IF NOT EXISTS grants (
            grant_id    TEXT PRIMARY KEY,
            subject     TEXT NOT NULL,      -- 谁
            resource    TEXT NOT NULL,      -- 对什么（最小范围）
            verbs       TEXT NOT NULL,      -- 能做什么（JSON 数组）
            issued_at   INTEGER NOT NULL,
            expire_at   INTEGER NOT NULL,   -- 到点必须回收
            severity    TEXT NOT NULL,
            status      TEXT NOT NULL DEFAULT 'active'  -- active/expired/revoked
        );
        -- audit_log：只追加，不允许 UPDATE/DELETE，保证审计链完整
        CREATE TABLE IF NOT EXISTS audit_log (
            id        INTEGER PRIMARY KEY AUTOINCREMENT,
            ts        INTEGER NOT NULL,
            action    TEXT NOT NULL,        -- grant/revoke/expire
            grant_id  TEXT,
            subject   TEXT,
            detail    TEXT NOT NULL         -- JSON：包含 reason 等“为什么”
        );
        CREATE INDEX IF NOT EXISTS idx_grants_expire ON grants(expire_at, status);
        """
    )
    con.commit()
    con.close()


def audit(con, action, grant_id, subject, detail):
    """审计写入与业务写入必须在同一事务里，避免“授了权但没留痕”。"""
    con.execute(
        "INSERT INTO audit_log(ts, action, grant_id, subject, detail) VALUES (?,?,?,?,?)",
        (int(time.time()), action, grant_id, subject, json.dumps(detail, ensure_ascii=False)),
    )


def issue(subject, resource, verbs, ttl_seconds, reason, severity):
    ttl = max(TTL_MIN, min(TTL_MAX, int(ttl_seconds)))
    now = int(time.time())
    gid = "g-" + secrets.token_hex(6)
    con = sqlite3.connect(DB_PATH)
    with con:  # 事务：业务与审计同生共死
        con.execute(
            "INSERT INTO grants(grant_id,subject,resource,verbs,issued_at,expire_at,severity)"
            " VALUES (?,?,?,?,?,?,?)",
            (gid, subject, resource, json.dumps(verbs), now, now + ttl, severity),
        )
        audit(con, "grant", gid, subject,
              {"resource": resource, "verbs": verbs, "ttl": ttl, "reason": reason})
    con.close()
    return {"grant_id": gid, "expire_at": now + ttl, "ttl": ttl}


def revoke(gid, by="system"):
    con = sqlite3.connect(DB_PATH)
    with con:
        cur = con.execute(
            "UPDATE grants SET status='revoked' WHERE grant_id=? AND status='active'", (gid,)
        )
        if cur.rowcount:
            audit(con, "revoke", gid, by, {"by": by})
    con.close()
    return {"grant_id": gid, "revoked": True}


def sweeper():
    """
    到期回收线程：每 5 秒扫一次，把过期的 active 授权置为 expired 并留痕。
    为什么要独立线程而不是在请求里顺手回收：授权可能在没有任何后续请求的情况下
    自然到期（例如工程师提权后什么都没干），如果只靠请求触发回收，
    这个“没被用到的权限”就会一直挂着，正是我们要消灭的僵尸权限。
    """
    while True:
        try:
            con = sqlite3.connect(DB_PATH)
            now = int(time.time())
            rows = con.execute(
                "SELECT grant_id, subject FROM grants WHERE status='active' AND expire_at<=?",
                (now,),
            ).fetchall()
            if rows:
                with con:
                    for gid, subject in rows:
                        con.execute("UPDATE grants SET status='expired' WHERE grant_id=?", (gid,))
                        audit(con, "expire", gid, subject, {"expired_at": now})
            con.close()
        except Exception as exc:  # 回收线程绝不能因单次异常死掉
            print("[sweeper] error:", exc)
        time.sleep(5)


class Handler(BaseHTTPRequestHandler):
    def _json(self, code, payload):
        body = json.dumps(payload, ensure_ascii=False).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        if urlparse(self.path).path != "/grants":
            return self._json(404, {"error": "not found"})
        con = sqlite3.connect(DB_PATH)
        rows = con.execute(
            "SELECT grant_id,subject,resource,verbs,expire_at,severity FROM grants"
            " WHERE status='active' AND expire_at>?", (int(time.time()),),
        ).fetchall()
        con.close()
        self._json(200, {"grants": rows})

    def do_POST(self):
        path = urlparse(self.path).path
        length = int(self.headers.get("Content-Length", 0))
        try:
            data = json.loads(self.rfile.read(length) or b"{}")
        except json.JSONDecodeError:
            return self._json(400, {"error": "invalid json"})

        if path == "/grant":
            missing = [k for k in ("subject", "resource", "reason") if not data.get(k)]
            if missing:
                # reason 必填：没有“为什么”的授权在事后审计里毫无价值
                return self._json(400, {"error": f"missing: {','.join(missing)}"})
            return self._json(200, issue(
                data["subject"], data["resource"], data.get("verbs", ["patch"]),
                data.get("ttl_seconds", 900), data["reason"], data.get("severity", "SEV3"),
            ))
        if path == "/revoke":
            return self._json(200, revoke(data.get("grant_id", ""), data.get("by", "system")))
        self._json(404, {"error": "not found"})

    def log_message(self, *args):
        pass  # 默认访问日志噪声大，审计由 audit_log 表承担


if __name__ == "__main__":
    init_db()
    threading.Thread(target=sweeper, daemon=True).start()
    print(f"jit-grantd listening on http://{LISTEN[0]}:{LISTEN[1]}  (db={DB_PATH})")
    ThreadingHTTPServer(LISTEN, Handler).serve_forever()

# 1) 申请一个 15 分钟的临时提权（TTL 会被服务端夹到 300~1800 秒区间内）
curl -s -X POST http://127.0.0.1:8787/grant \
  -H 'Content-Type: application/json' \
  -d '{"subject":"alice","resource":"ns/payment","verbs":["patch"],
       "ttl_seconds":900,"reason":"INC-20261027-0317","severity":"SEV1"}'
# => {"grant_id":"g-3f9a1c2b7e01","expire_at":1761500000,"ttl":900}

# 2) 查看当前有效授权
curl -s http://127.0.0.1:8787/grants

# 3) 一个 10 分钟 TTL 的授权，10 分钟后会被 sweeper 置为 expired，并留下 expire 审计记录

# file: jit-rolebinding.yaml
# 一次性、带 TTL 注解的命名空间级授权。注意 verbs 只给 patch，
# 不给 delete——应急恢复配置绝大多数只需要 patch 或 rollout restart，不需要删除权。
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: jit-patch-ns-payment
  namespace: payment
  annotations:
    sre.wang/issued-for: "alice"
    sre.wang/incident: "INC-20261027-0317"
    sre.wang/ttl-seconds: "900"
    sre.wang/expire-at: "1761500000"   # 由 jit-grantd 回填，控制器据此回收
rules:
  - apiGroups: ["apps"]
    resources: ["deployments"]
    verbs: ["patch", "get", "watch"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: jit-patch-ns-payment-binding
  namespace: payment
  annotations:
    sre.wang/ttl-seconds: "900"
    sre.wang/expire-at: "1761500000"
subjects:
  - kind: User
    name: alice
    apiGroup: rbac.authorization.k8s.io
roleRef:
  kind: Role
  name: jit-patch-ns-payment
  apiGroup: rbac.authorization.k8s.io

# file: require-ttl-rolebinding.yaml （Kyverno 策略）
# 为什么要在准入层拦：如果只靠事后巡检回收，巡检周期内的窗口（可能长达数小时）
# 就是漏洞。准入层保证“不符合规范的授权根本创建不出来”，是唯一能在创建瞬间收口的点。
apiVersion: kyverno.io/v1
kind: ClusterPolicy
metadata:
  name: require-ttl-on-jit-rolebinding
spec:
  validationFailureAction: Enforce
  rules:
    - name: jit-binding-must-have-ttl
      match:
        any:
          - resources:
              kinds: ["RoleBinding"]
      preconditions:
        all:
          - key: "{{ request.object.metadata.annotations.\"sre.wang/issued-for\" || '' }}"
            operator: NotEquals
            value: ""
      validate:
        message: "JIT RoleBinding 必须带 sre.wang/expire-at 注解，且有效期不超过 1800 秒"
        pattern:
          metadata:
            annotations:
              sre.wang/expire-at: "?*"
              sre.wang/ttl-seconds: "?*"

# file: breakglass-alert-rules.yaml
# 依赖：jit-grantd 暴露 /metrics（本文的示例服务可加一个 exporter），
# 或由巡检脚本把授权数量推成 Pushgateway 指标。
groups:
  - name: break-glass.rules
    rules:
      # 规则一：任何一次 break-glass 使用，立即告警（不等 for 时间）
      # 为什么 for: 0s：应急权限的使用本身就是一等事件，不需要“持续 N 分钟”才报。
      - alert: BreakGlassGrantIssued
        expr: increase(jit_grant_issued_total{channel="breakglass"}[5m]) > 0
        for: 0s
        labels:
          severity: critical
          category: privileged_access
        annotations:
          summary: "生产 break-glass 提权被使用"
          description: >-
            过去 5 分钟出现 {{ $value }} 次 break-glass 提权。
            subject={{ $labels.subject }} resource={{ $labels.resource }}
            incident={{ $labels.incident }}。请立即到事件群确认操作内容。            
          runbook_url: "https://www.sre.wang/posts/sre-break-glass-emergency-access/"

      # 规则二：授权数量异常升高（可能有人在批量申请，或系统被滥用）
      - alert: JitActiveGrantsTooMany
        expr: jit_grant_active > 20
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "当前有效 JIT 授权数量异常（{{ $value }} 条）"
          description: "正常水位应低于 5 条。请检查是否存在批量申请或授权未回收。"

      # 规则三：授权到期后仍在使用（回收失效 = 最危险的信号）
      # 为什么单独设这一条：到期未回收意味着账本状态和实际能力出现漂移，
      # 这比“权限给多了”更严重——因为它说明回收机制本身坏了。
      - alert: JitGrantOverdueStillActive
        expr: jit_grant_overdue_active > 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "存在 {{ $value }} 条已过期但仍处于 active 的授权"
          description: "自动回收可能失效，请立即手工 revoke 并排查 sweeper。"

      # 规则四：应急提权使用后 24 小时未提交复盘
      - alert: BreakGlassReviewOverdue
        expr: jit_breakglass_without_review_seconds > 86400
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "break-glass 使用已超过 24 小时未提交复盘"
          description: "incident={{ $labels.incident }}，请补齐事后复盘记录。"

-- file: audit_schema.sql
-- 设计要点：审计表只追加，不更新、不删除；所有写入必须和业务操作同事务。
-- 字段设计围绕复盘需要的五个问题：谁、对什么、做了什么、为什么、什么时候。

CREATE TABLE access_audit (
    id              BIGSERIAL PRIMARY KEY,
    ts              TIMESTAMPTZ NOT NULL DEFAULT now(),   -- when
    actor           TEXT        NOT NULL,                  -- who（必须是具体人，禁止 shared）
    actor_source_ip INET        NOT NULL,                  -- 从哪来（辅助归因）
    channel         TEXT        NOT NULL,                  -- 通道：jit / breakglass / standby
    grant_id        TEXT,                                  -- 关联授权记录
    resource_type   TEXT        NOT NULL,                  -- 对象类型：host/namespace/device
    resource_id     TEXT        NOT NULL,                  -- 对象标识（复盘定位的关键）
    action          TEXT        NOT NULL,                  -- 做了什么：patch/rollback/restart
    command         TEXT,                                  -- 原始命令（命令级审计）
    session_id      TEXT,                                  -- 会话 ID（与会话录制关联）
    incident_id     TEXT,                                  -- 关联事件号（串起时间线）
    reason          TEXT        NOT NULL,                  -- 为什么（复盘必需）
    outcome         TEXT        NOT NULL,                  -- success/failure
    approval_ref    TEXT                                   -- 审批工单号（有审批时非空）
);

-- 查询模式：复盘时按事件号聚合全部操作，按时间排序还原时间线
CREATE INDEX idx_audit_incident ON access_audit (incident_id, ts);
-- 查询模式：审计时按人查一段时间内的全部生产操作
CREATE INDEX idx_audit_actor_time ON access_audit (actor, ts DESC);

-- 审计表必须防篡改。生产做法：撤销普通账号的 UPDATE/DELETE 权限，
-- 只保留 INSERT，并通过数据库审计或外部日志落一份只读副本。
REVOKE UPDATE, DELETE ON access_audit FROM app_rw;

-- 还原某个事件下所有生产操作，按时间排序，与告警时间线对照
SELECT
    a.ts            AS 操作时间,
    a.actor         AS 操作人,
    a.channel       AS 通道,
    a.resource_id   AS 对象,
    a.action        AS 动作,
    a.outcome       AS 结果,
    a.reason        AS 原因
FROM access_audit a
WHERE a.incident_id = 'INC-20261027-0317'
ORDER BY a.ts;

#!/usr/bin/env python3
# file: stale_grant_audit.py
"""
超期权限巡检 —— 扫描各权限源，找出“已过期但仍存在”的授权并告警。

【依赖】标准库 + requests（pip install requests）；kubectl/aws 等 CLI 需在 PATH 中。
【输入】无（配置在脚本头部的 SOURCES 里）。
【输出】超期授权清单；有发现时退码 1。建议由定时任务每日运行，或接进值班日报。
【为什么用巡检兜底】自动回收服务负责“正常路径”上的回收，但如果 sweeper 崩了、
    权限被直接建在 K8s 里绕过了 jit-grantd、或者授权是通过控制台手工加的，
    这些都不会被自动回收覆盖。巡检是最后一道防线，作用类似数据库的定期一致性校验。
"""

import json
import subprocess
import sys
import time
from datetime import datetime, timezone

import requests

JIT_API = "http://127.0.0.1:8787"
GRACE_SECONDS = 300  # 允许 5 分钟宽限，避免与 sweeper 的扫描周期抢跑


def stale_from_jit():
    """检查 jit-grantd 里是否有过期但仍 active 的授权（回收失效信号）。"""
    stale = []
    try:
        grants = requests.get(f"{JIT_API}/grants", timeout=5).json().get("grants", [])
    except requests.RequestException as exc:
        # API 不可达本身也是异常，但不应让整轮巡检失败——只是无法覆盖这一源
        return [{"source": "jit", "error": f"api unreachable: {exc}"}]
    now = time.time()
    for g in grants:
        # 字段顺序与 jit_grantd 的 SELECT 一致
        gid, subject, resource, _verbs, expire_at, _sev = g
        if expire_at + GRACE_SECONDS < now:
            stale.append({
                "source": "jit", "grant_id": gid, "subject": subject,
                "resource": resource,
                "expired_for_seconds": int(now - expire_at),
            })
    return stale


def stale_from_k8s():
    """扫描带 sre.wang/expire-at 注解但已过期的 RoleBinding。"""
    stale = []
    cmd = [
        "kubectl", "get", "rolebindings", "--all-namespaces", "-o", "json",
    ]
    try:
        out = subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=30).stdout
    except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as exc:
        return [{"source": "k8s", "error": str(exc)}]
    now = int(time.time())
    for item in json.loads(out).get("items", []):
        ann = (item.get("metadata", {}).get("annotations") or {})
        expire_at = ann.get("sre.wang/expire-at")
        if not expire_at:
            continue
        try:
            if int(expire_at) + GRACE_SECONDS < now:
                stale.append({
                    "source": "k8s",
                    "binding": f'{item["metadata"]["namespace"]}/{item["metadata"]["name"]}',
                    "issued_for": ann.get("sre.wang/issued-for", "unknown"),
                    "expired_for_seconds": now - int(expire_at),
                })
        except (TypeError, ValueError):
            continue
    return stale


def main():
    findings = []
    for fn in (stale_from_jit, stale_from_k8s):
        findings.extend(fn())

    ts = datetime.now(timezone.utc).isoformat()
    print(f"[{ts}] 超期权限巡检，发现 {len(findings)} 项")
    for f in findings:
        print("  -", json.dumps(f, ensure_ascii=False))

    if findings:
        # 这里对接告警：可以推 webhook、发值班群、或写入日报
        print("存在超期权限，请立即处理：先手工 revoke，再排查自动回收为何失效。")
        return 1
    print("无超期权限。")
    return 0


if __name__ == "__main__":
    sys.exit(main())