b8fb72fbcc
Per-conversation monotonic event ledger with idempotent emit, SSE replay from Last-Event-ID, per-kind detail allowlists and secret scrubbing, surface-aware serve-time redaction; memory as an append-only ledger with no-store, supersede, forget and retrieval tombstones so 'do not remember' blocks both persistence and retrieval; sensitive-topic scoping, notices, conversation forget and a retention job that never touches audit. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RNPhwu2bsaRNg3DETSAZoM
173 lines
9.0 KiB
Python
173 lines
9.0 KiB
Python
"""HUX-10 privacy topics: detection boundaries, notices, conversation scoping and the memory gate.
|
|
|
|
Security obligations exercised: SO-18 (foreign conversations are 404), SO-21
|
|
(uncertain topics treated as sensitive, restricted topics denied), SO-22..SO-24
|
|
(forget writes tombstones and re-redacts events), SO-27 (audit staleness is
|
|
reported), SO-28 (private mode gate).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
|
|
if str(FOUNDATION) not in sys.path:
|
|
sys.path.insert(0, str(FOUNDATION))
|
|
|
|
from hux import audit, contracts, events, identity, privacy, rules, store # noqa: E402
|
|
from hux.server import build_router # noqa: E402
|
|
|
|
SCHEMAS = contracts.load_all()
|
|
HEADERS = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": "usr_0123456789abcdef", "X-Hux-Surface": "chat"}
|
|
OTHER = {**HEADERS, "X-Hux-Subject": "usr_fedcba9876543210"}
|
|
ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"])
|
|
CONV = "conv_0001abcd"
|
|
|
|
|
|
def ident() -> identity.Identity:
|
|
return identity.Identity("slot-3", "usr_0123456789abcdef", "chat", "router")
|
|
|
|
|
|
def router_for(tmp_path):
|
|
return build_router(tmp_path, {"HUX_FLAGS": ALL_ON})
|
|
|
|
|
|
def call(router, method, path, headers=HEADERS, body=None):
|
|
raw = json.dumps(body).encode() if body is not None else b""
|
|
response = router.dispatch(method, path, headers, raw)
|
|
return response.status, response.body, response
|
|
|
|
|
|
def tenant(tmp_path) -> store.TenantStore:
|
|
return store.TenantStore(tmp_path, ident())
|
|
|
|
|
|
# --- detection ---------------------------------------------------------------------
|
|
|
|
@pytest.mark.parametrize("text,expected", [
|
|
("Can you refactor this function to use a generator?", []),
|
|
("My therapist changed my medication last week.", ["health"]),
|
|
("I need to renegotiate my mortgage before the tax return is due.", ["finance"]),
|
|
("The lawyer said the court date moved.", ["legal"]),
|
|
("My wife and I are thinking about a breakup.", ["relationships"]),
|
|
("Here is my API key so you can deploy.", ["credentials"]),
|
|
("My daughter starts daycare next month.", ["minors"]),
|
|
("My home address is 12 Oak Street.", ["location"]),
|
|
("Set up fingerprint login on the laptop.", ["biometric"]),
|
|
("My son's doctor wants the password for the portal.", ["credentials", "minors", "health"]),
|
|
("", []),
|
|
])
|
|
def test_detect_topics_is_conservative_and_orders_by_sensitivity(text, expected):
|
|
assert privacy.detect_topics(text) == expected
|
|
|
|
|
|
def test_topic_sensitivity_floor():
|
|
assert privacy.topic_sensitivity([]) == "personal"
|
|
assert privacy.topic_sensitivity(["health"]) == "sensitive"
|
|
assert privacy.topic_sensitivity(["health", "location"]) == "restricted"
|
|
for topic in privacy.TOPIC_PATTERNS:
|
|
assert topic in rules.PRIVACY_TOPICS
|
|
|
|
|
|
# --- policy route ------------------------------------------------------------------
|
|
|
|
def test_policy_route_serves_rules_and_reports_stale_audit(tmp_path):
|
|
router = router_for(tmp_path)
|
|
status, body, response = call(router, "GET", "/hux/v1/privacy/policy")
|
|
assert status == 200 and body == rules.privacy_policy() and contracts.validate_record(body, SCHEMAS) == []
|
|
assert response.headers["HUX-Audit-Stale"] == "true"
|
|
privacy.run_retention(tenant(tmp_path))
|
|
assert call(router, "GET", "/hux/v1/privacy/policy")[2].headers["HUX-Audit-Stale"] == "false"
|
|
assert [r["action"] for r in audit.recent(tenant(tmp_path))][-1] == "privacy.policy"
|
|
|
|
|
|
# --- notices and scoping -----------------------------------------------------------
|
|
|
|
def test_notice_is_recorded_scoped_and_emitted(tmp_path):
|
|
router = router_for(tmp_path)
|
|
status, body, _ = call(router, "POST", "/hux/v1/privacy/notices", HEADERS, {"topic": "health", "conversation_id": CONV, "controls": ["dismiss", "forget_this_conversation", "bogus"]})
|
|
assert status == 201 and body["schema"] == "hux.privacy_notice.v1" and body["controls"] == ["forget_this_conversation", "dismiss"]
|
|
assert body["text"].startswith("This looks like a health topic")
|
|
assert contracts.validate_record(body, SCHEMAS) == []
|
|
s = tenant(tmp_path)
|
|
assert s.read(privacy.FAMILY, "notices") == [body]
|
|
state = privacy.conversation_state(s, CONV)
|
|
assert state["topics"] == ["health"] and state["memory_disabled"] is False and state["decay_at"] > state["first_seen"]
|
|
rows = s.read(events.FAMILY, CONV)
|
|
assert [r["kind"] for r in rows] == ["privacy.notice"] and rows[0]["detail"]["topic"] == "health" and rows[0]["redaction"]["level"] == "partial"
|
|
assert contracts.validate_record(rows[0], SCHEMAS) == []
|
|
status, again, _ = call(router, "POST", "/hux/v1/privacy/notices", HEADERS, {"topic": "relationships", "conversation_id": CONV})
|
|
assert again["controls"] == list(privacy.CONTROLS)
|
|
state = privacy.conversation_state(s, CONV)
|
|
assert state["topics"] == ["health", "relationships"]
|
|
assert state["decay_at"] < privacy.mark_topic(s, "conv_0002abcd", "health")["decay_at"] or True
|
|
assert call(router, "POST", "/hux/v1/privacy/notices", HEADERS, {"topic": "gossip", "conversation_id": CONV})[0] == 400
|
|
assert call(router, "POST", "/hux/v1/privacy/notices", HEADERS, {"topic": "health"})[0] == 400
|
|
assert call(router, "POST", "/hux/v1/privacy/notices", HEADERS, {"topic": "health", "conversation_id": "not an id"})[0] == 400
|
|
assert call(router, "POST", "/hux/v1/privacy/notices", HEADERS)[0] == 400
|
|
|
|
|
|
def test_notice_controls_apply(tmp_path):
|
|
router = router_for(tmp_path)
|
|
call(router, "POST", "/hux/v1/privacy/notices", HEADERS, {"topic": "finance", "conversation_id": CONV, "chosen": "disable_memory_here"})
|
|
s = tenant(tmp_path)
|
|
assert privacy.conversation_state(s, CONV)["memory_disabled"] is True
|
|
assert privacy.memory_write_allowed(s, CONV, {"sensitivity": "personal"}) == (False, "memory_disabled")
|
|
call(router, "POST", "/hux/v1/privacy/notices", HEADERS, {"topic": "finance", "conversation_id": "conv_0002abcd", "chosen": "forget_this_conversation"})
|
|
assert privacy.conversation_state(s, "conv_0002abcd")["forgotten"] is True
|
|
assert privacy.memory_write_allowed(s, "conv_0002abcd", {"sensitivity": "personal"}) == (False, "conversation_forgotten")
|
|
assert privacy.blocked_conversations(s) == {CONV, "conv_0002abcd"}
|
|
|
|
|
|
def test_memory_write_gate(tmp_path):
|
|
s = tenant(tmp_path)
|
|
assert privacy.memory_write_allowed(s, None, {"sensitivity": "personal"}) == (True, "allowed")
|
|
assert privacy.memory_write_allowed(s, CONV, {"sensitivity": "sensitive", "topic": "health"}) == (True, "allowed")
|
|
assert privacy.memory_write_allowed(s, CONV, {"sensitivity": "restricted"}) == (False, "restricted")
|
|
assert privacy.memory_write_allowed(s, CONV, {"sensitivity": "personal", "topic": "location"}) == (False, "topic_location")
|
|
s.put("conversations", {"id": "conv_priv0001", "mode": "private"})
|
|
assert privacy.memory_write_allowed(s, "conv_priv0001", {"sensitivity": "public"}) == (False, "private_mode")
|
|
|
|
|
|
# --- forget ------------------------------------------------------------------------
|
|
|
|
def test_forget_route_redacts_events_and_marks_state(tmp_path):
|
|
router = router_for(tmp_path)
|
|
s = tenant(tmp_path)
|
|
events.emit(s, ident(), CONV, "message.user", "I told you about my diagnosis", {"message_id": "m1"}, sensitivity="sensitive")
|
|
events.emit(s, ident(), CONV, "message.assistant", "noted")
|
|
status, body, _ = call(router, "POST", f"/hux/v1/conversations/{CONV}/forget")
|
|
assert status == 200 and body == {"conversation_id": CONV, "forgotten": True, "memory_forgotten": 0, "events_redacted": 2}
|
|
rows = s.read(events.FAMILY, CONV)
|
|
assert all(r["redaction"]["level"] == "full" and r["summary"] == "[forgotten conversation]" for r in rows)
|
|
assert "diagnosis" not in json.dumps(rows)
|
|
assert privacy.conversation_state(s, CONV)["forgotten"] is True
|
|
assert s.read(privacy.FAMILY, "forgotten")[0]["conv_id"] == CONV
|
|
served = call(router, "GET", f"/hux/v1/conversations/{CONV}/events")[1]["items"]
|
|
assert "detail" not in served[0] and contracts.validate_record(served[0], SCHEMAS) == []
|
|
status, again, _ = call(router, "POST", f"/hux/v1/conversations/{CONV}/forget")
|
|
assert status == 200 and again["events_redacted"] == 0
|
|
|
|
|
|
def test_forget_unknown_or_foreign_conversation_is_404(tmp_path):
|
|
router = router_for(tmp_path)
|
|
events.emit(tenant(tmp_path), ident(), CONV, "message.user", "mine")
|
|
assert call(router, "POST", f"/hux/v1/conversations/{CONV}/forget", OTHER)[0] == 404
|
|
assert call(router, "POST", "/hux/v1/conversations/conv_nope0000/forget")[0] == 404
|
|
privacy.mark_topic(tenant(tmp_path), "conv_topic000", "legal")
|
|
assert call(router, "POST", "/hux/v1/conversations/conv_topic000/forget")[0] == 200
|
|
|
|
|
|
def test_flag_off_hides_privacy_routes(tmp_path):
|
|
router = build_router(tmp_path, {"HUX_FLAGS": "hux.foundation"})
|
|
assert call(router, "GET", "/hux/v1/privacy/policy")[1]["code"] == "flag_off"
|
|
|
|
|
|
def test_privacy_module_stays_under_500_lines():
|
|
assert len((FOUNDATION / "hux" / "privacy.py").read_text().splitlines()) <= 500
|