feat: update plan 16 sec14-26

This commit is contained in:
thanhnv
2026-07-07 15:46:36 +09:00
parent 0c60ed33e9
commit ae4fc7112c
64 changed files with 2231 additions and 116 deletions
@@ -62,6 +62,24 @@ def build_proposals(metrics_rows, drift):
return proposals
def verify_metrics_integrity(path, sig, pub):
"""True only if the metrics file has a valid detached signature (openssl).
ARCH-08: proposals from telemetry that is not integrity-verified are marked
untrusted so a reviewer (and the apply gate) treats them with suspicion."""
if not (path and sig and pub):
return False
if not (os.path.isfile(path) and os.path.isfile(sig) and os.path.isfile(pub)):
return False
try:
r = subprocess.run(
["openssl", "dgst", "-sha256", "-verify", pub, "-signature", sig, path],
capture_output=True,
)
return r.returncode == 0
except Exception:
return False
def cmd_propose(args):
metrics = read_jsonl(args.metrics)
drift = None
@@ -70,8 +88,15 @@ def cmd_propose(args):
drift = json.load(open(args.drift, encoding="utf-8"))
except ValueError:
drift = None
# ARCH-08 telemetry-poisoning defence: tag every proposal with the trust level
# of its source telemetry. Unsigned/unverifiable metrics -> untrusted.
trusted = verify_metrics_integrity(args.metrics, args.metrics_sig, args.metrics_pub)
source_trust = "verified" if trusted else "untrusted"
proposals = build_proposals(metrics, drift)
print(json.dumps({"proposals": proposals, "count": len(proposals)}, ensure_ascii=False, indent=2))
for p in proposals:
p["source_trust"] = source_trust
print(json.dumps({"proposals": proposals, "count": len(proposals),
"source_trust": source_trust}, ensure_ascii=False, indent=2))
return 0
@@ -86,6 +111,19 @@ def cmd_apply(args):
print(f"IMPROVE_DENY UNKNOWN_PROPOSAL {args.id}", file=sys.stderr)
return 1
# ARCH-08: in enforced mode (prod / CASAN_SELFIMPROVE_STRICT=1) refuse to apply a
# proposal derived from unverified telemetry unless explicitly allowed with
# justification. Dev default only tags (backward compatible). Missing tag =
# untrusted (fail-closed).
enforced = (os.environ.get("CASAN_PROFILE") == "prod"
or os.environ.get("CASAN_SELFIMPROVE_STRICT") == "1")
if (proposal.get("source_trust", "untrusted") == "untrusted"
and enforced and not args.allow_untrusted):
print(f"IMPROVE_DENY UNTRUSTED_SOURCE {args.id} (telemetry not integrity-verified; "
f"re-run propose with --metrics-sig/--metrics-pub, or pass --allow-untrusted)",
file=sys.stderr)
return 1
# Proposal != application: applying ALWAYS requires human approval (Plan-04).
if not (args.approval or "").strip():
print(f"IMPROVE_DENY APPROVAL_REQUIRED {args.id}", file=sys.stderr)
@@ -117,10 +155,13 @@ def main() -> int:
pr = sub.add_parser("propose")
pr.add_argument("--metrics", default="")
pr.add_argument("--drift", default="")
pr.add_argument("--metrics-sig", default="")
pr.add_argument("--metrics-pub", default="")
ap_ = sub.add_parser("apply")
ap_.add_argument("--proposals", required=True)
ap_.add_argument("--id", required=True)
ap_.add_argument("--approval", default="")
ap_.add_argument("--allow-untrusted", action="store_true")
args = ap.parse_args()
if args.cmd == "propose":
return cmd_propose(args)