diff --git a/memorymaster/core/security.py b/memorymaster/core/security.py index 08a53824..b45926ca 100644 --- a/memorymaster/core/security.py +++ b/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)(?|`\r\n]|[ ](?=[^\s]*[\\/]))+" + ), + ), + ( + "absolute_path_unc", + re.compile( + r"(?i)(?|`\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)) diff --git a/tests/test_claim_intake_private_context.py b/tests/test_claim_intake_private_context.py new file mode 100644 index 00000000..bf7c7f9e --- /dev/null +++ b/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"] diff --git a/tests/test_deterministic_predicates.py b/tests/test_deterministic_predicates.py index 681354b5..82cec437 100644 --- a/tests/test_deterministic_predicates.py +++ b/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") diff --git a/tests/test_fts5_search.py b/tests/test_fts5_search.py index 961b5a4a..b342fd37 100644 --- a/tests/test_fts5_search.py +++ b/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()) diff --git a/tests/test_sqlite_core.py b/tests/test_sqlite_core.py index a9d7f9fa..a02c0e22 100644 --- a/tests/test_sqlite_core.py +++ b/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])