fix: block private context at claim intake

memorymaster/core/security.py

@@ -132,6 +132,42 @@
     )),
 ]
 
+# Claim memory has a stricter contract than user-selected raw source/evidence:
+# local topology and machine-specific absolute paths are not durable memories.
+# Keep these out of _SECRET_PATTERNS so ADR-0006 raw source preservation stays
+# unchanged while every SQLite/Postgres/service/spool claim writer is covered
+# by sanitize_claim_input.
+_CLAIM_ONLY_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = (
+    (
+        "private_ipv4",
+        re.compile(
+            r"\b(?:"
+            r"10\.(?:\d{1,3}\.){2}\d{1,3}"
+            r"|172\.(?:1[6-9]|2\d|3[01])\.\d{1,3}\.\d{1,3}"
+            r"|192\.168\.\d{1,3}\.\d{1,3}"
+            r")\b"
+        ),
+    ),
+    (
+        "absolute_path_windows",
+        re.compile(
+            r"(?i)(?<![A-Za-z0-9_])[A-Z]:[\\/](?![\\/])"
+            r"(?:[^\s\"'<>|`\r\n]|[ ](?=[^\s]*[\\/]))+"
+        ),
+    ),
+    (
+        "absolute_path_unc",
+        re.compile(
+            r"(?i)(?<!\\)\\\\[^\\\s]+\\"
+            r"(?:[^\s\"'<>|`\r\n]|[ ](?=[^\s]*[\\/]))+"
+        ),
+    ),
+)
+
+# Loopback and link-local addresses intentionally remain readable: they identify
+# the current host/link, not private fleet topology. CGNAT and IPv6 ULA policy is
+# separate from this RFC1918-specific intake rule.
+
 
 def redact_text(text: str) -> tuple[str, list[str]]:
     """Public API: redact secrets from arbitrary text.
@@ -533,6 +569,17 @@ def sanitize_persisted_text(text: str) -> tuple[str, list[str]]:
     return redacted, findings
 
 
+def _sanitize_memory_claim_text(text: str) -> tuple[str, list[str]]:
+    """Apply generic secret filtering plus the stricter claim-memory policy."""
+    redacted, findings = sanitize_persisted_text(text)
+    for name, pattern in _CLAIM_ONLY_PATTERNS:
+        if pattern.search(redacted) is None:
+            continue
+        redacted = pattern.sub(f"[REDACTED:{name}]", redacted)
+        findings.append(name)
+    return redacted, sorted(set(findings))
+
+
 def _structured_context_findings(value: str, context_keys: tuple[str, ...]) -> list[str]:
     independent_findings = scan_persisted_value(value)
     findings: list[str] = []
@@ -600,9 +647,9 @@ def sanitize_claim_structure_input(
     object_value: str | None,
 ) -> SanitizedClaimStructureInput:
     validate_persisted_metadata({"claim_type": claim_type})
-    sanitized_subject, subject_findings = _sanitize_optional_claim_text(subject)
-    sanitized_predicate, predicate_findings = _sanitize_optional_claim_text(predicate)
-    sanitized_object, object_findings = _sanitize_optional_claim_text(object_value)
+    sanitized_subject, subject_findings = _sanitize_optional_memory_claim_text(subject)
+    sanitized_predicate, predicate_findings = _sanitize_optional_memory_claim_text(predicate)
+    sanitized_object, object_findings = _sanitize_optional_memory_claim_text(object_value)
     findings = sorted(set(subject_findings + predicate_findings + object_findings))
     return SanitizedClaimStructureInput(
         claim_type=claim_type,
@@ -669,13 +716,21 @@ def _sanitize_optional_claim_text(value: str | None) -> tuple[str | None, list[s
     return sanitize_persisted_text(value)
 
 
+def _sanitize_optional_memory_claim_text(
+    value: str | None,
+) -> tuple[str | None, list[str]]:
+    if value is None:
+        return None, []
+    return _sanitize_memory_claim_text(value)
+
+
 def _sanitize_claim_citations(
     citations: list[CitationInput],
 ) -> tuple[list[CitationInput], list[str]]:
     sanitized: list[CitationInput] = []
     findings: list[str] = []
     for citation in citations:
-        excerpt, excerpt_findings = _sanitize_optional_claim_text(citation.excerpt)
+        excerpt, excerpt_findings = _sanitize_optional_memory_claim_text(citation.excerpt)
         findings.extend(excerpt_findings)
         sanitized.append(CitationInput(citation.source, citation.locator, excerpt))
     return sanitized, findings
@@ -722,10 +777,10 @@ def sanitize_claim_input(
         "tenant_id": tenant_id,
         **citation_metadata,
     })
-    redacted_text, findings = sanitize_persisted_text(text)
-    redacted_object, object_findings = _sanitize_optional_claim_text(object_value)
-    redacted_subject, subject_findings = _sanitize_optional_claim_text(subject)
-    redacted_predicate, predicate_findings = _sanitize_optional_claim_text(predicate)
+    redacted_text, findings = _sanitize_memory_claim_text(text)
+    redacted_object, object_findings = _sanitize_optional_memory_claim_text(object_value)
+    redacted_subject, subject_findings = _sanitize_optional_memory_claim_text(subject)
+    redacted_predicate, predicate_findings = _sanitize_optional_memory_claim_text(predicate)
     sanitized_citations, citation_findings = _sanitize_claim_citations(citations)
     findings.extend(object_findings + subject_findings + predicate_findings + citation_findings)
     dedup_findings = sorted(set(findings))

tests/test_claim_intake_private_context.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+import json
+import sqlite3
+from pathlib import Path
+
+from memorymaster.core.models import CitationInput
+from memorymaster.core.security import (
+    sanitize_claim_input,
+    sanitize_claim_structure_input,
+    sanitize_persisted_text,
+)
+from memorymaster.stores.storage import SQLiteStore
+
+
+def test_claim_policy_redacts_private_ips_in_every_content_field() -> None:
+    values = ("10.1.2.3", "172.16.2.4", "192.168.3.5", "10.6.7.8", "172.31.9.10")
+    result = sanitize_claim_input(
+        text=f"Topology uses {values[0]}",
+        subject=f"host {values[1]}",
+        predicate=f"routes through {values[2]}",
+        object_value=f"backup {values[3]}",
+        citations=[CitationInput("test", excerpt=f"observed {values[4]}")],
+    )
+
+    rendered = "\n".join(
+        (result.text, result.subject or "", result.predicate or "", result.object_value or "",
+         result.citations[0].excerpt or "")
+    )
+    assert all(value not in rendered for value in values)
+    assert rendered.count("[REDACTED:private_ipv4]") == len(values)
+    assert result.is_sensitive and result.findings == ["private_ipv4"]
+
+
+def test_claim_policy_redacts_absolute_windows_and_unc_paths() -> None:
+    drive_path = "Q:\\Synthetic User\\Private Project\\artifact.txt"
+    unc_path = "\\\\private-host\\operator-share\\artifact.txt"
+    result = sanitize_claim_structure_input(
+        claim_type="fact",
+        subject=f"workspace {drive_path}",
+        predicate="stored_at",
+        object_value=unc_path,
+    )
+
+    rendered = "\n".join((result.subject or "", result.object_value or ""))
+    assert drive_path not in rendered and unc_path not in rendered
+    assert "[REDACTED:absolute_path_windows]" in rendered
+    assert "[REDACTED:absolute_path_unc]" in rendered
+
+
+def test_claim_path_redaction_preserves_following_reasoning() -> None:
+    text = "The daemon at C:/Users/x/app.cfg died and the root cause was a boot race"
+    result = sanitize_claim_input(text=text, object_value=None, citations=[])
+
+    assert result.text == (
+        "The daemon at [REDACTED:absolute_path_windows] "
+        "died and the root cause was a boot race"
+    )
+    unc_text = "The daemon at \\\\host\\share\\app.cfg died after the boot race"
+    unc_result = sanitize_claim_input(text=unc_text, object_value=None, citations=[])
+    assert unc_result.text == (
+        "The daemon at [REDACTED:absolute_path_unc] died after the boot race"
+    )
+
+
+def test_claim_path_redaction_supports_spaces_before_later_segments() -> None:
+    path = "Q:\\Synthetic User\\Py Apps\\memorymaster\\artifact.txt"
+    text = f"Read {path} before deployment"
+    result = sanitize_claim_input(text=text, object_value=None, citations=[])
+
+    assert path not in result.text
+    assert result.text.endswith("before deployment")
+
+
+def test_claim_policy_allows_repo_relative_paths() -> None:
+    text = "Read _intel/briefs/task.md, scripts/run.py, and runs/evaluation/result.json."
+    result = sanitize_claim_input(text=text, object_value=None, citations=[])
+
+    assert result.text == text
+    assert not result.is_sensitive
+    assert result.findings == []
+
+
+def test_raw_source_policy_remains_unchanged() -> None:
+    text = "Raw source selected at 10.1.2.3 and Q:\\Synthetic User\\artifact.txt"
+
+    assert sanitize_persisted_text(text) == (text, [])
+
+
+def test_direct_sqlite_claim_write_never_persists_private_context(tmp_path: Path) -> None:
+    db_path = tmp_path / "private-context.db"
+    store = SQLiteStore(db_path)
+    store.init_db()
+    private_ip = "192.168.44.12"
+    local_path = "Q:\\Synthetic User\\Private Project\\artifact.txt"
+
+    claim = store.create_claim(
+        text=f"Mount {private_ip} from {local_path}",
+        citations=[CitationInput("unit-test", excerpt=f"Seen at {private_ip}")],
+        scope="project:intake-policy",
+        source_agent="intake-policy-test",
+    )
+
+    with sqlite3.connect(db_path) as conn:
+        payload = conn.execute(
+            "SELECT payload_json FROM events WHERE claim_id=? AND details='sensitive_redaction_applied'",
+            (claim.id,),
+        ).fetchone()[0]
+        persisted = "\n".join(
+            row[0] for row in conn.execute(
+                "SELECT text FROM claims WHERE id=? UNION ALL "
+                "SELECT COALESCE(excerpt, '') FROM citations WHERE claim_id=?",
+                (claim.id, claim.id),
+            )
+        )
+    assert private_ip not in persisted and local_path not in persisted
+    assert json.loads(payload)["findings"] == ["absolute_path_windows", "private_ipv4"]

tests/test_deterministic_predicates.py

@@ -33,7 +33,7 @@ def test_deterministic_validator_accepts_richer_valid_predicates() -> None:
     service.init_db()
 
     _ingest(service, predicate="ipv6", object_value="2001:db8::1")
-    _ingest(service, predicate="cidr", object_value="10.20.0.0/16")
+    _ingest(service, predicate="cidr", object_value="203.0.113.0/24")
     _ingest(service, predicate="uuid", object_value="123e4567-e89b-42d3-a456-426614174000")
     _ingest(service, predicate="phone_number", object_value="+14155550100")
     _ingest(service, predicate="country_code", object_value="US")

tests/test_fts5_search.py

@@ -51,12 +51,12 @@ def test_init_db_is_idempotent(self, store: SQLiteStore) -> None:
 
 class TestFTS5Search:
     def test_basic_text_match(self, store: SQLiteStore) -> None:
-        store.create_claim("Server IP is 10.0.0.1", _cite(), subject="server", predicate="ip")
+        store.create_claim("Server IP is 203.0.113.10", _cite(), subject="server", predicate="ip")
         store.create_claim("Database runs on port 5432", _cite(), subject="db", predicate="port")
 
         results = store.list_claims(text_query="server IP")
         assert len(results) == 1
-        assert "10.0.0.1" in results[0].text
+        assert "203.0.113.10" in results[0].text
 
     def test_single_token_match(self, store: SQLiteStore) -> None:
         store.create_claim("The authentication token expires daily", _cite())

tests/test_sqlite_core.py

@@ -33,34 +33,34 @@ def test_sqlite_cycle_and_hybrid_retrieval():
     service.init_db()
 
     service.ingest(
-        text="Server IP is 10.0.0.1",
+        text="Server IP is 203.0.113.10",
         citations=[CitationInput(source="session://chat", locator="turn-1", excerpt="first ip")],
         subject="server",
         predicate="ip_address",
-        object_value="10.0.0.1",
+        object_value="203.0.113.10",
         volatility="high",
     )
     service.ingest(
-        text="Server IP is 10.0.0.2",
+        text="Server IP is 203.0.113.11",
         citations=[CitationInput(source="session://chat", locator="turn-2", excerpt="corrected ip")],
         subject="server",
         predicate="ip_address",
-        object_value="10.0.0.2",
+        object_value="203.0.113.11",
         volatility="high",
     )
     service.ingest(
-        text="Credentials file path is C:\\secrets\\prod.env",
+        text="Credentials file path is config/prod.env",
         citations=[CitationInput(source="session://chat", locator="turn-3", excerpt="credential path")],
         subject="workspace",
         predicate="path",
-        object_value="C:\\secrets\\prod.env",
+        object_value="config/prod.env",
     )
 
     result = service.run_cycle(policy_mode="legacy", min_citations=1, min_score=0.5)
     assert result["validator"]["processed"] >= 3
 
     rows = service.query("server ip", retrieval_mode="hybrid", limit=10, allow_sensitive=True)
-    assert any("10.0.0.2" in row.text for row in rows)
+    assert any("203.0.113.11" in row.text for row in rows)
     assert all("Credentials file path" not in row.text for row in rows[:2])