# -*- coding: utf-8 -*-
"""签名接口层（蜂群互联 D-4）。

设计裁决 D-4（2026-09-10，设计者）：

  · 协议内核**不实现密码学**，只定义 `sign` / `verify` 契约；
  · 具体签名链（企业 CA / KMS / 签章服务）在**工程项目**中实现；
  · **搭载灵枢的智能体自行决定子系统如何使用签名链**（策略落盘 `_signers.json`）。

因此本模块只做三件事：
  ① 定义可插拔的 `Signer` 契约 + 注册表；
  ② 提供两个最简内置实现（`null` / `hmac-local`）；
  ③ 读取并执行**子系统签名策略**（何时签、是否要求对端签、验签失败如何处置）。

**内核不保存私钥**：私钥归工程侧，这里只持有 `sign` / `verify` 的可调用引用。
工程侧注入方式：

    # 1) 代码注入
    from md_cg.signer import register_signer
    register_signer("corp_pki", lambda: MyCorpSigner(...))

    # 2) 启动自动加载（MCP 子进程推荐）
    MDCG_SIGNER_MODULE=mycorp.signing   # 模块内提供 register()

零第三方依赖。
"""
from __future__ import annotations

import argparse
import base64
import hashlib
import hmac
import importlib
import json
import os
import secrets
import sys
import time

from .datapath import aux_root
from .fsutil import publish

SIGNER_ENV = "MDCG_SIGNER"
SIGNER_MODULE_ENV = "MDCG_SIGNER_MODULE"
SIGNERS_FILE_ENV = "MDCG_SIGNERS_FILE"
KEY_FILE_ENV = "MDCG_SIGNER_KEY_FILE"

DEFAULT_DIR = aux_root()
DEFAULT_SIGNERS_FILE = os.path.join(DEFAULT_DIR, "_signers.json")
DEFAULT_KEY_FILE = os.path.join(DEFAULT_DIR, "_signer.key")

SCHEMA = 1
DEFAULT_SIGNER = "hmac-local"
PREFIX = "hmac1"

# 需要签名的动作（子系统策略里引用）
SIGN_ACTIONS = ("handshake", "evidence", "withdrawal")
# 验签失败的处置档位（宪章第七条降级为默认）
ON_VERIFY_FAIL = ("degrade", "isolate", "reject")


class SignerError(Exception):
    """签名接口层错误。"""


# --------------------------------------------------------------------------
# 契约与内置实现
# --------------------------------------------------------------------------

# 生效条件：实现方提供可调用的 sign(payload, ctx) 与 verify(payload, signature, ctx) 即满足鸭子类型契约；name 缺省为 'abstract'，基类实现直接抛 NotImplementedError。
class Signer:
    """签名器契约。工程侧实现只需满足这两个方法（鸭子类型即可）。"""

    name = "abstract"

# 生效条件：基类实现，任意 payload 下直接 raise NotImplementedError，由子类覆写。
    def sign(self, payload: bytes, ctx: dict = None) -> str:
        raise NotImplementedError

# 生效条件：基类实现，任意 payload 与 signature 下直接 raise NotImplementedError，由子类覆写。
    def verify(self, payload: bytes, signature: str, ctx: dict = None) -> bool:
        raise NotImplementedError

# 生效条件：无前置；返回 {"name": self.name, "kind": "abstract"}，不含密钥材料，供注册表自描述用。
    def describe(self) -> dict:
        return {"name": self.name, "kind": "abstract"}

# 生效条件：无 required 形参，调用即返回仅含 {'name': self.name} 的 dict，不含密钥材料。
    def public(self) -> dict:
        """可对外公开的验证信息（**不得含密钥**）。"""
        return {"name": self.name}


# 生效条件：sign 对任意 payload 恒返回 ''，verify 仅当 signature 为 '' 或 None 时返回 True（name 为 'null'）。
class NullSigner(Signer):
    """不签名（观察期 / 纯本地）。验签时把「空签名」视为通过。"""

    name = "null"

# 生效条件：任意 payload 下均返回空串 ""，不做任何签名计算。
    def sign(self, payload: bytes, ctx: dict = None) -> str:
        return ""

# 生效条件：signature 恰为 "" 或 None 时返回 True，其余取值（含仅含空白的字符串）返回 False，与 payload 无关。
    def verify(self, payload: bytes, signature: str, ctx: dict = None) -> bool:
        return signature in ("", None)

# 生效条件：无前置；返回 {"name", "kind": "none", "note"}，其中 note 明示「不签名；仅用于观察期或纯本地场景」，不含密钥材料。
    def describe(self) -> dict:
        return {"name": self.name, "kind": "none",
                "note": "不签名；仅用于观察期或纯本地场景"}


# 生效条件：__init__ 无必需形参；key_file 为假值（None/空串）时回落 key_file_path()，key 为假值（None/空串）时回落 load_key(self.key_file)。
class HmacLocalSigner(Signer):
    """本地 HMAC-SHA256 实现——与 `_tokens.json` 同一信任根级别的最简方案。

    它**不是**企业级签名链（无不可否认性、无私钥托管），
    仅用于单机自测与「有签名位置但暂无 CA」的过渡期。
    """

    name = "hmac-local"

# 生效条件：key_file 为假值（None/空串）时 self.key_file 回落 key_file_path()；key 为假值（None/空字节串）时 self.key 回落 load_key(self.key_file)。
    def __init__(self, key: bytes = None, key_file: str = None):
        self.key_file = key_file or key_file_path()
        self.key = key or load_key(self.key_file)

# 生效条件：对任意 payload（ctx 为 None 时一并传入 canonical）以 self.key 做 HMAC-SHA256，返回 f"{PREFIX}.{mac.hexdigest()}"。
    def sign(self, payload: bytes, ctx: dict = None) -> str:
        mac = hmac.new(self.key, canonical(payload, ctx), hashlib.sha256)
        return f"{PREFIX}.{mac.hexdigest()}"

# 生效条件：signature 为假值时按 "" 处理并 strip，不以 PREFIX+"." 开头立即返回 False；否则与 self.sign(payload, ctx) 的结果做 hmac.compare_digest 并返回该布尔值。
    def verify(self, payload: bytes, signature: str, ctx: dict = None) -> bool:
        sig = (signature or "").strip()
        if not sig.startswith(PREFIX + "."):
            return False
        want = self.sign(payload, ctx)
        return hmac.compare_digest(want, sig)

# 生效条件：无前置；返回 {"name", "kind": "hmac-sha256", "key_file", "note"}——含 key_file 路径但不含密钥材料（密钥本身只经 public() 的 key_id 暴露）。
    def describe(self) -> dict:
        return {"name": self.name, "kind": "hmac-sha256",
                "key_file": self.key_file,
                "note": "本地实现，非企业级签名链；无不可否认性"}

# 生效条件：无必需形参，调用即返回 {"name": self.name, "kind": "hmac-sha256", "key_id": hashlib.sha256(self.key).hexdigest()[:16]}。
    def public(self) -> dict:
        return {"name": self.name, "kind": "hmac-sha256",
                "key_id": hashlib.sha256(self.key).hexdigest()[:16]}


_REGISTRY = {
    "null": NullSigner,
    DEFAULT_SIGNER: HmacLocalSigner,
}
_LOADED_MODULES = set()


# 生效条件：payload 为 str 时先按 utf-8 编码，其 base64 文本构成 body['payload']；ctx 非空时以 str 化并按 key 排序后并入 body，最终返回 json.dumps(body, ensure_ascii=False, sort_keys=True, separators=(',', ':')).encode('utf-8')。
def canonical(payload, ctx: dict = None) -> bytes:
    """规范化待签字节串：payload 走 base64，ctx 按 key 排序——避免表示歧义。"""
    if isinstance(payload, str):
        payload = payload.encode("utf-8")
    body = {"payload": base64.b64encode(bytes(payload)).decode("ascii")}
    if ctx:
        body["ctx"] = {str(k): str(ctx[k]) for k in sorted(ctx)}
    return json.dumps(body, ensure_ascii=False, sort_keys=True,
                      separators=(",", ":")).encode("utf-8")


# --------------------------------------------------------------------------
# 注册表（工程侧注入点）
# --------------------------------------------------------------------------

# 生效条件：name 去空白后非空且 factory 可调用时登记进 _REGISTRY 并返回 {'ok': True, 'name': name, 'signers': sorted(_REGISTRY)}，否则抛 SignerError。
def register_signer(name: str, factory) -> dict:
    """注册签名器。`factory` 为无参可调用（返回实例）或类本身。"""
    name = (name or "").strip()
    if not name:
        raise SignerError("签名器名不能为空")
    if not callable(factory):
        raise SignerError("factory 必须可调用（返回 Signer 实例）")
    _REGISTRY[name] = factory
    return {"ok": True, "name": name, "signers": sorted(_REGISTRY)}


# 生效条件：name 已在 _REGISTRY 且不是 'null' 或 DEFAULT_SIGNER 时从注册表移除并返回 {'ok': True, 'signers': sorted(_REGISTRY)}，否则抛 SignerError。
def unregister_signer(name: str) -> dict:
    if name not in _REGISTRY:
        raise SignerError(f"签名器不存在：{name}")
    if name in ("null", DEFAULT_SIGNER):
        raise SignerError(f"内置签名器不可注销：{name}")
    _REGISTRY.pop(name, None)
    return {"ok": True, "signers": sorted(_REGISTRY)}


# 生效条件：无 required 形参，调用即返回 {'default': default_name(), 'signers': sorted(_REGISTRY)}。
def list_signers() -> dict:
    return {"default": default_name(), "signers": sorted(_REGISTRY)}


# 生效条件：环境变量 SIGNER_ENV 取值去空白后非空则返回该值，否则返回 DEFAULT_SIGNER。
def default_name() -> str:
    return (os.environ.get(SIGNER_ENV) or "").strip() or DEFAULT_SIGNER


# 生效条件：环境变量 SIGNER_MODULE_ENV 给出的模块名非空且未加载过时导入该模块并调用其无参 register()，模块加载失败或缺 register() 可调用则抛 SignerError。
def _autoload():
    """按 MDCG_SIGNER_MODULE 自动加载工程侧注册模块（进程内只做一次）。"""
    mod = (os.environ.get(SIGNER_MODULE_ENV) or "").strip()
    if not mod or mod in _LOADED_MODULES:
        return
    _LOADED_MODULES.add(mod)
    try:
        m = importlib.import_module(mod)
    except Exception as e:                     # 加载失败不静默：留给调用方判定
        raise SignerError(f"签名器模块加载失败 {mod}：{e}")
    reg = getattr(m, "register", None)
    if not callable(reg):
        raise SignerError(f"签名器模块 {mod} 未提供 register()")
    reg()


# 生效条件：name 去空白后非空（为空取 default_name()）且该名已在 _REGISTRY 时构造实例，实例须具备可调用的 sign/verify 否则抛 SignerError；factory 形参含 key_file 时以 key_file 传入构造。
def get_signer(name: str = None, key_file: str = None) -> Signer:
    """取签名器实例。未注册即抛错（fail-closed，不静默降级为 null）。"""
    _autoload()
    nm = (name or "").strip() or default_name()
    factory = _REGISTRY.get(nm)
    if factory is None:
        raise SignerError(
            f"签名器未注册：{nm}（已注册：{sorted(_REGISTRY)}）；"
            "工程侧可用 register_signer() 或 MDCG_SIGNER_MODULE 注入")
    try:
        inst = factory(key_file=key_file) if _accepts_key_file(factory) else factory()
    except TypeError:
        inst = factory()
    if not isinstance(inst, Signer) and not (
            callable(getattr(inst, "sign", None))
            and callable(getattr(inst, "verify", None))):
        raise SignerError(f"签名器 {nm} 未实现 sign/verify 契约")
    if not getattr(inst, "name", None):
        inst.name = nm
    return inst


# 生效条件：inspect.signature(factory) 的形参名中含 'key_file' 时返回 True，否则（含 TypeError/ValueError）返回 False。
def _accepts_key_file(factory) -> bool:
    try:
        import inspect
        return "key_file" in inspect.signature(factory).parameters
    except (TypeError, ValueError):
        return False


# --------------------------------------------------------------------------
# 密钥（仅内置 hmac-local 需要；工程侧自行管理私钥）
# --------------------------------------------------------------------------

# 生效条件：显式 path 非空即返回该 path，否则返回环境变量 KEY_FILE_ENV 的值，两者皆空返回 DEFAULT_KEY_FILE。
def key_file_path(path: str = None) -> str:
    return path or os.environ.get(KEY_FILE_ENV) or DEFAULT_KEY_FILE


# 生效条件：key_file_path(path) 处可读出非空内容时返回该内容 bytes，否则生成 secrets.token_urlsafe(32) 密钥、经 save_key 落盘后返回该密钥。
def load_key(path: str = None) -> bytes:
    p = key_file_path(path)
    if os.path.exists(p):
        with open(p, "rb") as f:
            raw = f.read().strip()
        if raw:
            return raw
    key = secrets.token_urlsafe(32).encode("utf-8")
    save_key(key, p)
    return key


# 生效条件：key 为 bytes 时原样写、否则 str(key).encode('utf-8')，写入 key_file_path(path) 同目录的临时文件后 publish（带 Windows 短重试的 os.replace）覆盖到该路径并返回它。
def save_key(key: bytes, path: str = None) -> str:
    p = key_file_path(path)
    os.makedirs(os.path.dirname(p) or ".", exist_ok=True)
    tmp = p + ".tmp"
    with open(tmp, "wb") as f:
        f.write(key if isinstance(key, bytes) else str(key).encode("utf-8"))
    try:
        os.chmod(tmp, 0o600)
    except OSError:
        pass
    publish(tmp, p)
    return p


# --------------------------------------------------------------------------
# 子系统签名策略（D-4：智能体自行决定子系统怎么用签名链）
# --------------------------------------------------------------------------

# 生效条件：无参无外部依赖；返回默认策略 {"signer": None, "sign_on": ["handshake"], "require_peer_signature": False, "on_verify_fail": "degrade"}（调用方需经 policy() 归一化后再用）；
def _default_policy() -> dict:
    return {"signer": None, "sign_on": ["handshake"],
            "require_peer_signature": False, "on_verify_fail": "degrade"}


# 生效条件：pol 中 'signer'/'sign_on'/'require_peer_signature'/'on_verify_fail' 键覆盖默认策略，sign_on 只保留属于 SIGN_ACTIONS 的项，on_verify_fail 不在 ON_VERIFY_FAIL 时回落 'degrade'。
def _normalize(pol: dict) -> dict:
    out = _default_policy()
    out.update({k: pol[k] for k in ("signer", "sign_on",
                                    "require_peer_signature",
                                    "on_verify_fail") if k in pol})
    on = out.get("sign_on")
    if isinstance(on, str):
        on = [x for x in on.split(",") if x.strip()]
    out["sign_on"] = [x for x in (on or []) if x in SIGN_ACTIONS]
    if out.get("on_verify_fail") not in ON_VERIFY_FAIL:
        out["on_verify_fail"] = "degrade"
    out["require_peer_signature"] = bool(out.get("require_peer_signature"))
    return out


# 生效条件：path 为假值时返回环境变量 SIGNERS_FILE_ENV 或 DEFAULT_SIGNERS_FILE（受信面单点，不设闸）；path 非空时归一为绝对路径 norm，须以 .json 结尾（格式闸）且 realpath(norm) 位于受信目录 realpath(dirname(已配置路径)) 之下（os.path.commonpath 判定；跨盘符的 ValueError 按越界处理），越界即 raise ValueError（消息含两侧路径与 #279），全部通过时返回 norm（不返回原串）。
def signers_file(path: str = None) -> str:
    """策略文件路径解析——issue #279 路径闸（realpath 受信目录）。

    判据：显式 path 先归一（abspath）为 norm，再要求
    realpath(norm) 落在受信目录 = realpath(dirname(环境变量或缺省的
    已配置路径)) 之下（os.path.commonpath 判定；跨盘符触发的
    commonpath ValueError 按越界处理——不同盘必然越界）。

    `.json` 后缀只是**格式闸**；路径闸由 realpath+commonpath 承担——
    后缀挡不住 `<受信>/ok.json/../../evil.json` 穿越与 symlink/junction
    指向任意目标，realpath+commonpath 能挡。返回 **norm**（归一化后的
    受信形态），不返回原串——旧实现算了 norm 却把原串返回，校验成了
    对控制流零影响的死代码（值守 #316/#279 审阅结论）。

    env 与缺省路径是运维侧受信面（MCP 参数改不动进程环境），
    不设闸、逐位返回原值。
    """
    configured = os.environ.get(SIGNERS_FILE_ENV) or DEFAULT_SIGNERS_FILE
    if not path:
        return configured
    norm = os.path.abspath(str(path))
    if not norm.endswith(".json"):
        raise ValueError(
            f"#279 signers_file 只接受 .json 策略文件（实测 {norm!r}）"
            "——后缀只是格式闸，不得经参数读写任意文件")
    real = os.path.realpath(norm)
    trusted = os.path.realpath(os.path.dirname(os.path.abspath(configured)))
    try:
        inside = (os.path.commonpath([os.path.normcase(real),
                                      os.path.normcase(trusted)])
                  == os.path.normcase(trusted))
    except ValueError:
        inside = False  # 跨盘符（Windows 不同盘）/ UNC 混形：必然越界
    if not inside:
        raise ValueError(
            f"#279 signers_file 路径越界：{norm!r}（realpath → {real!r}）"
            f" 不在受信目录 {trusted!r}"
            f"（dirname(已配置路径 {configured!r})）之下"
            "——不得经参数读写受信目录外的文件")
    return norm


# 生效条件：signers_file(path) 处可读出 dict 时补上缺省 schema/subsystems 后返回该 dict，否则返回 {'schema': SCHEMA, 'default_signer': DEFAULT_SIGNER, 'subsystems': {}, 'updated_at': None}。
def load_policies(path: str = None) -> dict:
    p = signers_file(path)
    if os.path.exists(p):
        try:
            with open(p, "r", encoding="utf-8") as f:
                d = json.load(f)
            if isinstance(d, dict):
                d.setdefault("schema", SCHEMA)
                d.setdefault("subsystems", {})
                return d
        except (OSError, ValueError):
            pass
    return {"schema": SCHEMA, "default_signer": DEFAULT_SIGNER,
            "subsystems": {}, "updated_at": None}


# 生效条件：data 传入任意映射即被 dict(data) 浅拷贝并覆写 schema/updated_at 后写入 signers_file(path)（path 为假值时由 signers_file(path) 决定路径）。
def save_policies(data: dict, path: str = None) -> str:
    p = signers_file(path)
    os.makedirs(os.path.dirname(p) or ".", exist_ok=True)
    data = dict(data)
    data["schema"] = SCHEMA
    data["updated_at"] = time.time()
    tmp = p + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=1, sort_keys=True)
    try:
        os.chmod(tmp, 0o600)
    except OSError:
        pass
    publish(tmp, p)
    return p


# 生效条件：subsystem 为假值（None/空串）时按空串键在 d.get("subsystems") or {} 中查策略，取不到则用默认策略，且 out["signer"] 为假值时回落 d.get("default_signer") or DEFAULT_SIGNER。
def policy_for(subsystem: str = None, path: str = None) -> dict:
    """取子系统策略；未声明则用默认（`default_signer` + handshake 签名）。"""
    d = load_policies(path)
    pol = (d.get("subsystems") or {}).get(subsystem or "") or {}
    out = _normalize(pol)
    if not out.get("signer"):
        out["signer"] = d.get("default_signer") or DEFAULT_SIGNER
    out["subsystem"] = subsystem or ""
    return out


# 生效条件：subsystem 为假值（None/空串）时抛 SignerError("子系统名不能为空")；否则 kw 中非 None 项合并进该子系统策略并落盘，且 kw.get("default_signer") 为真值时同时更新 d["default_signer"]。
_FAIL_RANK = {"degrade": 0, "isolate": 1, "reject": 2}


def weaker_than(pol: dict, base: dict) -> str:
    """pol 相对 base 是否**更松**（issue #167）：更松返回原因串，否则 ''。

    验签要求单调性的**唯一判据单点**（只能更严不能更松），供 import / observe /
    policy set 三处闸共用——三处各写一份判据必然漂开（本仓既有教训）。
    """
    if not isinstance(pol, dict) or not isinstance(base, dict):
        return ""
    if (bool(base.get("require_peer_signature"))
            and not bool(pol.get("require_peer_signature"))):
        return "require_peer_signature 由 True 变 False"
    if (_FAIL_RANK.get(pol.get("on_verify_fail"), 0)
            < _FAIL_RANK.get(base.get("on_verify_fail"), 0)):
        return ("on_verify_fail 由 %s 降为 %s"
                % (base.get("on_verify_fail"), pol.get("on_verify_fail")))
    return ""


def loosening(subsystem: str, path: str = None, **kw) -> str:
    """按 set_policy 的**同一合并口径**预判「新策略相对现行策略是否放宽」。

    放宽返回原因串，否则 ''。更换 signer（换验签实现）按放宽处理（保守侧：要求
    管理权）。本函数**不写盘、无副作用**；`set_policy` 仍是唯一落盘点。
    """
    cur = policy_for(subsystem, path)
    new = {k: v for k, v in kw.items() if v is not None and k != "path"}
    if "signer" in new and new["signer"] != cur.get("signer"):
        return "signer 由 %s 换为 %s" % (cur.get("signer"), new["signer"])
    nxt = dict(cur)
    nxt.update(new)
    return weaker_than(nxt, cur)


def set_policy(subsystem: str, path: str = None, **kw) -> dict:
    """声明/更新子系统签名策略——**这是智能体行使 D-4 的入口**。"""
    if not subsystem:
        raise SignerError("子系统名不能为空")
    d = load_policies(path)
    cur = (d["subsystems"].get(subsystem) or {})
    cur.update({k: v for k, v in kw.items() if v is not None})
    d["subsystems"][subsystem] = _normalize(cur)
    if kw.get("default_signer"):
        d["default_signer"] = kw["default_signer"]
    save_policies(d, path)
    return {"ok": True, "subsystem": subsystem,
            "policy": policy_for(subsystem, path),
            "file": signers_file(path)}


# 生效条件：subsystem 命中 d.get("subsystems") 的键（含空串）时才弹出并 save_policies 落盘；未命中则不写盘，两种情况都返回 ok=True 与 sorted(d.get("subsystems") or {})。
def remove_policy(subsystem: str, path: str = None) -> dict:
    d = load_policies(path)
    if subsystem in (d.get("subsystems") or {}):
        d["subsystems"].pop(subsystem)
        save_policies(d, path)
    return {"ok": True, "subsystem": subsystem,
            "subsystems": sorted(d.get("subsystems") or {})}


# --------------------------------------------------------------------------
# 策略驱动的签名 / 验签（连接层调用）
# --------------------------------------------------------------------------

# 生效条件：action 不在 policy_for(subsystem, path)["sign_on"] 中时返回 signed=False、signer=None、signature=""；在列内时由 get_signer(pol.get("signer")) 对 payload 签名并返回 signed=True 与 s.public()。
def sign_for(subsystem: str, payload, ctx: dict = None,
             action: str = "handshake", path: str = None) -> dict:
    """按子系统策略签名。策略未把该 action 列入 `sign_on` 则**不签**。"""
    pol = policy_for(subsystem, path)
    if action not in pol["sign_on"]:
        return {"ok": True, "signed": False, "action": action,
                "signer": None, "signature": "",
                "reason": f"子系统 {subsystem or '(默认)'} 策略未对 {action} 要求签名"}
    s = get_signer(pol.get("signer"))
    sig = s.sign(payload, ctx)
    return {"ok": True, "signed": True, "action": action,
            "signer": getattr(s, "name", None), "signature": sig,
            "public": s.public()}


# 生效条件：required=bool(pol["require_peer_signature"])，signature 经 (signature or "").strip() 后为空且 required 为假时返回 ok=True/required=False，required 为真且为空时返回 ok=False/missing=True/on_fail="reject"，签名非空时由 get_signer(pol.get("signer")) 验签、异常置 ok=False 且 err=str(e)，失败时 on_fail 取 pol["on_verify_fail"]。
def verify_for(subsystem: str, payload, signature, ctx: dict = None,
               action: str = "handshake", path: str = None) -> dict:
    """按子系统策略验对端签名。

    语义分工（避免歧义）：
      · `sign_on`      —— **本节点**在哪些动作上签名（见 `sign_for`）；
      · `require_peer_signature` —— **对端**是否必须提供签名。

    三种结果：
      · 不要求且未提供 → `ok=True, required=False`（放行）；
      · 要求但缺失     → `ok=False, missing=True, on_fail="reject"`（fail-closed）；
      · 提供了就验     → 失败时 `on_fail` 取策略值，由调用方执行处置。
    """
    pol = policy_for(subsystem, path)
    required = bool(pol["require_peer_signature"])
    sig = (signature or "").strip()

    if not required and not sig:
        return {"ok": True, "required": False, "missing": False,
                "action": action, "signer": None, "on_fail": None,
                "require_peer_signature": False,
                "reason": "策略未要求对端签名，且对端未提供"}
    if required and not sig:
        return {"ok": False, "required": True, "missing": True,
                "action": action, "signer": None, "on_fail": "reject",
                "require_peer_signature": True,
                "reason": "策略要求对端签名，但对端未提供"}

    s = get_signer(pol.get("signer"))
    try:
        ok = bool(s.verify(payload, sig, ctx))
    except Exception as e:
        ok, err = False, str(e)
    else:
        err = ""
    return {"ok": ok, "required": required, "missing": False, "action": action,
            "signer": getattr(s, "name", None),
            "on_fail": None if ok else pol["on_verify_fail"],
            "require_peer_signature": required,
            "reason": "" if ok else (err or "验签失败")}


# --------------------------------------------------------------------------
# 自描述 / CLI
# --------------------------------------------------------------------------

# 生效条件：无必需形参，default_signer 取 d.get("default_signer") or DEFAULT_SIGNER，subsystems 缺失或为假值时按 {} 处理并逐个 _normalize，updated_at 取 d.get("updated_at")。
def show(path: str = None) -> dict:
    d = load_policies(path)
    return {"file": signers_file(path),
            "default_signer": d.get("default_signer") or DEFAULT_SIGNER,
            "subsystems": {k: _normalize(v) for k, v in
                           (d.get("subsystems") or {}).items()},
            "updated_at": d.get("updated_at")}


# 生效条件：无必需形参，返回 SIGN_ACTIONS、ON_VERIFY_FAIL 的列表、list_signers()、_REGISTRY 中名为 "null" 或 DEFAULT_SIGNER 的内置签名器 describe()，以及 SIGNER_ENV/SIGNER_MODULE_ENV/SIGNERS_FILE_ENV/KEY_FILE_ENV 常量。
def catalog() -> dict:
    return {
        "layer": "签名接口（蜂群互联 D-4）",
        "decision": "协议内核只定义契约；实现落工程项目；子系统自行决定用法",
        "actions": list(SIGN_ACTIONS),
        "on_verify_fail": list(ON_VERIFY_FAIL),
        "signers": list_signers(),
        "builtin": {n: _REGISTRY[n]().describe() for n in sorted(_REGISTRY)
                    if n in ("null", DEFAULT_SIGNER)},
        "policy_file": signers_file(),
        "env": {"signer": SIGNER_ENV, "module": SIGNER_MODULE_ENV,
                "policy_file": SIGNERS_FILE_ENV, "key_file": KEY_FILE_ENV},
    }


# 生效条件：obj 为任意对象时以 json.dumps(ensure_ascii=False, indent=1, default=str) 打印（不可序列化值经 default=str 转换），源码无返回值。
def _print(obj):
    print(json.dumps(obj, ensure_ascii=False, indent=1, default=str))


# 生效条件：argv（默认取 sys.argv）经 argparse 解析出必填 cmd（catalog/list/show/set/rm/sign/verify 之一）后分派到对应分支；任一分支抛出 SignerError 时向 stderr 打印并返回 2，否则返回 0。
def main(argv=None):
    ap = argparse.ArgumentParser(
        prog="python -m md_cg.signer",
        description="签名接口层：契约 + 子系统策略（D-4：内核不实现密码学）")
    ap.add_argument("--signers-file", default=None,
                    help="策略文件（默认 ~/.mdcg/_signers.json）")
    ap.add_argument("--key-file", default=None,
                    help="本地密钥文件（默认 ~/.mdcg/_signer.key）")
    sub = ap.add_subparsers(dest="cmd", required=True)

    sub.add_parser("catalog", help="签名接口自描述")
    sub.add_parser("list", help="已注册签名器")
    sub.add_parser("show", help="子系统签名策略")

    p_set = sub.add_parser("set", help="声明子系统签名策略")
    p_set.add_argument("subsystem")
    p_set.add_argument("--signer", default=None)
    p_set.add_argument("--sign-on", default=None, help="逗号分隔：handshake,evidence")
    p_set.add_argument("--require-peer", action="store_true", default=None)
    p_set.add_argument("--on-verify-fail", default=None,
                       choices=list(ON_VERIFY_FAIL))

    p_rm = sub.add_parser("rm", help="删除子系统策略")
    p_rm.add_argument("subsystem")

    p_s = sub.add_parser("sign", help="按策略签名（payload 取自 --payload 或 stdin）")
    p_s.add_argument("--subsystem", default=None)
    p_s.add_argument("--action", default="handshake", choices=list(SIGN_ACTIONS))
    p_s.add_argument("--payload", default=None)

    p_v = sub.add_parser("verify", help="按策略验签")
    p_v.add_argument("--subsystem", default=None)
    p_v.add_argument("--action", default="handshake", choices=list(SIGN_ACTIONS))
    p_v.add_argument("--payload", required=True)
    p_v.add_argument("--signature", required=True)

    a = ap.parse_args(argv)
    try:
        if a.cmd == "catalog":
            _print(catalog())
        elif a.cmd == "list":
            _print(list_signers())
        elif a.cmd == "show":
            _print(show(a.signers_file))
        elif a.cmd == "set":
            on = ([x for x in a.sign_on.split(",") if x.strip()]
                  if a.sign_on else None)
            _print(set_policy(a.subsystem, path=a.signers_file,
                              signer=a.signer, sign_on=on,
                              require_peer_signature=a.require_peer,
                              on_verify_fail=a.on_verify_fail))
        elif a.cmd == "rm":
            _print(remove_policy(a.subsystem, a.signers_file))
        elif a.cmd == "sign":
            payload = a.payload if a.payload is not None else sys.stdin.read()
            _print(sign_for(a.subsystem, payload, action=a.action,
                            path=a.signers_file))
        elif a.cmd == "verify":
            _print(verify_for(a.subsystem, a.payload, a.signature,
                              action=a.action, path=a.signers_file))
    except SignerError as e:
        print(f"[md_cg.signer] {e}", file=sys.stderr)
        return 2
    return 0


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