diff --git a/README.md b/README.md index 4efdfec..7b47b37 100644 --- a/README.md +++ b/README.md @@ -11,7 +11,7 @@ The submission is a complete local application: a deterministic audit engine, mu - **Evidence adapters** for JSON, SQLite-compatible SQL, structured logs, text, PDFs, and screenshots. - **Temporal safety** at both evidence-capture and activity-link levels; post-`audit_at` data cannot influence a verdict. - **Constraint-based workflow attribution** that resolves anonymous batched targets only when every cutoff-safe one-to-one assignment agrees. -- **Policy planner** for delete, redact, anonymize, retain, block-restore, and disable-flow actions. +- **Data-driven policy planner** that loads `deletion_rules.csv` and resolves delete, redact, anonymize, retain, block-restore, and disable-flow actions. - **Fail-closed uncertainty**: missing subject-linked proof produces `INCOMPLETE` with a warning instead of an invented `COMPLETE` verdict. - **Investigation UI** with search, status filters, case drill-down, evidence lineage, ordered plan, responsive layout, theme support, and keyboard-aware controls. - **Zero-runtime-model architecture**: reproducible predictions, no network dependency, no API key, and no per-request inference cost. @@ -82,7 +82,7 @@ Then open [http://127.0.0.1:8787](http://127.0.0.1:8787). The UI and API are ser 1. **Intent:** multilingual deletion language is accepted; correction, export-only, marketing-only, and explicit non-deletion requests are rejected as `NOT_APPLICABLE`. 2. **Identity:** the request email is resolved to one stable user, including historical email and phone aliases valid for the request timeline. -3. **Evidence:** structured sources are parsed semantically; PDFs and images use local extraction with a content-addressable text cache. +3. **Evidence:** JSON, cloud-envelope, syslog, pipe-delimited, SQL, and structured-text sources are parsed semantically; PDFs and images use local extraction with a content-addressable text cache. 4. **Time:** evidence captures, source-level event times, and resource-to-subject activity links must exist at or before `audit_at`. 5. **Attribution:** anonymous workflow targets are solved as a cutoff-aware bipartite assignment; a target is accepted only when it is invariant across every valid assignment. 6. **State:** the latest eligible observation per system/resource is selected. @@ -117,7 +117,7 @@ On the 32 participant-visible labeled cases: GhostTrace reaches every participant-visible sample verdict without reading labels at prediction time. Remaining exact-sequence gaps are concentrated in canonical resource links that are not present in cutoff-eligible participant evidence; the engine does not invent those links. Full methodology and limitations are documented in [`code/evaluation/evaluation_report.md`](code/evaluation/evaluation_report.md). -The final full-data run currently produces 148 `COMPLETE`, 360 `INCOMPLETE`, and 10 `NOT_APPLICABLE` decisions. Its `output.csv` passes the strict validator with 518/518 rows. Step count is bounded after parser hardening (median 5, 95th percentile 12, maximum 14), preventing service instance IDs from being promoted to user resources. The workflow solver is independently checked against 164 participant-evidence subject links with zero mismatches; those links validate the assignments but are not permitted to bypass each case's runtime cutoff. +The final full-data run currently produces 148 `COMPLETE`, 360 `INCOMPLETE`, and 10 `NOT_APPLICABLE` decisions. Its `output.csv` passes the strict validator with 518/518 rows. Every authorized case now has an evidence-grounded plan containing the status-required phases; no `COMPLETE` or `INCOMPLETE` row is empty. Step count remains bounded after parser hardening (median 6, 95th percentile 13, maximum 14), preventing service instance IDs from being promoted to user resources. The workflow solver is independently checked against 164 participant-evidence subject links with zero mismatches; those links validate the assignments but are not permitted to bypass each case's runtime cutoff. ## Security and privacy posture @@ -140,6 +140,7 @@ code/ │ ├── intent.py multilingual intent gate │ ├── evidence.py multi-format evidence adapters/index │ ├── workflow.py cutoff-safe batch assignment solver +│ ├── policy.py deletion_rules.csv policy interpreter │ ├── engine.py policy reducer and plan generator │ ├── validation.py strict output contract validator │ └── server.py local investigation API/server diff --git a/code/evaluation/evaluation_report.md b/code/evaluation/evaluation_report.md index 27cf26e..f59ff20 100644 --- a/code/evaluation/evaluation_report.md +++ b/code/evaluation/evaluation_report.md @@ -10,7 +10,7 @@ Command: python3 code/evaluation/main.py ``` -Environment: Python 3.14.0, deterministic local execution, 32 public sample cases, 518 full-dataset cases, 94 evidence files, and 2,903 extracted observations. Extraction completed with zero adapter warnings in the measured environment. +Environment: Python 3.14.6, deterministic local execution, 32 public sample cases, 518 full-dataset cases, 94 evidence files, and 2,984 extracted observations. Extraction completed with zero adapter warnings in the measured environment. ## Public-sample metrics @@ -37,7 +37,7 @@ The production command emitted 518/518 canonically sorted rows and passed every - Empty evidence IDs on future remediation/verification. - Canonical phase, operation, system, and resource ordering. -Plan-size distribution after parser hardening: median 5 steps, 95th percentile 12, maximum 14. No plan exceeds 20 steps. +Plan-size distribution after parser hardening: median 6 steps, 95th percentile 13, maximum 14. No plan exceeds 20 steps, and every authorized case contains a non-empty status-appropriate plan. ## Additional evaluation cases @@ -77,6 +77,10 @@ Generate the entire 518-row file in a temporary directory and run the strict val 10. **Evidence integrity.** The validator now rejects missing finding citations and evidence IDs absent from every participant-visible evidence namespace. 11. **Anonymous batch attribution.** Local nearest-event matching was insufficient when several deletion workflows shared a report time. A cutoff-aware bipartite solver now filters candidates by workflow-result compatibility and ownership conflicts, anchors explicit subject matches, enforces one-to-one assignment, and emits a resource only when every valid assignment agrees. An independent participant-evidence audit validates 164 resolved targets with zero mismatches. Evidence after a target cutoff is used only by this offline regression check, never by runtime prediction. 12. **Source event-time reconstruction.** Structured text logs previously inherited the file capture time for every line. Leading ISO timestamps are now preserved per observation, so later `NOT_FOUND` states correctly supersede earlier `FOUND` states without violating the case cutoff. +13. **Multi-dialect trace loss.** Structured JSON lines, cloud-runtime envelopes, RFC-style syslog, and pipe-delimited operations streams placed subject and resource state on different lines under a shared trace identifier. The previous parser recognized only `trace_id=` syntax and left 15 authorized cases without plans. The trace correlator now parses all supplied dialects, preserves terminal resource timestamps, and reduces empty authorized plans from 15 to zero. Against the preceding release, this reconstructs 158 additional contract steps across 53 full-data rows while retaining the public verdict gate. +14. **Static policy duplication.** Required actions were previously mirrored in source code. The runtime now loads and validates `deletion_rules.csv`; only operational interpretations absent from its action vocabulary—disabling active flows and blocking an unsafe restore while retaining the archive—are derived by the policy interpreter. +15. **Catalog-to-artifact reconciliation.** Object-storage screenshots expose filenames while the data catalog exposes stable resource IDs. A time-scoped filename alias join now connects those two participant-visible facts, recovering `object_19` for case 61 without deriving an identifier from naming coincidence alone. +16. **Incomplete-plan validation.** The validator now rejects empty or phase-incomplete `INCOMPLETE` plans and empty `COMPLETE` verification arrays. This makes the brief's status-specific content requirement an executable release gate. ## Remaining limitations @@ -93,4 +97,4 @@ python3 code/main.py run python3 code/main.py validate ``` -Final regression result: **21/21 tests pass**. The suite includes manifest-completeness, extraction-health, temporal-safety, invariant-assignment, and independent-attribution gates so a checkout cannot silently omit evidence or accept an unsupported workflow mapping. +Final regression result: **29/29 tests pass**. The suite includes manifest-completeness, extraction-health, temporal-safety, trace-dialect, dynamic-policy, full-plan, alias-validity, invariant-assignment, independent-attribution, and byte-reproducible packaging gates so a checkout cannot silently omit evidence, accept an unsupported mapping, or emit a timestamp-dependent submission archive. diff --git a/code/ghosttrace/engine.py b/code/ghosttrace/engine.py index 8e6cea4..1484dc1 100644 --- a/code/ghosttrace/engine.py +++ b/code/ghosttrace/engine.py @@ -9,6 +9,7 @@ from .evidence import EvidenceIndex from .identity import IdentityIndex from .intent import authorizes_deletion +from .policy import PolicyIndex from .workflow import WorkflowAttributor @@ -31,39 +32,11 @@ "VERIFY_RESTORE_BLOCKED": 5, } -ACTION_BY_CATEGORY = { - "USER_PROFILE": "DELETE", - "SECONDARY_IDENTIFIER": "DELETE", - "SEARCH_INDEX_ENTRY": "DELETE", - "CACHE_ENTRY": "DELETE", - "SUPPORT_MESSAGE": "REDACT", - "SUPPORT_ATTACHMENT": "DELETE", - "ANALYTICS_PROFILE": "ANONYMIZE", - "PRODUCT_TELEMETRY": "ANONYMIZE", - "USER_EXPORT": "DELETE", - "MARKETING_PROFILE": "DELETE", - "EMAIL_DELIVERY_LOG": "REDACT", - "SECURITY_AUDIT_LOG": "RETAIN", - "LEGAL_HOLD_RECORD": "RETAIN", - "BACKUP_ARCHIVE": "BLOCK_RESTORE", - "ACCOUNT_TOMBSTONE": "REDACT", - "DATA_FLOW": "DISABLE_DATA_FLOW", -} - -TARGET_STATE = { - "DELETE": "ABSENT", - "REDACT": "REDACTED", - "ANONYMIZE": "ANONYMIZED", - "RETAIN": "PRESENT", - "BLOCK_RESTORE": "BLOCKED", - "DISABLE_DATA_FLOW": "DISABLED", -} - - class GhostTraceEngine: def __init__(self, dataset_dir: Path, cache_dir: Path | None = None): self.dataset_dir = dataset_dir.resolve() self.identities = IdentityIndex(self.dataset_dir) + self.policy = PolicyIndex(self.dataset_dir) self.evidence = EvidenceIndex(self.dataset_dir, cache_dir=cache_dir) self.cases = self._load_cases(self.dataset_dir / "deletion_requests.csv") context_cases = list(self.cases) @@ -113,6 +86,10 @@ def audit(self, case: Case) -> Decision: observation for observation in self.evidence.eligible(case.audit_at) if self.evidence.matches(observation, aliases, case.audit_at) + and not ( + observation.attributes.get("inference") == "catalog_file_alias" + and observation.observed_at < case.requested_at + ) ] workflow_account = self.workflow.account_resource( case @@ -131,7 +108,9 @@ def audit(self, case: Case) -> Decision: ): observations.append(anonymization_proof) observations = latest_by_resource(observations) - observations = [item for item in observations if item.category in ACTION_BY_CATEGORY] + observations = [ + item for item in observations if self.policy.supports(item.category) + ] # Prefer state evidence close to the audit cutoff while retaining one # authoritative state per resource. Resource history is still available @@ -171,10 +150,8 @@ def audit(self, case: Case) -> Decision: warnings=warnings, ) - @staticmethod - def _compliant(observation: Observation) -> bool: - action = ACTION_BY_CATEGORY[observation.category] - return observation.state == TARGET_STATE[action] + def _compliant(self, observation: Observation) -> bool: + return observation.state == self.policy.target_state(observation) def _steps( self, @@ -187,7 +164,7 @@ def _steps( for observation in observations: if self._compliant(observation): continue - action = ACTION_BY_CATEGORY[observation.category] + action = self.policy.action_for(observation) if action == "DELETE": finding = ("REPORT_REMAINING_RECORD", "PRESENT") remediation = ("DELETE_RECORD", "ABSENT") @@ -210,7 +187,7 @@ def _steps( *(planned_verifications or []), ] for observation in verification_observations: - action = ACTION_BY_CATEGORY[observation.category] + action = self.policy.action_for(observation) operation = { "DELETE": "VERIFY_RECORD_ABSENT", "RETAIN": "VERIFY_RECORD_PRESENT", @@ -220,7 +197,14 @@ def _steps( "BLOCK_RESTORE": "VERIFY_RESTORE_BLOCKED", }[action] evidence_id = observation.evidence_id if status == "COMPLETE" else "" - raw.append(("VERIFICATION", observation.system_id, observation.resource_id, operation, TARGET_STATE[action], evidence_id)) + raw.append(( + "VERIFICATION", + observation.system_id, + observation.resource_id, + operation, + self.policy.target_state(observation), + evidence_id, + )) raw.sort(key=lambda item: (PHASE_ORDER[item[0]], OPERATION_ORDER[item[3]], item[1], item[2])) return [ @@ -232,6 +216,7 @@ def diagnostics(self) -> dict[str, object]: return { "cases": len(self.cases), "identities": len(self.identities.identities), + "policy_rules": len(self.policy.rules), "evidence_files": len(self.evidence.files), "observations": len(self.evidence.observations), "extraction_warnings": list(self.evidence.extraction_warnings), diff --git a/code/ghosttrace/evidence.py b/code/ghosttrace/evidence.py index bead19e..11b3111 100644 --- a/code/ghosttrace/evidence.py +++ b/code/ghosttrace/evidence.py @@ -20,8 +20,14 @@ EMAIL_RE = re.compile(r"[A-Z0-9._%+\-]+@[A-Z0-9.\-]+\.[A-Z]{2,}", re.I) USER_RE = re.compile(r"\buser[_\s-]?(\d+)\b", re.I) HASH_RE = re.compile(r"\b[a-f0-9]{64}\b", re.I) +FILE_REF_RE = re.compile( + r"\b[A-Za-z0-9][A-Za-z0-9_.-]*\.(?:csv|jpeg|jpg|json|pdf|png|zip)\b", + re.I, +) KV_RE = re.compile(r"([A-Za-z_][\w.-]*)=(\"[^\"]*\"|'[^']*'|\S+)") ISO_PREFIX_RE = re.compile(r"^\s*(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z)\b") +MESSAGE_RE = re.compile(r'\bmessage="((?:\\.|[^"\\])*)"') +TRACE_RE = re.compile(r"\b(?:tr|rb)-\d{5,}\b", re.I) PREFIX_META = { "account": ("account_service", "USER_PROFILE"), @@ -176,7 +182,12 @@ def _subjects(record: dict[str, Any]) -> set[str]: def _has_pii(record: dict[str, Any]) -> bool: for key, value in _flatten(record): leaf = key.rsplit(".", 1)[-1].casefold() - if value in (None, "", [], {}): + if value in (None, "", [], {}) or normalize(value) in { + "null", + "none", + "redacted", + "[redacted]", + }: continue if any(token in leaf for token in ("email", "phone", "full_name", "requester_name", "email_hash", "display_name")): return True @@ -201,29 +212,102 @@ def _record_time(record: dict[str, Any], fallback): return fallback +def _as_bool(value: Any) -> bool | None: + if isinstance(value, bool): + return value + folded = normalize(value) + if folded in {"true", "1", "yes", "enabled", "allow", "allowed"}: + return True + if folded in {"false", "0", "no", "disabled", "deny", "denied"}: + return False + return None + + +def _structured_line_record(line: str, line_no: int) -> dict[str, Any]: + """Parse the structured log dialects used by participant evidence.""" + record: dict[str, Any] = {} + leading_time = ISO_PREFIX_RE.search(line) + remainder = line[leading_time.end():].strip() if leading_time else line.strip() + if remainder.startswith("{"): + try: + payload = json.loads(remainder) + if isinstance(payload, dict): + record.update(payload) + except json.JSONDecodeError: + pass + if not record: + record.update( + (key, value.strip("\"'")) for key, value in KV_RE.findall(line) + ) + embedded = MESSAGE_RE.search(line) + if embedded: + message = embedded.group(1).replace('\\"', '"').replace("\\\\", "\\") + record.update( + (key, value.strip("\"'")) + for key, value in KV_RE.findall(message) + ) + parts = [part.strip() for part in line.split("|")] + if len(parts) >= 6: + record.setdefault("service", parts[2]) + record.setdefault("trace_id", parts[3]) + record.setdefault("event", parts[4]) + record["_line_no"] = line_no + if leading_time: + record["observed_at"] = leading_time.group(1) + return record + + +def _trace_key(line: str, record: dict[str, Any]) -> str: + for key in ("trace_id", "requestId", "request_id"): + value = str(record.get(key, "")).strip() + if value and value.casefold().startswith(("tr-", "rb-")): + return value + match = TRACE_RE.search(line) + return match.group(0) if match else "" + + def _infer_state(category: str, record: dict[str, Any], text: str = "") -> str: result = normalize(record.get("result") or record.get("state") or record.get("status")) if result in RESULT_STATE: return RESULT_STATE[result] - status_code = record.get("status_code") - if status_code == 404 or (category == "USER_PROFILE" and "profile" in record and record.get("profile") is None): + status_code = str(record.get("status_code") or record.get("http_status") or "") + rows = str(record.get("rows", "")) + if status_code == "404" or ( + category == "USER_PROFILE" + and (rows == "0" or ("profile" in record and record.get("profile") is None)) + ): return "ABSENT" - if status_code == 200: + if status_code == "200" or (category == "USER_PROFILE" and rows not in {"", "0"}): return "PRESENT" if category == "DATA_FLOW": - enabled = record.get("enabled") + enabled = _as_bool(record.get("enabled")) if enabled is not None: - return "ACTIVE" if bool(enabled) else "DISABLED" + return "ACTIVE" if enabled else "DISABLED" if category == "BACKUP_ARCHIVE": - allowed = record.get("restore_allowed") + decision = normalize(record.get("decision")) + if decision in {"deny", "denied", "block", "blocked"}: + return "BLOCKED" + deletion_block = _as_bool(record.get("deletion_block")) + if deletion_block is not None: + return "BLOCKED" if deletion_block else "ACTIVE" + allowed = _as_bool(record.get("restore_allowed")) if allowed is not None: - return "ACTIVE" if bool(allowed) else "BLOCKED" + return "ACTIVE" if allowed else "BLOCKED" + eligible = _as_bool(record.get("eligible")) + if eligible is not None: + return "ACTIVE" if eligible else "BLOCKED" if category == "SEARCH_INDEX_ENTRY": if "hits" in record: return "PRESENT" if record.get("hits") else "ABSENT" if normalize(record.get("result")) in {"hit", "found"}: return "PRESENT" if category == "ANALYTICS_PROFILE": + direct_identifiers = record.get("direct_identifiers") + if direct_identifiers is not None: + try: + return "PRESENT" if int(direct_identifiers) > 0 else "ANONYMIZED" + except (TypeError, ValueError): + pass return "PRESENT" if _has_pii(record) else "ANONYMIZED" if category in {"SUPPORT_MESSAGE", "ACCOUNT_TOMBSTONE", "EMAIL_DELIVERY_LOG"}: text_has_pii = bool(EMAIL_RE.search(text) or re.search(r"\+?\d[\d\s()\-]{7,}", text)) @@ -248,6 +332,7 @@ def __init__(self, dataset_dir: Path, cache_dir: Path | None = None): self.cache_dir = cache_dir self.files = self._load_manifest() self.resource_links: dict[str, list[tuple[Any, set[str], str]]] = defaultdict(list) + self.resource_aliases: dict[str, list[tuple[str, Any]]] = defaultdict(list) self.activity_events: list[dict[str, Any]] = [] self.observations: list[Observation] = [] self.observations_by_resource: dict[str, list[Observation]] = defaultdict(list) @@ -289,6 +374,10 @@ def _load_activity_links(self) -> None: self.resource_links[resource].append( (parse_time(row.get("event_time")), subjects, row.get("activity_id", "")) ) + for reference in FILE_REF_RE.findall(details): + self.resource_aliases[normalize(reference)].append( + (resource, parse_time(row.get("event_time"))) + ) def _extract_all(self) -> None: for evidence in self.files: @@ -435,9 +524,38 @@ def _from_text( ) -> list[Observation]: if not text.strip(): return [] - groups: dict[str, list[tuple[int, str]]] = defaultdict(list) + groups: dict[str, list[tuple[int, str, dict[str, Any]]]] = defaultdict(list) free_lines: list[str] = [] result: list[Observation] = [] + + def append_observation( + record: dict[str, Any], + context: str, + *, + resource_id: str | None = None, + ) -> None: + resource = canonical_resource(resource_id or record.get("resource_id", "")) + if not resource or not RESOURCE_RE.search(resource): + return + system_id, category = meta_for_text_resource( + resource, evidence, record, context + ) + subjects = _subjects(record) + subjects.update(normalize(value) for value in EMAIL_RE.findall(context)) + subjects.update(f"user_{value}" for value in USER_RE.findall(context)) + result.append( + Observation( + system_id=system_id, + resource_id=resource, + category=category, + state=_infer_state(category, record, context), + observed_at=_record_time(record, evidence.captured_at), + evidence_id=evidence.file_id, + subjects=subjects, + attributes=dict(record), + ) + ) + for line_no, line in enumerate(text.splitlines(), 1): parts = [part.strip().lstrip("- ") for part in line.split("|")] if len(parts) >= 6 and RESOURCE_RE.search(parts[2]): @@ -467,68 +585,55 @@ def _from_text( ) ) continue - match = re.search(r"\btrace_id=([^\s]+)", line) - if match: - groups[match.group(1)].append((line_no, line)) - elif "resource_id=" in line: - record = {key: value.strip("\"'") for key, value in KV_RE.findall(line)} - record["_line_no"] = line_no - leading_time = ISO_PREFIX_RE.search(line) - if leading_time: - record["observed_at"] = leading_time.group(1) - resource = canonical_resource(record.get("resource_id", "")) - if resource: - system_id, category = meta_for_text_resource( - resource, evidence, record, line - ) - subjects = _subjects(record) - subjects.update(normalize(v) for v in EMAIL_RE.findall(line)) - subjects.update(f"user_{v}" for v in USER_RE.findall(line)) - result.append( - Observation( - system_id=system_id, - resource_id=resource, - category=category, - state=_infer_state(category, record, line), - observed_at=_record_time(record, evidence.captured_at), - evidence_id=evidence.file_id, - subjects=subjects, - attributes=record, - ) - ) + record = _structured_line_record(line, line_no) + trace = _trace_key(line, record) + if trace: + groups[trace].append((line_no, line, record)) + elif record.get("resource_id"): + append_observation(record, line) else: free_lines.append(line) for entries in groups.values(): - lines = [line for _line_no, line in entries] + lines = [line for _line_no, line, _record in entries] merged = " ".join(lines) - record = {key: value.strip("\"'") for key, value in KV_RE.findall(merged)} - record["_line_no"] = min(line_no for line_no, _line in entries) - leading_time = ISO_PREFIX_RE.search(lines[0]) - if leading_time: - record["observed_at"] = leading_time.group(1) - explicit_resource = record.get("resource_id", "") - resources = {canonical_resource(explicit_resource)} if explicit_resource else set() + record: dict[str, Any] = {} + resources: dict[str, Any] = {} + for line_no, _line, line_record in entries: + record.update(line_record) + explicit_resource = canonical_resource( + str(line_record.get("resource_id", "")) + ) + if explicit_resource and RESOURCE_RE.search(explicit_resource): + resources[explicit_resource] = _record_time( + line_record, evidence.captured_at + ) if not resources: continue + record["_line_no"] = min(line_no for line_no, _line, _record in entries) subjects = _subjects(record) subjects.update(normalize(v) for v in EMAIL_RE.findall(merged)) subjects.update(f"user_{v}" for v in USER_RE.findall(merged)) - for resource in resources: - system_id, category = meta_for_text_resource( - resource, evidence, record, merged - ) - result.append( - Observation( - system_id=system_id, - resource_id=resource, - category=category, - state=_infer_state(category, record, merged), - observed_at=_record_time(record, evidence.captured_at), - evidence_id=evidence.file_id, - subjects=set(subjects), - attributes=record, - ) - ) + for resource, observed_at in resources.items(): + resource_record = dict(record) + resource_record["resource_id"] = resource + resource_record["observed_at"] = observed_at.isoformat() + before = len(result) + append_observation(resource_record, merged, resource_id=resource) + if len(result) > before: + result[-1].subjects = set(subjects) + + folded_text = normalize(text) + for reference, links in self.resource_aliases.items(): + if reference not in folded_text: + continue + for resource, linked_at in links: + alias_record = { + "resource_id": resource, + "file_reference": reference, + "inference": "catalog_file_alias", + "observed_at": max(evidence.captured_at, linked_at).isoformat(), + } + append_observation(alias_record, reference, resource_id=resource) if not allow_freeform: return result diff --git a/code/ghosttrace/policy.py b/code/ghosttrace/policy.py new file mode 100644 index 0000000..34e0024 --- /dev/null +++ b/code/ghosttrace/policy.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +import csv +from pathlib import Path + +from .domain import Observation + + +SUPPORTED_ACTIONS = {"DELETE", "REDACT", "ANONYMIZE", "RETAIN"} +TARGET_STATE = { + "DELETE": "ABSENT", + "REDACT": "REDACTED", + "ANONYMIZE": "ANONYMIZED", + "RETAIN": "PRESENT", + "BLOCK_RESTORE": "BLOCKED", + "DISABLE_DATA_FLOW": "DISABLED", +} + + +class PolicyIndex: + """Load the participant-supplied deletion policy and resolve required actions.""" + + def __init__(self, dataset_dir: Path) -> None: + self.path = dataset_dir / "deletion_rules.csv" + self.rules: dict[str, dict[str, str]] = {} + with self.path.open(encoding="utf-8", newline="") as handle: + for row in csv.DictReader(handle): + category = row.get("data_category", "").strip().upper() + action = row.get("required_action", "").strip().upper() + if not category or action not in SUPPORTED_ACTIONS: + raise ValueError( + f"Invalid deletion rule {row.get('rule_id', '')}: " + f"category={category!r}, action={action!r}" + ) + if category in self.rules: + raise ValueError(f"Duplicate deletion rule for {category}") + self.rules[category] = dict(row) + if not self.rules: + raise ValueError("deletion_rules.csv contains no usable rules") + + def supports(self, category: str) -> bool: + return category in self.rules or category == "DATA_FLOW" + + def action_for(self, observation: Observation) -> str: + if observation.category == "DATA_FLOW": + return "DISABLE_DATA_FLOW" + rule = self.rules.get(observation.category) + if rule is None: + raise KeyError(f"No deletion policy for {observation.category}") + action = rule["required_action"].strip().upper() + if observation.category == "BACKUP_ARCHIVE" and observation.state in { + "ACTIVE", + "BLOCKED", + }: + # The archive remains retained; the privacy control required by the + # rule's condition is to block subject restoration from that copy. + return "BLOCK_RESTORE" + return action + + def target_state(self, observation: Observation) -> str: + return TARGET_STATE[self.action_for(observation)] diff --git a/code/ghosttrace/validation.py b/code/ghosttrace/validation.py index dce461e..b1235f0 100644 --- a/code/ghosttrace/validation.py +++ b/code/ghosttrace/validation.py @@ -70,8 +70,17 @@ def validate_output(output_path: Path, request_path: Path) -> dict[str, Any]: continue if status == "NOT_APPLICABLE" and steps: errors.append(f"{case_id}: NOT_APPLICABLE must have []") + if status == "COMPLETE" and not steps: + errors.append(f"{case_id}: COMPLETE lacks evidence-backed verification") if status == "COMPLETE" and any(step.get("phase") != "VERIFICATION" for step in steps): errors.append(f"{case_id}: COMPLETE contains non-verification steps") + if status == "INCOMPLETE": + phases = {step.get("phase") for step in steps} + missing_phases = {"FINDING", "REMEDIATION", "VERIFICATION"} - phases + if missing_phases: + errors.append( + f"{case_id}: INCOMPLETE plan lacks phases {sorted(missing_phases)}" + ) keys = [] for index, step in enumerate(steps, 1): if set(step) != FIELDS: diff --git a/code/package_submission.py b/code/package_submission.py index e3e6db7..8336eb6 100644 --- a/code/package_submission.py +++ b/code/package_submission.py @@ -11,6 +11,7 @@ OUTPUT = ROOT / "code.zip" EXCLUDED_PARTS = {"__pycache__", ".venv", "venv", "node_modules", ".git"} EXCLUDED_SUFFIXES = {".pyc", ".pyo", ".zip", ".tar", ".gz", ".tgz", ".7z", ".rar"} +FIXED_ZIP_TIMESTAMP = (1980, 1, 1, 0, 0, 0) def included_files() -> list[Path]: @@ -24,11 +25,30 @@ def included_files() -> list[Path]: ) +def build_archive(output: Path = OUTPUT) -> list[Path]: + """Write a byte-reproducible archive independent of checkout timestamps.""" + files = included_files() + with zipfile.ZipFile(output, "w") as archive: + for path in files: + info = zipfile.ZipInfo( + path.relative_to(ROOT).as_posix(), + date_time=FIXED_ZIP_TIMESTAMP, + ) + info.compress_type = zipfile.ZIP_DEFLATED + info.create_system = 3 + info.external_attr = 0o100644 << 16 + archive.writestr( + info, + path.read_bytes(), + compress_type=zipfile.ZIP_DEFLATED, + compresslevel=9, + ) + return files + + def main() -> int: - with zipfile.ZipFile(OUTPUT, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as archive: - for path in included_files(): - archive.write(path, path.relative_to(ROOT)) - print(f"Created {OUTPUT} with {len(included_files())} files") + files = build_archive() + print(f"Created {OUTPUT} with {len(files)} files") return 0 diff --git a/code/tests/test_engine.py b/code/tests/test_engine.py index 97861c0..1239d88 100644 --- a/code/tests/test_engine.py +++ b/code/tests/test_engine.py @@ -1,9 +1,11 @@ from __future__ import annotations import csv +import hashlib import json import tempfile import unittest +import zipfile from pathlib import Path @@ -14,12 +16,14 @@ sys.path.insert(0, str(CODE_DIR)) -from ghosttrace.domain import parse_time # noqa: E402 +from ghosttrace.domain import Observation, parse_time # noqa: E402 from ghosttrace.engine import GhostTraceEngine, OPERATION_ORDER, PHASE_ORDER # noqa: E402 from ghosttrace.intent import authorizes_deletion # noqa: E402 +from ghosttrace.policy import PolicyIndex # noqa: E402 from ghosttrace.validation import validate_output # noqa: E402 from ghosttrace.workflow import RESULT_FOR_STATUS # noqa: E402 from main import write_output # noqa: E402 +from package_submission import build_archive # noqa: E402 class GhostTraceEngineTests(unittest.TestCase): @@ -41,6 +45,63 @@ def test_every_manifest_artifact_exists_and_extracts(self) -> None: self.assertEqual(missing, []) self.assertEqual(self.engine.evidence.extraction_warnings, []) + def test_every_authorized_case_has_an_evidence_grounded_plan(self) -> None: + decisions = self.engine.audit_all() + empty = [ + decision.case_id + for decision in decisions + if decision.deletion_status != "NOT_APPLICABLE" and not decision.steps + ] + self.assertEqual(empty, []) + self.assertEqual( + [decision.case_id for decision in decisions if decision.warnings], + [], + ) + + def test_structured_log_dialects_preserve_trace_ownership(self) -> None: + expected = { + "case_265": ("account_5015", "evidence_25", "ABSENT"), + "case_363": ("account_9205", "evidence_34", "ABSENT"), + "case_416": ("account_3075", "evidence_39", "ABSENT"), + "case_521": ("account_1482", "evidence_48", "ABSENT"), + } + cases = {case.case_id: case for case in self.engine.cases} + for case_id, (resource_id, evidence_id, state) in expected.items(): + case = cases[case_id] + identity = self.engine.identities.resolve(case) + aliases = self.engine.identities.aliases_at(identity, case) + matching = [ + item + for item in self.engine.evidence.eligible(case.audit_at) + if item.resource_id == resource_id + and item.evidence_id == evidence_id + and self.engine.evidence.matches(item, aliases, case.audit_at) + ] + self.assertEqual(len(matching), 1, case_id) + self.assertEqual(matching[0].state, state, case_id) + + def test_catalog_file_alias_joins_object_storage_evidence(self) -> None: + case = next(item for item in self.engine.cases if item.case_id == "case_61") + decision = self.engine.audit(case) + finding = [ + step + for step in decision.steps + if step.resource_id == "object_19" and step.phase == "FINDING" + ] + self.assertEqual(len(finding), 1) + self.assertEqual(finding[0].evidence_id, "evidence_46") + + def test_catalog_file_alias_is_not_reused_for_a_later_request(self) -> None: + case = next( + item + for item in self.engine._load_cases( + self.dataset / "sample_deletion_requests.csv" + ) + if item.case_id == "case_07" + ) + decision = self.engine.audit(case) + self.assertNotIn("object_19", {step.resource_id for step in decision.steps}) + def test_intent_gate_matches_known_non_deletion_requests(self) -> None: not_applicable = {f"case_{number}" for number in range(533, 543)} classified = {case.case_id for case in self.engine.cases if not authorizes_deletion(case.request_text)} @@ -224,6 +285,29 @@ def test_output_is_complete_and_contract_valid(self) -> None: validation = validate_output(output, self.dataset / "deletion_requests.csv") self.assertTrue(validation["valid"], validation["errors"][:5]) + def test_submission_archive_is_reproducible_and_dataset_free(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + first = Path(temp_dir) / "first.zip" + second = Path(temp_dir) / "second.zip" + build_archive(first) + build_archive(second) + self.assertEqual( + hashlib.sha256(first.read_bytes()).digest(), + hashlib.sha256(second.read_bytes()).digest(), + ) + with zipfile.ZipFile(first) as archive: + names = archive.namelist() + self.assertEqual(archive.testzip(), None) + self.assertFalse( + any( + name == "log.txt" + or name.startswith("dataset/") + or "__pycache__" in name + or name.endswith(".zip") + for name in names + ) + ) + def test_validator_rejects_unknown_evidence(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: output = Path(temp_dir) / "output.csv" @@ -248,6 +332,31 @@ def test_validator_rejects_unknown_evidence(self) -> None: any("unknown evidence_id" in error for error in validation["errors"]) ) + def test_validator_rejects_incomplete_without_a_complete_plan(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + output = Path(temp_dir) / "output.csv" + write_output(self.engine, output) + with output.open(encoding="utf-8", newline="") as handle: + rows = list(csv.DictReader(handle)) + target = next( + row for row in rows if row["deletion_status"] == "INCOMPLETE" + ) + target["steps_json"] = "[]" + with output.open("w", encoding="utf-8", newline="") as handle: + writer = csv.DictWriter( + handle, + fieldnames=["case_id", "deletion_status", "steps_json"], + ) + writer.writeheader() + writer.writerows(rows) + validation = validate_output( + output, self.dataset / "deletion_requests.csv" + ) + self.assertFalse(validation["valid"]) + self.assertTrue( + any("INCOMPLETE plan lacks phases" in error for error in validation["errors"]) + ) + def test_steps_are_canonically_sorted(self) -> None: for decision in self.engine.audit_all(): keys = [ @@ -285,6 +394,66 @@ def test_non_deletion(self) -> None: self.assertTrue(all(not authorizes_deletion(text) for text in examples)) +class PolicyTests(unittest.TestCase): + def test_policy_actions_are_loaded_from_the_supplied_csv(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + path = Path(temp_dir) / "deletion_rules.csv" + with path.open("w", encoding="utf-8", newline="") as handle: + writer = csv.DictWriter( + handle, + fieldnames=( + "rule_id", + "data_category", + "condition", + "required_action", + ), + ) + writer.writeheader() + writer.writerow( + { + "rule_id": "test_rule", + "data_category": "USER_PROFILE", + "condition": "Always", + "required_action": "REDACT", + } + ) + policy = PolicyIndex(Path(temp_dir)) + observation = Observation( + system_id="account_service", + resource_id="account_test", + category="USER_PROFILE", + state="PRESENT", + observed_at=parse_time("2026-01-01T00:00:00Z"), + evidence_id="evidence_test", + ) + self.assertEqual(policy.action_for(observation), "REDACT") + self.assertEqual(policy.target_state(observation), "REDACTED") + + def test_retained_backup_uses_restore_control_when_active(self) -> None: + policy = PolicyIndex(REPO_ROOT / "dataset") + self.assertEqual( + policy.rules["BACKUP_ARCHIVE"]["required_action"], "RETAIN" + ) + active = Observation( + system_id="backup_service", + resource_id="backup_test", + category="BACKUP_ARCHIVE", + state="ACTIVE", + observed_at=parse_time("2026-01-01T00:00:00Z"), + evidence_id="evidence_test", + ) + blocked = Observation( + system_id=active.system_id, + resource_id=active.resource_id, + category=active.category, + state="BLOCKED", + observed_at=active.observed_at, + evidence_id=active.evidence_id, + ) + self.assertEqual(policy.action_for(active), "BLOCK_RESTORE") + self.assertEqual(policy.target_state(blocked), "BLOCKED") + + if __name__ == "__main__": unittest.main()