From a75ca29934f64949d3a13af59b883811f5e028e9 Mon Sep 17 00:00:00 2001 From: jenkins Date: Mon, 24 Aug 2026 00:48:19 -0300 Subject: [PATCH] hermes(hux): force restricted sensitivity on secret-bearing artifact content User artifacts are never rewritten, but a secret pattern in a version marks the artifact restricted and audits the reason (review a2-C, SO-12). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01RNPhwu2bsaRNg3DETSAZoM --- .../hermes-hux-foundation/hux/artifacts.py | 21 +++++++++++++++++-- .../tests/test_hermes_hux_artifact_auth.py | 16 ++++++++++++++ 2 files changed, 35 insertions(+), 2 deletions(-) diff --git a/dockerfiles/hermes-hux-foundation/hux/artifacts.py b/dockerfiles/hermes-hux-foundation/hux/artifacts.py index cb198d4c..38a4f4e9 100644 --- a/dockerfiles/hermes-hux-foundation/hux/artifacts.py +++ b/dockerfiles/hermes-hux-foundation/hux/artifacts.py @@ -130,6 +130,17 @@ def _string(body: dict[str, Any], key: str, limit: int, required: bool = True) - return value +def secret_hits(data: bytes) -> list[str]: + """Secret-pattern classes found in text content; binary content is not scanned (SO-12).""" + from hux.redaction import scrub_text + + try: + text = data.decode("utf-8") + except UnicodeDecodeError: + return [] + return scrub_text(text)[1] + + def _content(body: dict[str, Any]) -> tuple[bytes, str]: """Decode the submitted content and return ``(bytes, mime)``.""" text, encoded = body.get("content"), body.get("content_base64") @@ -243,6 +254,9 @@ def create(request: Request) -> Response: sensitivity = body.get("sensitivity", "personal") if sensitivity not in SENSITIVITIES: raise Invalid("unknown sensitivity") + hits = secret_hits(_content(body)[0]) + if hits: + sensitivity = "restricted" stamp = now_iso() version: dict[str, Any] = {"version": 1, "created_at": stamp, "created_by": _actor(request.identity), "content_ref": _store_content(request.store, body)} for field, limit in (("message_id", 120), ("note", 200)): @@ -273,7 +287,7 @@ def create(request: Request) -> Response: _check(artifact) stored = request.store.put(FAMILY, artifact) remember(request.store, FAMILY, key, stored["id"]) - request.audit("artifacts.create", stored["id"]) + request.audit("artifacts.create", stored["id"], reason="secret_pattern_restricted" if hits else "") emit_event(request.store, request.identity, stored, "artifact.created", f"Created {stored['type']} artifact", {"artifact_id": stored["id"], "version": 1}) return Response(201, stored, {"ETag": str(stored["revision"])}) @@ -318,6 +332,9 @@ def add_version(request: Request) -> Response: if expected != artifact["revision"]: raise Conflict(f"revision {expected} does not match current revision {artifact['revision']}", [str(artifact["revision"])]) number = artifact["current_version"] + 1 + hits = secret_hits(_content(body)[0]) + if hits: + artifact = {**artifact, "sensitivity": "restricted"} entry: dict[str, Any] = {"version": number, "created_at": now_iso(), "created_by": _actor(request.identity), "content_ref": _store_content(request.store, body), "diff_from": artifact["current_version"]} if body.get("diff_from") is not None: entry["diff_from"] = _version(artifact, body["diff_from"])["version"] @@ -329,7 +346,7 @@ def add_version(request: Request) -> Response: entry["lineage"] = lineage stored = append_version(request.store, artifact, entry, expected) remember(request.store, FAMILY, key, f"{stored['id']}@{number}") - request.audit("artifacts.version", f"{stored['id']}@{number}") + request.audit("artifacts.version", f"{stored['id']}@{number}", reason="secret_pattern_restricted" if hits else "") emit_event(request.store, request.identity, stored, "artifact.version", f"New version {number}", {"artifact_id": stored["id"], "version": number}) return Response(201, stored, {"ETag": str(stored["revision"])}) diff --git a/testing/tests/test_hermes_hux_artifact_auth.py b/testing/tests/test_hermes_hux_artifact_auth.py index 96c8e373..c38b1602 100644 --- a/testing/tests/test_hermes_hux_artifact_auth.py +++ b/testing/tests/test_hermes_hux_artifact_auth.py @@ -128,3 +128,19 @@ def test_audit_rows_name_family_verbs(router, artifact): rows = audit.recent(store.TenantStore(router.data_root, identity.Identity("slot-3", "usr_0123456789abcdef", "chat", "router"))) assert [r["action"] for r in rows] == ["artifacts.create", "artifacts.get"] assert all(contracts.validate("common.schema.json", r, SCHEMAS, "/$defs/audit_outcome") == [] for r in rows) + + +def test_secret_bearing_content_is_kept_but_forced_restricted(router): + """Review a2-C / SO-12: user artifacts are not rewritten, but a secret pattern forces restricted sensitivity and is audited.""" + token = "ghp_" + "A" * 36 + status, record = call(router, "POST", "/hux/v1/artifacts", {"type": "code", "title": "env", "content": f"TOKEN={token}", "sensitivity": "personal"}) + assert status == 201 and record["sensitivity"] == "restricted" + status, version = call(router, "GET", f"/hux/v1/artifacts/{record['id']}/versions/1") + assert status == 200 and token in version["content"] + clean = call(router, "POST", "/hux/v1/artifacts", {"type": "code", "title": "clean", "content": "x = 1"})[1] + assert clean["sensitivity"] == "personal" + status, bumped = call(router, "POST", f"/hux/v1/artifacts/{clean['id']}/versions", {"content": f"key: {token}"}, {**OWNER, "If-Match": "1"}) + assert status == 201 and bumped["sensitivity"] == "restricted" + from hux import audit, store + reasons = [r.get("reason") for r in audit.recent(store.TenantStore(router.data_root, identity.resolve(OWNER, {}))) if r["action"].startswith("artifacts.")] + assert reasons.count("secret_pattern_restricted") == 2