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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RNPhwu2bsaRNg3DETSAZoM

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"])})
 

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