feat(governance): enforce public export boundary

This commit is contained in:
muratcankoylan
2026-08-10 04:06:39 -04:00
committed by Muratcan Koylan
parent 4deb96cf1f
commit 61bb052268
25 changed files with 2156 additions and 8 deletions
+8
View File
@@ -37,10 +37,13 @@ jobs:
researcher/scripts/skill_frontmatter.py \
researcher/scripts/governance_policy.py \
researcher/scripts/build_inventory.py \
researcher/scripts/export_policy.py \
researcher/scripts/tests/test_skill_frontmatter.py \
researcher/scripts/tests/test_build_inventory.py \
researcher/scripts/tests/test_export_policy.py \
researcher/scripts/tests/test_governance_policy.py \
researcher/scripts/validate_governance.py \
researcher/scripts/validate_export.py \
researcher/scripts/validate_platform_compat.py \
researcher/scripts/validate_repo.py \
researcher/scripts/validate_run.py \
@@ -67,6 +70,11 @@ jobs:
- name: Generated repository inventory
run: python researcher/scripts/build_inventory.py --check
- name: Public export boundary
run: >-
python researcher/scripts/validate_export.py check
--staging-dir researcher/exports/examples/restricted-citation-v1
- name: Benchmark runner contract
working-directory: researcher/benchmarks/sdk-runner
run: |
+2
View File
@@ -68,6 +68,8 @@ researcher/queue/inbox.jsonl
researcher/queue/parked.jsonl
researcher/queue/done.jsonl
researcher/queue/quarantine.jsonl
researcher/exports/private/
researcher/exports/staging/
# Active research runs live under researcher/runs/; the seed run is kept as a
# committed fixture, everything else is local runtime state.
+1
View File
@@ -26,6 +26,7 @@ Workspace memory for agents collaborating on this repository. Keep entries durab
- Detailed lessons from building the researcher OS live in `researcher/insights/auto-research-experiment.md` (engineering rationale) and `researcher/insights/how-we-built-this.md` (project narrative and sharing templates); read both before extending the harness or writing release-facing prose.
- Benchmarks are staged in `researcher/benchmarks/`: Stage 0 deterministic harness (shipped), Stage 1 per-skill health via `researcher/scripts/skill_health.py` (shipped; output `researcher/reports/skill-health.json` is gitignored), Stage 2 router (shipped; results in `researcher/benchmarks/router/results-published/`), Stage 3 effectiveness (scaffolded, one task built), Stage 4 composition (future). `researcher/benchmarks/PLAN.md` is the methodology source of truth.
- Current corpus counts and compatibility status are generated in `researcher/generated/corpus-summary.md`; do not copy live totals into workspace memory. Published benchmark reports remain dated snapshots. Do not describe a skill improvement as complete unless the prose, mechanism registry, claim index, corpus index, activation fixtures, generated inventory, and validators all agree.
- Public export uses `validate_export.py plan|render|check` and registered transforms from `governance/export-policy.yaml`. Private plans and receipts are ignored; public manifests contain projection and output digests, never private source paths or input digests.
- Benchmark execution uses the Cursor SDK runner at `researcher/benchmarks/sdk-runner/` (TypeScript, `@cursor/sdk` 1.0.13). The runner supports `--concurrency N`, `--no-resume`, per-run progress logging, format-failure retry, and worst-case retry-aware cost forecasting; default behavior is to resume by skipping plan items that already have result files. Result artifacts under `researcher/benchmarks/{router,effectiveness}/results/` and history JSONLs (`router-history.jsonl`, `effectiveness-history.jsonl`) are gitignored.
- Published Stage 2 router-benchmark results: `researcher/benchmarks/router/results-published/2026-05-15.md` (baseline), `researcher/benchmarks/router/results-published/2026-05-15-v2.md` (post-rewrite with delta-vs-baseline table), and `researcher/benchmarks/router/results-published/2026-05-19.md` (post-corpus-hardening validation: 600/600 usable records, 0 format failures, top-1 Gemini 0.920 / Composer 0.913 / GPT-5.5 0.913 / Claude Opus 4.7 0.840). Headline finding: targeted description rewrites moved `context-fundamentals` top-1 by +23.4pp and `project-development` top-1 to 1.000; corpus-wide hardening did not cause broad routing collapse.
+1
View File
@@ -282,6 +282,7 @@ Current counts and compatibility status are generated in the [live corpus invent
- **Adversarial benchmark harness** (`researcher/benchmarks/`): scenarios that try to game the loop (duplicate mechanisms, unretrieved evidence, wrong rubric math, self-approved rubric changes, weak-evidence novelty).
- **Continuous loop** (`researcher/scripts/loop_*.py` + `researcher/orchestration/launchd/`): inbox, source discovery, one-state-at-a-time advancement, daily ops, parked review queue, launchd service definitions.
- **Skill health gate** (`researcher/scripts/skill_health.py`): deterministic body-quality scoring. Published scores remain dated evidence; local runs produce ignored runtime reports.
- **Public export boundary** (`governance/export-policy.yaml` + `researcher/scripts/validate_export.py`): allowlisted projections from private or restricted records into reviewable public staging trees without publishing private source locators or digests.
### Operator commands
@@ -0,0 +1,44 @@
# ADR-0003: Publish allowlisted projections, not redacted private records
- Status: accepted for proposal
- Date: 2026-08-10
- Spec: SPEC-002
## Context
Future research feeds, human feedback, notification destinations, licensed source bodies, and operational traces cannot all be stored in the public repository. A generic redaction pass would be difficult to reason about and could publish private locators, low-entropy source hashes, or nested fields while appearing sanitized.
## Decision
Movement to the public repository creates a new projection artifact through a registered transform. The source remains private or restricted. Rendering accepts exact request fields, source classifications, artifact kinds, transforms, and output prefixes. Unknown routes deny.
Export state is split into immutable records:
- a private `ExportPlan` binds source paths and exact source digests;
- a public `ExportManifest` binds only new projection IDs, transformation digests, output paths, and output digests;
- a private render receipt retains source-to-projection audit data;
- public validation, later approval, merge receipt, correction, and tombstone records point backward rather than mutating earlier records.
Private source digests are deliberately absent from public manifests. Even a cryptographic hash can reveal the presence of a low-entropy private value through guessing. The public projection receives a new opaque ID.
Rendering occurs in a fresh sibling directory and becomes visible through atomic rename only after the complete tree validates. Structural field allowlists and complete-tree closure are the primary boundary. Seeded canaries and high-confidence pattern detectors are supplementary checks, not a universal data-loss-prevention claim.
## Alternatives considered
- Copy and redact arbitrary records. Rejected because unknown and nested fields fail open.
- Publish source hashes for auditability. Rejected because hashes can be existence oracles.
- Put private data on a private Git branch. Rejected because branch visibility is not a durable data classification boundary.
- Make scanning the only control. Rejected because encodings and semantic disclosures cannot be exhaustively detected.
- Mutate one manifest through planned, rendered, and merged states. Rejected because durable artifacts should be immutable and replayable.
## Consequences
- Adding an exportable artifact kind requires a reviewed policy and transform change.
- Public lineage is useful but intentionally does not let a public consumer resolve the private source.
- Real private plans and receipts stay outside Git or under ignored private paths.
- Publishing remains irreversible in practice. Corrections and tombstones supersede public artifacts but cannot undo disclosure.
- SPEC-003 registers these bootstrap records in the common schema and identity system; SPEC-024 later owns production credential and private-storage resolution.
## Verification
Tests cover unknown fields, classification denial, traversal, Unicode/case collisions, symlinks, source mutation, atomic staging, extra files, digest tampering, private manifest fields, plain/hex/base64 canaries, high-confidence credential structures, duplicate JSON keys, policy drift, unsupported output fields, and the restricted citation projection.
@@ -0,0 +1,54 @@
# SPEC-002: Public and private boundary
- Status: implementing
- Wave: 0
- Classification: split
- Depends on: SPEC-000
## Decision
The public repository contains reproducible skills, rubrics, mechanisms, claims, public evidence metadata, evaluations, decisions, schemas, and sanitized examples. Private control-plane state contains credentials, identity material, personal destinations, restricted raw data, hidden evaluations, unreleased traces, and private human notes.
Private-to-public movement creates a new allowlisted projection. It never republishes a private record in place.
## Classifications
- `public`: authored for unrestricted repository publication.
- `public_derived`: a reviewed projection produced from another class.
- `private_operational`: queues, plans, receipts, traces, costs, and deployment state.
- `private_human`: private feedback, notes, and review context.
- `restricted_source`: raw content whose redistribution is limited.
- `secret_reference`: an opaque credential or capability reference; never exportable.
Each new export record declares classification and retention. Legacy artifact envelopes are registered and migrated by SPEC-003 rather than rewritten in this layer.
## Invariants
1. Public manifests contain no private input digest, source path, storage locator, credential reference, capability token, or private destination.
2. Restricted raw content may produce citation metadata through a registered transform but may not enter the staged public tree.
3. Unknown classifications, artifact kinds, transforms, fields, destinations, and files deny.
4. Rendering does not mutate sources and verifies the exact source bytes pinned by the private plan.
5. Staging becomes visible only after complete-tree validation; symlinks, non-regular files, traversal, collisions, extras, and missing outputs fail.
6. A public projection has a new identity. Its reference is provenance, not access authority.
7. Publication is not reversible; correction and tombstone records supersede prior artifacts without rewriting ordinary Git history.
## Interfaces
- `governance/export-policy.yaml`: classifications, retention defaults, routes, transformations, public fields, and correction rules.
- `validate_export.py plan`: create a private immutable source-bound plan.
- `validate_export.py render`: project into a fresh public staging tree and persist a private receipt.
- `validate_export.py check`: verify manifest closure, digests, policy pins, schemas, and supported detectors without private-source access.
- `researcher/exports/schemas/export-records.schema.json`: bootstrap request, plan, and manifest contract.
## Acceptance criteria
- [x] Export routes and classifications are allowlisted and versioned.
- [x] Secret-reference sources cannot enter a supported public projection.
- [x] Public records expose no private hash or locator.
- [x] Restricted-source raw bodies are structurally excluded while citation metadata remains useful.
- [x] Rendering is deterministic across fresh directories and atomic at the tree boundary.
- [x] Public staging closure, private canaries, source mutation, and path attacks fail closed.
- [x] Correction and removal procedures are documented.
- [x] A committed restricted-source example validates in CI.
This boundary does not claim full security hardening or universal detection of unknown semantic disclosures. Its guarantee is limited to registered transforms and validated staging paths.
+51
View File
@@ -0,0 +1,51 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://github.com/muratcankoylan/Agent-Skills-for-Context-Engineering/governance/export-policy.schema.json",
"title": "Public export policy",
"type": "object",
"additionalProperties": false,
"required": [
"schema_version",
"policy_version",
"default_export",
"max_source_bytes",
"classifications",
"public_output_classifications",
"artifact_kinds",
"transforms",
"forbidden_public_fields",
"allowed_public_digest_fields",
"public_manifest_fields",
"public_manifest_entry_fields",
"supplementary_detectors",
"correction_policy"
],
"properties": {
"schema_version": {"const": "1.0.0"},
"policy_version": {"type": "string", "pattern": "^[0-9]+\\.[0-9]+\\.[0-9]+$"},
"default_export": {"const": "deny"},
"max_source_bytes": {"type": "integer", "minimum": 1},
"classifications": {
"type": "object",
"minProperties": 1,
"additionalProperties": {
"type": "object",
"additionalProperties": false,
"required": ["default_retention", "exportable"],
"properties": {
"default_retention": {"type": "string"},
"exportable": {"type": "boolean"}
}
}
},
"public_output_classifications": {"type": "array", "items": {"type": "string"}, "uniqueItems": true},
"artifact_kinds": {"type": "object"},
"transforms": {"type": "object"},
"forbidden_public_fields": {"type": "array", "items": {"type": "string"}, "uniqueItems": true},
"allowed_public_digest_fields": {"type": "array", "items": {"type": "string"}, "uniqueItems": true},
"public_manifest_fields": {"type": "array", "items": {"type": "string"}, "uniqueItems": true},
"public_manifest_entry_fields": {"type": "array", "items": {"type": "string"}, "uniqueItems": true},
"supplementary_detectors": {"type": "object"},
"correction_policy": {"type": "object"}
}
}
+154
View File
@@ -0,0 +1,154 @@
schema_version: "1.0.0"
policy_version: "1.0.0"
default_export: deny
max_source_bytes: 1500000
classifications:
public:
default_retention: durable
exportable: true
public_derived:
default_retention: durable
exportable: true
private_operational:
default_retention: review_cycle
exportable: true
private_human:
default_retention: until_superseded
exportable: true
restricted_source:
default_retention: source_policy
exportable: true
secret_reference:
default_retention: key_lifetime
exportable: false
public_output_classifications:
- public
- public_derived
artifact_kinds:
citation_metadata:
output_prefix: citation/
allowed_transforms:
- restricted_citation_v1
research_summary:
output_prefix: research-summary/
allowed_transforms:
- research_summary_v1
transforms:
restricted_citation_v1:
source_classifications:
- restricted_source
input_media_type: application/json
output_media_type: application/json
required_source_fields:
- title
- url
- source_type
allowed_source_fields:
- title
- url
- source_type
- authors
- published_at
- license_note
output_fields:
- schema_version
- id
- kind
- classification
- retention
- title
- url
- source_type
- authors
- published_at
- license_note
research_summary_v1:
source_classifications:
- private_operational
- private_human
- public_derived
input_media_type: application/json
output_media_type: application/json
required_source_fields:
- title
- summary
- reason_code
allowed_source_fields:
- title
- summary
- reason_code
- evidence_links
- uncertainty
output_fields:
- schema_version
- id
- kind
- classification
- retention
- title
- summary
- reason_code
- evidence_links
- uncertainty
forbidden_public_fields:
- source_path
- source_digest
- private_digest
- storage_binding
- storage_locator
- signed_url
- credential
- credential_ref
- credential_value
- capability_token
- email_destination
- phone_destination
- private_reason
- raw_body
- raw_content
- body
- content
allowed_public_digest_fields:
- output_digest
- policy_digest
- transformation_digest
- manifest_digest
public_manifest_fields:
- schema_version
- id
- kind
- state
- classification
- retention
- policy_version
- policy_digest
- entries
public_manifest_entry_fields:
- entry_id
- projection_id
- artifact_kind
- transformation_id
- transformation_digest
- output_path
- output_digest
- media_type
- size_bytes
supplementary_detectors:
pem_private_key: true
authorization_header: true
signed_url: true
private_absolute_path: true
correction_policy:
rewrite_git_history_by_default: false
correction_requires_new_record: true
removal_requires_public_tombstone: true
+2
View File
@@ -2,6 +2,8 @@
Current corpus counts, source digests, compatibility status, and unresolved-reference status are generated in the [live corpus inventory](generated/corpus-summary.md). Historical reports retain the values measured at their dated snapshot.
The public/private boundary is defined by `governance/export-policy.yaml` and `scripts/validate_export.py`. Real private plans and receipts stay under ignored local paths; only validated projection trees are proposed for public review.
This directory defines the repo-native workflow for turning external research into skill changes. It is intentionally file-based so agents can inspect, resume, and audit work without requiring a hosted scheduler.
## Mission
+126 -7
View File
@@ -875,6 +875,60 @@
}
]
},
"export_contracts": {
"count": 8,
"owner": "governance/export-policy.yaml and export fixtures",
"records": [
{
"digest": "sha256:bbc9a19a969246698c80a69c0692108a2ffabf84dbdbcb02a5ed127c1de2ced1",
"id": "governance/export-policy.yaml",
"path": "governance/export-policy.yaml",
"size_bytes": 3091
},
{
"digest": "sha256:500c1d138249a135e462e2fcbdbab0874e6ef841c7ca005560106bb727359722",
"id": "governance/export-policy.schema.json",
"path": "governance/export-policy.schema.json",
"size_bytes": 1982
},
{
"digest": "sha256:cbece8b7d22d83de802cb2c4484a82dc529808a8e6f6f67eac5c8d2a2727c14a",
"id": "researcher/exports/schemas/export-records.schema.json",
"path": "researcher/exports/schemas/export-records.schema.json",
"size_bytes": 5921
},
{
"digest": "sha256:a9d63b7a247082485f84e2a9df51e9b7b009f2e6841c05bb41bc275bf75a7304",
"id": "researcher/fixtures/export/restricted-request.json",
"path": "researcher/fixtures/export/restricted-request.json",
"size_bytes": 553
},
{
"digest": "sha256:6a0ee2e312f7798802bbc159120c8b621edc3cb0b52d17c9f337a8f61bfbf33b",
"id": "researcher/fixtures/export/private-root/restricted-source.json",
"path": "researcher/fixtures/export/private-root/restricted-source.json",
"size_bytes": 353
},
{
"digest": "sha256:6db00594ea7cac8b8a467be57e0407bb8d737ab70279132aaa2569d48123d9bd",
"id": "researcher/exports/examples/restricted-citation-v1/export-manifest.json",
"path": "researcher/exports/examples/restricted-citation-v1/export-manifest.json",
"size_bytes": 860
},
{
"digest": "sha256:3cd4d08885889eb70364bb5a3bfbff76ddcf49d5ebc5e9d11cc4bd79f6fd66dd",
"id": "researcher/exports/examples/restricted-citation-v1/citation/restricted-fixture.json",
"path": "researcher/exports/examples/restricted-citation-v1/citation/restricted-fixture.json",
"size_bytes": 453
},
{
"digest": "sha256:826eb51007060a637823c454e79460caf1478de7f1fa1e7371ab1d8573cccb57",
"id": "researcher/scripts/export_policy.py",
"path": "researcher/scripts/export_policy.py",
"size_bytes": 40910
}
]
},
"manifests": {
"count": 3,
"owner": "publication manifests",
@@ -1789,7 +1843,7 @@
]
},
"validators": {
"count": 7,
"count": 8,
"owner": "declared validator ownership",
"records": [
{
@@ -1803,7 +1857,7 @@
"path": "researcher/scripts/validate_governance.py"
},
{
"digest": "sha256:7a2fd204985b43715950d2ef406de3b6841b23c77d52eba470f32f5be327cfb3",
"digest": "sha256:c354b8821c9651458703d8e235730c36c485f02ba50405673ccaa274c8e69eb1",
"id": "repository-inventory",
"owns": [
"identifier uniqueness",
@@ -1813,6 +1867,16 @@
],
"path": "researcher/scripts/build_inventory.py"
},
{
"digest": "sha256:9bc56b6f2602a1efec29f40e3cb40cd906e79f1b7fd5dd3d5872da4ef69ea9da",
"id": "export-policy",
"owns": [
"classification export routes",
"public projections",
"staged export closure"
],
"path": "researcher/scripts/validate_export.py"
},
{
"digest": "sha256:facebf497d771a018adff8c0bbbbcbb1b28cf38a797c82d48e47fcc9c46fd81f",
"id": "platform-compatibility",
@@ -2249,12 +2313,12 @@
]
},
"repository_revision": {
"digest": "sha256:f46f5cc8faccc4ab64a23f68aea2ac6a7c7802fa3f0860c1ce05392ea2ea1542",
"digest": "sha256:2b5adf00cf1c92aabaff97aba86e2b21956faf4f1eb6357e4faa8c1271ffa22b",
"git_commit_excluded_to_avoid_self_reference": true,
"kind": "canonical_source_tree"
},
"schema_version": "1.0.0",
"source_tree_digest": "sha256:f46f5cc8faccc4ab64a23f68aea2ac6a7c7802fa3f0860c1ce05392ea2ea1542",
"source_tree_digest": "sha256:2b5adf00cf1c92aabaff97aba86e2b21956faf4f1eb6357e4faa8c1271ffa22b",
"sources": [
{
"digest": "sha256:2aebace36e5bbdd8ad93bc9c7563a0c673787838eb0f4f62a5376132d43ce616",
@@ -2301,6 +2365,16 @@
"path": "examples/x-to-book-system/README.md",
"size_bytes": 10108
},
{
"digest": "sha256:500c1d138249a135e462e2fcbdbab0874e6ef841c7ca005560106bb727359722",
"path": "governance/export-policy.schema.json",
"size_bytes": 1982
},
{
"digest": "sha256:bbc9a19a969246698c80a69c0692108a2ffabf84dbdbcb02a5ed127c1de2ced1",
"path": "governance/export-policy.yaml",
"size_bytes": 3091
},
{
"digest": "sha256:4ae1af9564863e50508b70e39965c31ddc034fef59dc887e8655736435f52dd6",
"path": "researcher/benchmarks/effectiveness/tasks/001-filesystem-context-offload/README.md",
@@ -2391,11 +2465,36 @@
"path": "researcher/corpus/inventory.schema.json",
"size_bytes": 2054
},
{
"digest": "sha256:3cd4d08885889eb70364bb5a3bfbff76ddcf49d5ebc5e9d11cc4bd79f6fd66dd",
"path": "researcher/exports/examples/restricted-citation-v1/citation/restricted-fixture.json",
"size_bytes": 453
},
{
"digest": "sha256:6db00594ea7cac8b8a467be57e0407bb8d737ab70279132aaa2569d48123d9bd",
"path": "researcher/exports/examples/restricted-citation-v1/export-manifest.json",
"size_bytes": 860
},
{
"digest": "sha256:cbece8b7d22d83de802cb2c4484a82dc529808a8e6f6f67eac5c8d2a2727c14a",
"path": "researcher/exports/schemas/export-records.schema.json",
"size_bytes": 5921
},
{
"digest": "sha256:a6f79df5cbfe39617aa6ce349548821d59d9b7cee363edbbc4d8640a90d88b0a",
"path": "researcher/fixtures/activation-cases.jsonl",
"size_bytes": 11684
},
{
"digest": "sha256:6a0ee2e312f7798802bbc159120c8b621edc3cb0b52d17c9f337a8f61bfbf33b",
"path": "researcher/fixtures/export/private-root/restricted-source.json",
"size_bytes": 353
},
{
"digest": "sha256:a9d63b7a247082485f84e2a9df51e9b7b009f2e6841c05bb41bc275bf75a7304",
"path": "researcher/fixtures/export/restricted-request.json",
"size_bytes": 553
},
{
"digest": "sha256:e658564d470a184b5062a17bde5e265096458922b2405870c6394498427bffbf",
"path": "researcher/mechanisms/ledgers/accepted.jsonl",
@@ -2412,15 +2511,20 @@
"size_bytes": 14630
},
{
"digest": "sha256:7a2fd204985b43715950d2ef406de3b6841b23c77d52eba470f32f5be327cfb3",
"digest": "sha256:c354b8821c9651458703d8e235730c36c485f02ba50405673ccaa274c8e69eb1",
"path": "researcher/scripts/build_inventory.py",
"size_bytes": 48407
"size_bytes": 49862
},
{
"digest": "sha256:91d32d5ebcdbaf8d5a417ed29f43ace7ce126bf427ef6c64176b67612d24987b",
"path": "researcher/scripts/check_activation_cases.py",
"size_bytes": 4529
},
{
"digest": "sha256:826eb51007060a637823c454e79460caf1478de7f1fa1e7371ab1d8573cccb57",
"path": "researcher/scripts/export_policy.py",
"size_bytes": 40910
},
{
"digest": "sha256:ef6cb1f2f03ebd77c6ba56b2adc7dc0f6d0ba3e260272ea7620d744de0f01f3a",
"path": "researcher/scripts/run_benchmarks.py",
@@ -2431,6 +2535,11 @@
"path": "researcher/scripts/skill_health.py",
"size_bytes": 13616
},
{
"digest": "sha256:9bc56b6f2602a1efec29f40e3cb40cd906e79f1b7fd5dd3d5872da4ef69ea9da",
"path": "researcher/scripts/validate_export.py",
"size_bytes": 5382
},
{
"digest": "sha256:7f189e876bc7cba64683a187d1f4f0699864602802cb73cccb30084eaa5d8c08",
"path": "researcher/scripts/validate_governance.py",
@@ -2545,7 +2654,7 @@
"path": "researcher/scripts/validate_governance.py"
},
{
"digest": "sha256:7a2fd204985b43715950d2ef406de3b6841b23c77d52eba470f32f5be327cfb3",
"digest": "sha256:c354b8821c9651458703d8e235730c36c485f02ba50405673ccaa274c8e69eb1",
"id": "repository-inventory",
"owns": [
"identifier uniqueness",
@@ -2555,6 +2664,16 @@
],
"path": "researcher/scripts/build_inventory.py"
},
{
"digest": "sha256:9bc56b6f2602a1efec29f40e3cb40cd906e79f1b7fd5dd3d5872da4ef69ea9da",
"id": "export-policy",
"owns": [
"classification export routes",
"public projections",
"staged export closure"
],
"path": "researcher/scripts/validate_export.py"
},
{
"digest": "sha256:facebf497d771a018adff8c0bbbbcbb1b28cf38a797c82d48e47fcc9c46fd81f",
"id": "platform-compatibility",
+37
View File
@@ -0,0 +1,37 @@
# Public export boundary
The exporter turns explicitly classified private or restricted JSON records into new public projection artifacts. It is an allowlist projector, not a claim that arbitrary files can be made safe by scanning.
The record sequence is immutable:
1. `ExportRequest` selects source-relative paths and registered transforms.
2. `ExportPlan` privately binds the exact source bytes and policy version.
3. `ExportManifest` publicly binds only projection IDs, transform digests, output paths, and output digests.
4. `ExportValidation` proves the staged tree matches its manifest and supported detectors.
5. Human approval, merge receipt, correction, and tombstone records are separate later lifecycle records. Earlier records are never edited in place.
Private input digests, storage locators, source paths, capability material, and notification destinations never appear in the public manifest. A public projection receives a new `proj_` identity. Possessing that identity does not grant access to the private source.
## Commands
```bash
python researcher/scripts/validate_export.py plan \
--request researcher/fixtures/export/restricted-request.json \
--private-root researcher/fixtures/export/private-root \
--plan-out researcher/exports/private/example-plan.json
python researcher/scripts/validate_export.py render \
--plan researcher/exports/private/example-plan.json \
--private-root researcher/fixtures/export/private-root \
--staging-dir researcher/exports/staging/example \
--receipt-out researcher/exports/private/example-receipt.json
python researcher/scripts/validate_export.py check \
--staging-dir researcher/exports/staging/example
```
`researcher/exports/examples/restricted-citation-v1/` is a committed deterministic example. Local `private/` and `staging/` directories are ignored.
## Supported guarantee
Supported rendering paths reject unknown request and manifest fields, unknown artifact kinds, unregistered transforms, secret-reference sources, path traversal, symlinks, path collisions, source mutation, extra staged files, digest drift, private metadata fields, supplied canaries in plain/hex/base64 forms, and a small set of high-confidence secret structures. The structural field allowlist is the primary boundary. Supplementary scanning cannot prove absence of every possible semantic disclosure.
@@ -0,0 +1,12 @@
# Restricted citation projection example
The adjacent generated staging tree proves that a synthetic `restricted_source` record can export allowlisted citation metadata while excluding its raw `body`. `export-manifest.json` contains output and transformation digests but no source path or private input digest.
Validate with:
```bash
python researcher/scripts/validate_export.py check \
--staging-dir researcher/exports/examples/restricted-citation-v1
```
The example source is intentionally synthetic and committed under `researcher/fixtures/export/`; real restricted bodies do not belong in this repository.
@@ -0,0 +1,15 @@
{
"authors": [
"Fixture Author"
],
"classification": "public_derived",
"id": "proj_restricted_fixture",
"kind": "CitationMetadata",
"license_note": "Metadata may be redistributed; raw body may not.",
"published_at": "2026-08-10",
"retention": "durable",
"schema_version": "1.0.0",
"source_type": "restricted_fixture",
"title": "Synthetic Restricted Harness Study",
"url": "https://example.invalid/restricted-harness-study"
}
@@ -0,0 +1,23 @@
{
"classification": "public_derived",
"entries": [
{
"artifact_kind": "citation_metadata",
"entry_id": "entry_restricted_fixture",
"media_type": "application/json",
"output_digest": "sha256:3cd4d08885889eb70364bb5a3bfbff76ddcf49d5ebc5e9d11cc4bd79f6fd66dd",
"output_path": "citation/restricted-fixture.json",
"projection_id": "proj_restricted_fixture",
"size_bytes": 453,
"transformation_digest": "sha256:25a93628ef7d29c48874e74e68ff9bc80fa0f3a942b969ad19833017e818c031",
"transformation_id": "restricted_citation_v1"
}
],
"id": "export_restricted_fixture_v1",
"kind": "ExportManifest",
"policy_digest": "sha256:bbc9a19a969246698c80a69c0692108a2ffabf84dbdbcb02a5ed127c1de2ced1",
"policy_version": "1.0.0",
"retention": "durable",
"schema_version": "1.0.0",
"state": "rendered"
}
@@ -0,0 +1,134 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://github.com/muratcankoylan/Agent-Skills-for-Context-Engineering/researcher/exports/schemas/export-records.schema.json",
"title": "Export boundary records",
"$defs": {
"classification": {
"enum": ["public", "public_derived", "private_operational", "private_human", "restricted_source", "secret_reference"]
},
"retention": {"type": "string", "minLength": 1},
"digest": {"type": "string", "pattern": "^sha256:[0-9a-f]{64}$"},
"exportRequest": {
"type": "object",
"additionalProperties": false,
"required": ["schema_version", "id", "kind", "classification", "retention", "entries"],
"properties": {
"schema_version": {"const": "1.0.0"},
"id": {"type": "string", "pattern": "^export_[0-9A-Za-z_-]+$"},
"kind": {"const": "ExportRequest"},
"classification": {"enum": ["private_operational", "private_human"]},
"retention": {"$ref": "#/$defs/retention"},
"entries": {"type": "array", "minItems": 1}
}
},
"exportPlan": {
"type": "object",
"additionalProperties": false,
"required": ["schema_version", "id", "kind", "state", "classification", "retention", "policy_version", "policy_digest", "request_digest", "entries"],
"properties": {
"schema_version": {"const": "1.0.0"},
"id": {"type": "string"},
"kind": {"const": "ExportPlan"},
"state": {"const": "planned"},
"classification": {"const": "private_operational"},
"retention": {"$ref": "#/$defs/retention"},
"policy_version": {"type": "string"},
"policy_digest": {"$ref": "#/$defs/digest"},
"request_digest": {"$ref": "#/$defs/digest"},
"entries": {"type": "array", "minItems": 1}
}
},
"exportManifest": {
"type": "object",
"additionalProperties": false,
"required": ["schema_version", "id", "kind", "state", "classification", "retention", "policy_version", "policy_digest", "entries"],
"properties": {
"schema_version": {"const": "1.0.0"},
"id": {"type": "string"},
"kind": {"const": "ExportManifest"},
"state": {"const": "rendered"},
"classification": {"const": "public_derived"},
"retention": {"const": "durable"},
"policy_version": {"type": "string"},
"policy_digest": {"$ref": "#/$defs/digest"},
"entries": {"type": "array", "minItems": 1}
}
},
"exportValidation": {
"type": "object",
"additionalProperties": false,
"required": ["schema_version", "id", "kind", "state", "classification", "retention", "policy_version", "policy_digest", "manifest_digest", "outputs", "result"],
"properties": {
"schema_version": {"const": "1.0.0"},
"id": {"type": "string"},
"kind": {"const": "ExportValidation"},
"state": {"const": "validated"},
"classification": {"const": "public_derived"},
"retention": {"const": "durable"},
"policy_version": {"type": "string"},
"policy_digest": {"$ref": "#/$defs/digest"},
"manifest_digest": {"$ref": "#/$defs/digest"},
"outputs": {"type": "array", "minItems": 1},
"result": {"const": "pass"}
}
},
"exportApproval": {
"type": "object",
"additionalProperties": false,
"required": ["schema_version", "id", "kind", "state", "classification", "retention", "manifest_digest", "decision", "reviewer_ref", "reason_code"],
"properties": {
"schema_version": {"const": "1.0.0"},
"id": {"type": "string"},
"kind": {"const": "ExportApproval"},
"state": {"enum": ["approved", "rejected"]},
"classification": {"const": "private_human"},
"retention": {"const": "durable"},
"manifest_digest": {"$ref": "#/$defs/digest"},
"decision": {"enum": ["approve", "reject"]},
"reviewer_ref": {"type": "string", "minLength": 1},
"reason_code": {"type": "string", "pattern": "^[A-Z][A-Z0-9_]+$"}
}
},
"exportReceipt": {
"type": "object",
"additionalProperties": false,
"required": ["schema_version", "id", "kind", "state", "classification", "retention", "manifest_digest", "public_commit", "pull_request"],
"properties": {
"schema_version": {"const": "1.0.0"},
"id": {"type": "string"},
"kind": {"const": "ExportReceipt"},
"state": {"const": "merged"},
"classification": {"const": "public_derived"},
"retention": {"const": "durable"},
"manifest_digest": {"$ref": "#/$defs/digest"},
"public_commit": {"type": "string", "pattern": "^[0-9a-f]{40}$"},
"pull_request": {"type": "integer", "minimum": 1}
}
},
"exportCorrection": {
"type": "object",
"additionalProperties": false,
"required": ["schema_version", "id", "kind", "classification", "retention", "projection_id", "supersedes_manifest_digest", "reason_code", "disposition"],
"properties": {
"schema_version": {"const": "1.0.0"},
"id": {"type": "string"},
"kind": {"const": "ExportCorrection"},
"classification": {"const": "public_derived"},
"retention": {"const": "durable"},
"projection_id": {"type": "string", "pattern": "^proj_[0-9A-Za-z_-]+$"},
"supersedes_manifest_digest": {"$ref": "#/$defs/digest"},
"reason_code": {"type": "string", "pattern": "^[A-Z][A-Z0-9_]+$"},
"disposition": {"enum": ["corrected", "removed", "tombstoned"]}
}
}
},
"oneOf": [
{"$ref": "#/$defs/exportRequest"},
{"$ref": "#/$defs/exportPlan"},
{"$ref": "#/$defs/exportManifest"},
{"$ref": "#/$defs/exportValidation"},
{"$ref": "#/$defs/exportApproval"},
{"$ref": "#/$defs/exportReceipt"},
{"$ref": "#/$defs/exportCorrection"}
]
}
+5
View File
@@ -0,0 +1,5 @@
# Export boundary fixtures
`private-root/` contains synthetic test inputs, not actual private data. The restricted-source fixture intentionally includes a raw body that policy permits the citation projector to read but forbids it to emit. This proves the transform structurally rather than relying on broad text redaction.
Real private roots, export plans, receipts, notification destinations, licensed source bodies, and credentials remain outside Git or under ignored local paths.
@@ -0,0 +1,11 @@
{
"title": "Synthetic Restricted Harness Study",
"url": "https://example.invalid/restricted-harness-study",
"source_type": "restricted_fixture",
"authors": [
"Fixture Author"
],
"published_at": "2026-08-10",
"license_note": "Metadata may be redistributed; raw body may not.",
"body": "SYNTHETIC_RESTRICTED_BODY_MUST_NOT_EXPORT_42"
}
@@ -0,0 +1,18 @@
{
"schema_version": "1.0.0",
"id": "export_restricted_fixture_v1",
"kind": "ExportRequest",
"classification": "private_operational",
"retention": "review_cycle",
"entries": [
{
"entry_id": "entry_restricted_fixture",
"projection_id": "proj_restricted_fixture",
"source_path": "restricted-source.json",
"source_classification": "restricted_source",
"artifact_kind": "citation_metadata",
"transformation_id": "restricted_citation_v1",
"output_path": "citation/restricted-fixture.json"
}
]
}
+1 -1
View File
@@ -4,7 +4,7 @@
This is a generated view of canonical repository artifacts, not a second source of truth.
- Schema: `1.0.0`
- Source tree: `sha256:f46f5cc8faccc4ab64a23f68aea2ac6a7c7802fa3f0860c1ce05392ea2ea1542`
- Source tree: `sha256:2b5adf00cf1c92aabaff97aba86e2b21956faf4f1eb6357e4faa8c1271ffa22b`
- Unresolved references: `0`
| Artifact | Count |
@@ -0,0 +1,14 @@
# Public export correction and removal
Publication is not reversible. A later deletion cannot make already distributed data private again.
When a public projection is incorrect or should no longer be used:
1. Disable further renders of the affected request or transform.
2. Preserve the private plan, render receipt, and failed validation evidence.
3. Open a corrective pull request with a new public correction or tombstone record that names the public projection and safe reason code.
4. Remove or replace the projected body in that pull request when appropriate. Do not rewrite Git history as the normal response.
5. A human maintainer reviews and merges the correction.
6. Downstream indexes mark the prior projection superseded or unavailable while retaining non-sensitive decision lineage.
If the event exposed an actual credential, revoke and rotate it outside this repository. Do not add the value or matched scanner excerpt to the public incident record.
+28
View File
@@ -42,6 +42,11 @@ VALIDATOR_OWNERSHIP = (
"path": "researcher/scripts/build_inventory.py",
"owns": ["identifier uniqueness", "cross-artifact references", "live counts", "generated inventory"],
},
{
"id": "export-policy",
"path": "researcher/scripts/validate_export.py",
"owns": ["classification export routes", "public projections", "staged export closure"],
},
{
"id": "platform-compatibility",
"path": "researcher/scripts/validate_platform_compat.py",
@@ -275,6 +280,7 @@ class InventoryBuilder:
manifests = self.build_manifests()
validators = self.build_validators()
schemas = self.build_schemas()
export_contracts = self.build_export_contracts()
self.validate_live_document_links()
artifacts = {
@@ -293,6 +299,7 @@ class InventoryBuilder:
"manifests": manifests,
"validators": validators,
"schemas": schemas,
"export_contracts": export_contracts,
}
source_records = sorted(self.sources.values(), key=lambda item: item["path"])
source_tree_digest = sha256_bytes(
@@ -932,6 +939,27 @@ class InventoryBuilder:
)
return self._category("researcher/corpus/inventory.schema.json", records)
def build_export_contracts(self) -> dict[str, Any]:
paths = [
"governance/export-policy.yaml",
"governance/export-policy.schema.json",
"researcher/exports/schemas/export-records.schema.json",
"researcher/fixtures/export/restricted-request.json",
"researcher/fixtures/export/private-root/restricted-source.json",
"researcher/exports/examples/restricted-citation-v1/export-manifest.json",
"researcher/exports/examples/restricted-citation-v1/citation/restricted-fixture.json",
"researcher/scripts/export_policy.py",
]
records: list[dict[str, Any]] = []
for relative in paths:
path = self.root / relative
if not path.exists():
self.add_finding("PARSE_ERROR", path, "export contract artifact is missing", relative)
continue
digest, size = self.add_source(path)
records.append({"id": relative, "path": relative, "digest": digest, "size_bytes": size})
return self._category("governance/export-policy.yaml and export fixtures", records)
def validate_live_document_links(self) -> None:
for relative, required_link in LIVE_DOCUMENT_LINKS.items():
path = self.root / relative
+856
View File
@@ -0,0 +1,856 @@
#!/usr/bin/env python3
"""Allowlisted, deterministic projections across the public/private boundary."""
from __future__ import annotations
import base64
import json
import os
import re
import shutil
import stat
import tempfile
import unicodedata
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
from typing import Any, Iterable, Mapping
try:
import yaml
except ImportError as exc: # pragma: no cover - CI installs requirements-dev.txt
raise RuntimeError("PyYAML is required to load governance/export-policy.yaml") from exc
try:
from build_inventory import (
DuplicateKeyError,
atomic_write_text,
parse_json_text,
pretty_json,
record_digest,
sha256_bytes,
)
except ModuleNotFoundError: # Imported as researcher.scripts.export_policy.
from researcher.scripts.build_inventory import (
DuplicateKeyError,
atomic_write_text,
parse_json_text,
pretty_json,
record_digest,
sha256_bytes,
)
ROOT = Path(__file__).resolve().parents[2]
DEFAULT_POLICY_PATH = ROOT / "governance" / "export-policy.yaml"
SCHEMA_VERSION = "1.0.0"
EXPORT_ID = re.compile(r"^export_[0-9A-Za-z_-]+$")
ENTRY_ID = re.compile(r"^entry_[0-9A-Za-z_-]+$")
PROJECTION_ID = re.compile(r"^proj_[0-9A-Za-z_-]+$")
PRIVATE_KEY_PATTERN = re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----")
AUTHORIZATION_PATTERN = re.compile(r"(?i)authorization\s*:\s*(?:bearer|basic)\s+\S+")
SIGNED_URL_PATTERN = re.compile(r"(?i)[?&](?:x-amz-signature|signature|access_token)=[^&\s]+")
PRIVATE_PATH_PATTERN = re.compile(r"(?:/Users/[^/\s]+/|/home/[^/\s]+/|[A-Za-z]:\\Users\\[^\\\s]+\\)")
class ExportBoundaryError(ValueError):
"""Typed failure that never includes matched secret content."""
def __init__(self, code: str, message: str) -> None:
super().__init__(f"{code}: {message}")
self.code = code
self.safe_message = message
@dataclass(frozen=True)
class StableFile:
body: bytes
digest: str
size_bytes: int
def _require_exact_keys(value: Mapping[str, Any], allowed: set[str], location: str) -> None:
unknown = set(value) - allowed
missing = allowed - set(value)
if unknown:
raise ExportBoundaryError("UNKNOWN_FIELD", f"{location} has unknown fields: {sorted(unknown)}")
if missing:
raise ExportBoundaryError("UNKNOWN_FIELD", f"{location} is missing fields: {sorted(missing)}")
def _require_nonempty_string(value: Any, location: str) -> str:
if not isinstance(value, str) or not value.strip():
raise ExportBoundaryError("UNKNOWN_FIELD", f"{location} must be a non-empty string")
return value
def normalize_relative_path(value: Any, location: str) -> str:
raw = _require_nonempty_string(value, location)
if "\x00" in raw or "\\" in raw:
raise ExportBoundaryError("PATH_ESCAPE", f"{location} must use normalized POSIX syntax")
if unicodedata.normalize("NFC", raw) != raw:
raise ExportBoundaryError("PATH_COLLISION", f"{location} must use NFC Unicode")
path = PurePosixPath(raw)
if path.is_absolute() or any(part in {"", ".", ".."} for part in path.parts):
raise ExportBoundaryError("PATH_ESCAPE", f"{location} must remain relative")
return path.as_posix()
def _walk_without_symlinks(root: Path, relative: str) -> Path:
if root.is_symlink():
raise ExportBoundaryError("PATH_ESCAPE", "private root must not be a symlink")
current = root.resolve(strict=True)
if not current.is_dir():
raise ExportBoundaryError("PATH_ESCAPE", "private root must be a regular directory")
for part in PurePosixPath(relative).parts:
current = current / part
if current.is_symlink():
raise ExportBoundaryError("PATH_ESCAPE", "source path contains a symlink")
try:
resolved = current.resolve(strict=True)
resolved.relative_to(root.resolve(strict=True))
except (OSError, ValueError) as exc:
raise ExportBoundaryError("PATH_ESCAPE", "source is unavailable or escapes its private root") from exc
return resolved
def read_stable_file(root: Path, relative: str, max_bytes: int) -> StableFile:
path = _walk_without_symlinks(root, relative)
flags = os.O_RDONLY
if hasattr(os, "O_NOFOLLOW"):
flags |= os.O_NOFOLLOW
try:
descriptor = os.open(path, flags)
except OSError as exc:
raise ExportBoundaryError("PATH_ESCAPE", "source could not be opened safely") from exc
try:
before = os.fstat(descriptor)
if not stat.S_ISREG(before.st_mode):
raise ExportBoundaryError("PATH_ESCAPE", "source must be a regular file")
if before.st_nlink != 1:
raise ExportBoundaryError("PATH_ESCAPE", "source must not be hardlinked")
if before.st_size > max_bytes:
raise ExportBoundaryError("SOURCE_TOO_LARGE", "source exceeds policy byte limit")
chunks: list[bytes] = []
total = 0
while True:
chunk = os.read(descriptor, min(64 * 1024, max_bytes + 1 - total))
if not chunk:
break
chunks.append(chunk)
total += len(chunk)
if total > max_bytes:
raise ExportBoundaryError("SOURCE_TOO_LARGE", "source exceeds policy byte limit")
after = os.fstat(descriptor)
finally:
os.close(descriptor)
identity_before = (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns)
identity_after = (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns)
if identity_before != identity_after or total != after.st_size:
raise ExportBoundaryError("SOURCE_CHANGED", "source changed while it was being read")
body = b"".join(chunks)
return StableFile(body=body, digest=sha256_bytes(body), size_bytes=len(body))
class ExportPolicy:
def __init__(self, document: Mapping[str, Any], digest: str, source: Path) -> None:
self.document = dict(document)
self.digest = digest
self.source = source
self.version = _require_nonempty_string(document.get("policy_version"), "policy_version")
self._validate()
@classmethod
def load(cls, path: Path = DEFAULT_POLICY_PATH) -> "ExportPolicy":
try:
raw = path.read_bytes()
document = yaml.safe_load(raw.decode("utf-8"))
except (OSError, UnicodeError, yaml.YAMLError) as exc:
raise ExportBoundaryError("POLICY_INVALID", "export policy could not be loaded") from exc
if not isinstance(document, dict):
raise ExportBoundaryError("POLICY_INVALID", "export policy root must be a mapping")
return cls(document, sha256_bytes(raw), path.resolve())
@property
def classifications(self) -> Mapping[str, Any]:
return self.document["classifications"]
@property
def transforms(self) -> Mapping[str, Any]:
return self.document["transforms"]
@property
def artifact_kinds(self) -> Mapping[str, Any]:
return self.document["artifact_kinds"]
@property
def max_source_bytes(self) -> int:
return int(self.document.get("max_source_bytes", 1_500_000))
def _validate(self) -> None:
required = {
"schema_version",
"policy_version",
"default_export",
"classifications",
"public_output_classifications",
"artifact_kinds",
"transforms",
"forbidden_public_fields",
"allowed_public_digest_fields",
"public_manifest_fields",
"public_manifest_entry_fields",
"supplementary_detectors",
"correction_policy",
}
optional = {"max_source_bytes"}
unknown = set(self.document) - required - optional
missing = required - set(self.document)
if unknown or missing:
raise ExportBoundaryError(
"POLICY_INVALID", f"policy fields differ: unknown={sorted(unknown)} missing={sorted(missing)}"
)
if self.document["schema_version"] != SCHEMA_VERSION:
raise ExportBoundaryError("POLICY_INVALID", "unsupported export policy schema")
if self.document["default_export"] != "deny":
raise ExportBoundaryError("POLICY_INVALID", "default_export must be deny")
if not isinstance(self.document.get("max_source_bytes"), int) or self.max_source_bytes < 1:
raise ExportBoundaryError("POLICY_INVALID", "max_source_bytes must be a positive integer")
if not isinstance(self.classifications, dict) or not self.classifications:
raise ExportBoundaryError("POLICY_INVALID", "classifications must be a non-empty mapping")
if self.classifications.get("secret_reference", {}).get("exportable") is not False:
raise ExportBoundaryError("POLICY_INVALID", "secret_reference must not be exportable")
public_classes = self.document["public_output_classifications"]
if not isinstance(public_classes, list) or set(public_classes) - set(self.classifications):
raise ExportBoundaryError("POLICY_INVALID", "public output classifications are invalid")
if not isinstance(self.transforms, dict) or not isinstance(self.artifact_kinds, dict):
raise ExportBoundaryError("POLICY_INVALID", "transforms and artifact_kinds must be mappings")
for kind, descriptor in self.artifact_kinds.items():
if not isinstance(descriptor, dict):
raise ExportBoundaryError("POLICY_INVALID", f"artifact kind {kind} must be a mapping")
normalize_relative_path(descriptor.get("output_prefix"), f"artifact_kinds.{kind}.output_prefix")
transform_ids = descriptor.get("allowed_transforms")
if not isinstance(transform_ids, list) or not transform_ids:
raise ExportBoundaryError("POLICY_INVALID", f"artifact kind {kind} needs transforms")
if set(transform_ids) - set(self.transforms):
raise ExportBoundaryError("POLICY_INVALID", f"artifact kind {kind} references unknown transform")
for transform_id, descriptor in self.transforms.items():
if not isinstance(descriptor, dict):
raise ExportBoundaryError("POLICY_INVALID", f"transform {transform_id} must be a mapping")
source_classes = descriptor.get("source_classifications")
if not isinstance(source_classes, list) or set(source_classes) - set(self.classifications):
raise ExportBoundaryError("POLICY_INVALID", f"transform {transform_id} has invalid source classes")
for field_name in ["required_source_fields", "allowed_source_fields", "output_fields"]:
fields = descriptor.get(field_name)
if not isinstance(fields, list) or not all(isinstance(field, str) for field in fields):
raise ExportBoundaryError("POLICY_INVALID", f"transform {transform_id}.{field_name} is invalid")
if len(fields) != len(set(fields)):
raise ExportBoundaryError(
"POLICY_INVALID", f"transform {transform_id}.{field_name} contains duplicates"
)
if set(descriptor["required_source_fields"]) - set(descriptor["allowed_source_fields"]):
raise ExportBoundaryError(
"POLICY_INVALID", f"transform {transform_id} required fields are not allowed"
)
required_output = {
"schema_version",
"id",
"kind",
"classification",
"retention",
} | set(descriptor["required_source_fields"])
if required_output - set(descriptor["output_fields"]):
raise ExportBoundaryError("POLICY_INVALID", f"transform {transform_id} omits required output fields")
if set(descriptor["output_fields"]) & set(self.document["forbidden_public_fields"]):
raise ExportBoundaryError("POLICY_INVALID", f"transform {transform_id} emits forbidden fields")
def validate_route(self, source_classification: str, artifact_kind: str, transform_id: str) -> None:
classification = self.classifications.get(source_classification)
if not isinstance(classification, dict) or classification.get("exportable") is not True:
raise ExportBoundaryError("CLASSIFICATION_DENIED", "source classification cannot be exported")
kind = self.artifact_kinds.get(artifact_kind)
if not isinstance(kind, dict):
raise ExportBoundaryError("UNKNOWN_ARTIFACT_KIND", "artifact kind is not allowlisted")
if transform_id not in kind["allowed_transforms"]:
raise ExportBoundaryError("UNAPPROVED_TRANSFORM", "transform is not allowed for artifact kind")
transform = self.transforms.get(transform_id)
if not isinstance(transform, dict) or source_classification not in transform["source_classifications"]:
raise ExportBoundaryError("CLASSIFICATION_DENIED", "transform is not allowed for source class")
REQUEST_FIELDS = {"schema_version", "id", "kind", "classification", "retention", "entries"}
REQUEST_ENTRY_FIELDS = {
"entry_id",
"projection_id",
"source_path",
"source_classification",
"artifact_kind",
"transformation_id",
"output_path",
}
PLAN_FIELDS = {
"schema_version",
"id",
"kind",
"state",
"classification",
"retention",
"policy_version",
"policy_digest",
"request_digest",
"entries",
}
PLAN_ENTRY_FIELDS = REQUEST_ENTRY_FIELDS | {"source_digest", "source_size_bytes"}
def _parse_json_object(body: bytes, location: str) -> dict[str, Any]:
try:
value = parse_json_text(body.decode("utf-8"))
except (UnicodeError, json.JSONDecodeError, DuplicateKeyError) as exc:
raise ExportBoundaryError("SOURCE_INVALID", f"{location} must be strict UTF-8 JSON") from exc
if not isinstance(value, dict):
raise ExportBoundaryError("SOURCE_INVALID", f"{location} must be a JSON object")
return value
def _validate_ids_and_paths(entries: list[dict[str, Any]], policy: ExportPolicy) -> None:
entry_ids: set[str] = set()
projection_ids: set[str] = set()
output_paths: set[str] = set()
collision_keys: set[str] = set()
for index, entry in enumerate(entries):
entry_id = _require_nonempty_string(entry.get("entry_id"), f"entries[{index}].entry_id")
projection_id = _require_nonempty_string(
entry.get("projection_id"), f"entries[{index}].projection_id"
)
if not ENTRY_ID.fullmatch(entry_id) or not PROJECTION_ID.fullmatch(projection_id):
raise ExportBoundaryError("INVALID_ID", "export entry or projection ID has invalid form")
if entry_id in entry_ids or projection_id in projection_ids:
raise ExportBoundaryError("DUPLICATE_ID", "export entry and projection IDs must be unique")
entry_ids.add(entry_id)
projection_ids.add(projection_id)
output_path = normalize_relative_path(entry.get("output_path"), f"entries[{index}].output_path")
if output_path == "export-manifest.json":
raise ExportBoundaryError("PATH_COLLISION", "output path collides with export manifest")
collision_key = unicodedata.normalize("NFC", output_path).casefold()
if output_path in output_paths or collision_key in collision_keys:
raise ExportBoundaryError("PATH_COLLISION", "output paths collide after normalization")
output_paths.add(output_path)
collision_keys.add(collision_key)
kind = entry.get("artifact_kind")
if kind not in policy.artifact_kinds:
raise ExportBoundaryError("UNKNOWN_ARTIFACT_KIND", "artifact kind is not allowlisted")
prefix = policy.artifact_kinds[kind]["output_prefix"]
if not output_path.startswith(prefix):
raise ExportBoundaryError("UNAPPROVED_DESTINATION", "output path is outside artifact prefix")
if not output_path.endswith(".json"):
raise ExportBoundaryError("UNAPPROVED_DESTINATION", "current transforms require .json output")
def plan_export(request: Mapping[str, Any], private_root: Path, policy: ExportPolicy) -> dict[str, Any]:
_require_exact_keys(request, REQUEST_FIELDS, "ExportRequest")
if request["schema_version"] != SCHEMA_VERSION or request["kind"] != "ExportRequest":
raise ExportBoundaryError("UNKNOWN_SCHEMA", "unsupported ExportRequest version or kind")
export_id = _require_nonempty_string(request["id"], "ExportRequest.id")
if not EXPORT_ID.fullmatch(export_id):
raise ExportBoundaryError("INVALID_ID", "ExportRequest.id has invalid form")
if request["classification"] not in {"private_operational", "private_human"}:
raise ExportBoundaryError("CLASSIFICATION_DENIED", "ExportRequest must remain private")
_require_nonempty_string(request["retention"], "ExportRequest.retention")
raw_entries = request["entries"]
if not isinstance(raw_entries, list) or not raw_entries:
raise ExportBoundaryError("UNKNOWN_FIELD", "ExportRequest.entries must be non-empty")
entries: list[dict[str, Any]] = []
for index, raw_entry in enumerate(raw_entries):
if not isinstance(raw_entry, dict):
raise ExportBoundaryError("UNKNOWN_FIELD", f"entries[{index}] must be an object")
_require_exact_keys(raw_entry, REQUEST_ENTRY_FIELDS, f"entries[{index}]")
source_path = normalize_relative_path(raw_entry["source_path"], f"entries[{index}].source_path")
source_class = _require_nonempty_string(
raw_entry["source_classification"], f"entries[{index}].source_classification"
)
artifact_kind = _require_nonempty_string(
raw_entry["artifact_kind"], f"entries[{index}].artifact_kind"
)
transform_id = _require_nonempty_string(
raw_entry["transformation_id"], f"entries[{index}].transformation_id"
)
policy.validate_route(source_class, artifact_kind, transform_id)
stable = read_stable_file(private_root, source_path, policy.max_source_bytes)
source = _parse_json_object(stable.body, f"entries[{index}].source")
transform = policy.transforms[transform_id]
missing_source = set(transform["required_source_fields"]) - set(source)
if missing_source:
raise ExportBoundaryError(
"SOURCE_INVALID", f"entries[{index}] lacks required fields: {sorted(missing_source)}"
)
entries.append(
{
**raw_entry,
"source_path": source_path,
"output_path": normalize_relative_path(
raw_entry["output_path"], f"entries[{index}].output_path"
),
"source_digest": stable.digest,
"source_size_bytes": stable.size_bytes,
}
)
_validate_ids_and_paths(entries, policy)
return {
"schema_version": SCHEMA_VERSION,
"id": export_id,
"kind": "ExportPlan",
"state": "planned",
"classification": "private_operational",
"retention": "review_cycle",
"policy_version": policy.version,
"policy_digest": policy.digest,
"request_digest": record_digest(dict(request)),
"entries": entries,
}
def validate_plan(plan: Mapping[str, Any], policy: ExportPolicy) -> None:
_require_exact_keys(plan, PLAN_FIELDS, "ExportPlan")
if (
plan["schema_version"] != SCHEMA_VERSION
or plan["kind"] != "ExportPlan"
or plan["state"] != "planned"
):
raise ExportBoundaryError("UNKNOWN_SCHEMA", "unsupported ExportPlan")
if plan["classification"] != "private_operational":
raise ExportBoundaryError("CLASSIFICATION_DENIED", "ExportPlan must remain private")
if plan["policy_version"] != policy.version or plan["policy_digest"] != policy.digest:
raise ExportBoundaryError("POLICY_MISMATCH", "ExportPlan is pinned to another policy")
if not EXPORT_ID.fullmatch(_require_nonempty_string(plan["id"], "ExportPlan.id")):
raise ExportBoundaryError("INVALID_ID", "ExportPlan.id has invalid form")
entries = plan["entries"]
if not isinstance(entries, list) or not entries:
raise ExportBoundaryError("UNKNOWN_FIELD", "ExportPlan.entries must be non-empty")
for index, entry in enumerate(entries):
if not isinstance(entry, dict):
raise ExportBoundaryError("UNKNOWN_FIELD", f"entries[{index}] must be an object")
_require_exact_keys(entry, PLAN_ENTRY_FIELDS, f"entries[{index}]")
policy.validate_route(
entry["source_classification"], entry["artifact_kind"], entry["transformation_id"]
)
if not re.fullmatch(r"sha256:[0-9a-f]{64}", str(entry["source_digest"])):
raise ExportBoundaryError("UNKNOWN_FIELD", "source_digest is invalid")
if not isinstance(entry["source_size_bytes"], int) or entry["source_size_bytes"] < 0:
raise ExportBoundaryError("UNKNOWN_FIELD", "source_size_bytes is invalid")
_validate_ids_and_paths(entries, policy)
def _validate_source_field_types(source: Mapping[str, Any], fields: Iterable[str]) -> None:
list_fields = {"authors", "evidence_links"}
for field in fields:
if field not in source:
continue
value = source[field]
if field in list_fields:
if not isinstance(value, list) or not all(isinstance(item, str) for item in value):
raise ExportBoundaryError("SOURCE_INVALID", f"{field} must be a string list")
elif not isinstance(value, str):
raise ExportBoundaryError("SOURCE_INVALID", f"{field} must be a string")
def project_source(
source: Mapping[str, Any], projection_id: str, artifact_kind: str, transform: Mapping[str, Any]
) -> dict[str, Any]:
required = set(transform["required_source_fields"])
if required - set(source):
raise ExportBoundaryError("SOURCE_INVALID", "source lacks a required projection field")
_validate_source_field_types(source, transform["allowed_source_fields"])
kind_name = {"citation_metadata": "CitationMetadata", "research_summary": "ResearchSummary"}.get(
artifact_kind
)
if kind_name is None:
raise ExportBoundaryError("UNKNOWN_ARTIFACT_KIND", "projector does not implement artifact kind")
projection: dict[str, Any] = {
"schema_version": SCHEMA_VERSION,
"id": projection_id,
"kind": kind_name,
"classification": "public_derived",
"retention": "durable",
}
for field in transform["allowed_source_fields"]:
if field in source:
projection[field] = source[field]
if set(projection) - set(transform["output_fields"]):
raise ExportBoundaryError("POLICY_INVALID", "projector emitted a field absent from policy")
return projection
def canary_variants(canaries: Iterable[str]) -> list[bytes]:
variants: set[bytes] = set()
for canary in canaries:
if not canary:
continue
raw = canary.encode("utf-8")
variants.add(raw)
variants.add(raw.hex().encode("ascii"))
variants.add(base64.b64encode(raw))
variants.add(base64.urlsafe_b64encode(raw).rstrip(b"="))
return sorted(variants)
def _scan_public_value(value: Any, policy: ExportPolicy, location: str) -> None:
if isinstance(value, dict):
forbidden = set(policy.document["forbidden_public_fields"])
allowed_digests = set(policy.document["allowed_public_digest_fields"])
for key, child in value.items():
if key in forbidden or (key.endswith("digest") and key not in allowed_digests):
raise ExportBoundaryError("PRIVATE_METADATA", f"forbidden public field at {location}")
if key == "classification" and child not in policy.document["public_output_classifications"]:
raise ExportBoundaryError("CLASSIFICATION_DENIED", f"non-public classification at {location}")
if key == "kind" and child in {"CredentialRef", "CapabilityToken", "StorageBinding"}:
raise ExportBoundaryError("PRIVATE_METADATA", f"private record kind at {location}")
_scan_public_value(child, policy, f"{location}.{key}")
elif isinstance(value, list):
for index, child in enumerate(value):
_scan_public_value(child, policy, f"{location}[{index}]")
elif isinstance(value, str):
scan_supplementary_text(value, policy)
def scan_supplementary_text(text: str, policy: ExportPolicy) -> None:
detectors = policy.document["supplementary_detectors"]
patterns = [
("pem_private_key", PRIVATE_KEY_PATTERN),
("authorization_header", AUTHORIZATION_PATTERN),
("signed_url", SIGNED_URL_PATTERN),
("private_absolute_path", PRIVATE_PATH_PATTERN),
]
for detector, pattern in patterns:
if detectors.get(detector) and pattern.search(text):
raise ExportBoundaryError("SECRET_PATTERN", f"supplementary detector triggered: {detector}")
def scan_public_bytes(
body: bytes,
policy: ExportPolicy,
*,
canaries: Iterable[str] = (),
media_type: str = "application/json",
) -> Any:
for variant in canary_variants(canaries):
if variant and variant in body:
raise ExportBoundaryError("SECRET_CANARY", "seeded private canary detected")
try:
text = body.decode("utf-8")
except UnicodeError as exc:
raise ExportBoundaryError("OUTPUT_INVALID", "public output must be UTF-8") from exc
scan_supplementary_text(text, policy)
if media_type == "application/json":
try:
value = parse_json_text(text)
except (json.JSONDecodeError, DuplicateKeyError) as exc:
raise ExportBoundaryError("OUTPUT_INVALID", "public JSON is invalid") from exc
_scan_public_value(value, policy, "public")
return value
return text
def _write_staged_file(root: Path, relative: str, body: bytes) -> None:
target = root.joinpath(*PurePosixPath(relative).parts)
target.parent.mkdir(parents=True, exist_ok=True)
with target.open("xb") as handle:
handle.write(body)
handle.flush()
os.fsync(handle.fileno())
def render_export(
plan: Mapping[str, Any],
private_root: Path,
staging_directory: Path,
policy: ExportPolicy,
*,
canaries: Iterable[str] = (),
) -> tuple[dict[str, Any], dict[str, Any]]:
staging_directory = staging_directory.absolute()
staging_directory.parent.mkdir(parents=True, exist_ok=True)
reservation = staging_directory.parent / f".{staging_directory.name}.render-lock"
try:
reservation.mkdir(mode=0o700)
except FileExistsError as exc:
raise ExportBoundaryError("OUTPUT_BUSY", "another render owns the staging reservation") from exc
try:
return _render_export_reserved(
plan,
private_root,
staging_directory,
policy,
canaries=canaries,
)
finally:
try:
reservation.rmdir()
except FileNotFoundError:
pass
def _render_export_reserved(
plan: Mapping[str, Any],
private_root: Path,
staging_directory: Path,
policy: ExportPolicy,
*,
canaries: Iterable[str] = (),
) -> tuple[dict[str, Any], dict[str, Any]]:
validate_plan(plan, policy)
if staging_directory.exists():
raise ExportBoundaryError("OUTPUT_EXISTS", "staging directory must not already exist")
temporary = Path(tempfile.mkdtemp(prefix=f".{staging_directory.name}.", dir=staging_directory.parent))
manifest_entries: list[dict[str, Any]] = []
receipt_sources: list[dict[str, Any]] = []
try:
for index, entry in enumerate(plan["entries"]):
stable = read_stable_file(private_root, entry["source_path"], policy.max_source_bytes)
if stable.digest != entry["source_digest"] or stable.size_bytes != entry["source_size_bytes"]:
raise ExportBoundaryError("SOURCE_CHANGED", f"source for entry {index} changed after planning")
source = _parse_json_object(stable.body, f"entries[{index}].source")
transform = policy.transforms[entry["transformation_id"]]
projection = project_source(source, entry["projection_id"], entry["artifact_kind"], transform)
body = pretty_json(projection).encode("utf-8")
scan_public_bytes(body, policy, canaries=canaries)
output_digest = sha256_bytes(body)
_write_staged_file(temporary, entry["output_path"], body)
manifest_entries.append(
{
"entry_id": entry["entry_id"],
"projection_id": entry["projection_id"],
"artifact_kind": entry["artifact_kind"],
"transformation_id": entry["transformation_id"],
"transformation_digest": record_digest(transform),
"output_path": entry["output_path"],
"output_digest": output_digest,
"media_type": transform["output_media_type"],
"size_bytes": len(body),
}
)
receipt_sources.append(
{
"entry_id": entry["entry_id"],
"source_digest": stable.digest,
"source_size_bytes": stable.size_bytes,
}
)
manifest = {
"schema_version": SCHEMA_VERSION,
"id": plan["id"],
"kind": "ExportManifest",
"state": "rendered",
"classification": "public_derived",
"retention": "durable",
"policy_version": policy.version,
"policy_digest": policy.digest,
"entries": sorted(manifest_entries, key=lambda item: item["entry_id"]),
}
manifest_body = pretty_json(manifest).encode("utf-8")
scan_public_bytes(manifest_body, policy, canaries=canaries)
_write_staged_file(temporary, "export-manifest.json", manifest_body)
check_staging(temporary, policy, canaries=canaries)
receipt = {
"schema_version": SCHEMA_VERSION,
"id": plan["id"],
"kind": "ExportRenderReceipt",
"state": "rendered",
"classification": "private_operational",
"retention": "review_cycle",
"policy_version": policy.version,
"policy_digest": policy.digest,
"plan_digest": record_digest(dict(plan)),
"manifest_digest": sha256_bytes(manifest_body),
"sources": receipt_sources,
}
os.replace(temporary, staging_directory)
if hasattr(os, "O_DIRECTORY"):
directory_fd = os.open(staging_directory.parent, os.O_RDONLY | os.O_DIRECTORY)
try:
os.fsync(directory_fd)
finally:
os.close(directory_fd)
return manifest, receipt
except Exception:
shutil.rmtree(temporary, ignore_errors=True)
raise
def validate_public_manifest(manifest: Mapping[str, Any], policy: ExportPolicy) -> None:
_require_exact_keys(manifest, set(policy.document["public_manifest_fields"]), "ExportManifest")
if (
manifest["schema_version"] != SCHEMA_VERSION
or manifest["kind"] != "ExportManifest"
or manifest["state"] != "rendered"
or manifest["classification"] != "public_derived"
or manifest["retention"] != "durable"
):
raise ExportBoundaryError("UNKNOWN_SCHEMA", "unsupported ExportManifest")
if manifest["policy_version"] != policy.version or manifest["policy_digest"] != policy.digest:
raise ExportBoundaryError("POLICY_MISMATCH", "manifest policy pin does not match")
if not EXPORT_ID.fullmatch(_require_nonempty_string(manifest["id"], "ExportManifest.id")):
raise ExportBoundaryError("INVALID_ID", "ExportManifest.id has invalid form")
entries = manifest["entries"]
if not isinstance(entries, list) or not entries:
raise ExportBoundaryError("UNKNOWN_FIELD", "ExportManifest.entries must be non-empty")
seen_entries: set[str] = set()
seen_projections: set[str] = set()
collision_keys: set[str] = set()
for index, entry in enumerate(entries):
if not isinstance(entry, dict):
raise ExportBoundaryError("UNKNOWN_FIELD", f"entries[{index}] must be an object")
_require_exact_keys(
entry, set(policy.document["public_manifest_entry_fields"]), f"entries[{index}]"
)
entry_id = entry["entry_id"]
if not isinstance(entry_id, str) or not ENTRY_ID.fullmatch(entry_id) or entry_id in seen_entries:
raise ExportBoundaryError("DUPLICATE_ID", "manifest entry IDs must be unique and valid")
seen_entries.add(entry_id)
projection_id = entry["projection_id"]
if (
not isinstance(projection_id, str)
or not PROJECTION_ID.fullmatch(projection_id)
or projection_id in seen_projections
):
raise ExportBoundaryError("DUPLICATE_ID", "manifest projection IDs must be unique and valid")
seen_projections.add(projection_id)
output_path = normalize_relative_path(entry["output_path"], f"entries[{index}].output_path")
collision_key = unicodedata.normalize("NFC", output_path).casefold()
if collision_key in collision_keys:
raise ExportBoundaryError("PATH_COLLISION", "manifest outputs collide after normalization")
collision_keys.add(collision_key)
kind = policy.artifact_kinds.get(entry["artifact_kind"])
transform = policy.transforms.get(entry["transformation_id"])
if not isinstance(kind, dict):
raise ExportBoundaryError("UNKNOWN_ARTIFACT_KIND", "manifest artifact kind is not allowlisted")
if not isinstance(transform, dict) or entry["transformation_id"] not in kind["allowed_transforms"]:
raise ExportBoundaryError("UNAPPROVED_TRANSFORM", "manifest transform is not allowlisted")
prefix = kind["output_prefix"]
if not output_path.startswith(prefix):
raise ExportBoundaryError("UNAPPROVED_DESTINATION", "manifest output is outside artifact prefix")
for digest_field in ["transformation_digest", "output_digest"]:
if not re.fullmatch(r"sha256:[0-9a-f]{64}", str(entry[digest_field])):
raise ExportBoundaryError("UNKNOWN_FIELD", f"{digest_field} is invalid")
if entry["transformation_digest"] != record_digest(transform):
raise ExportBoundaryError("POLICY_MISMATCH", "transformation digest is stale")
if entry["media_type"] != transform["output_media_type"]:
raise ExportBoundaryError("UNKNOWN_FIELD", "manifest media type disagrees with transform")
if not isinstance(entry["size_bytes"], int) or entry["size_bytes"] < 0:
raise ExportBoundaryError("UNKNOWN_FIELD", "manifest byte size is invalid")
def check_staging(
staging_directory: Path,
policy: ExportPolicy,
*,
canaries: Iterable[str] = (),
) -> dict[str, Any]:
if not staging_directory.is_dir() or staging_directory.is_symlink():
raise ExportBoundaryError("PATH_ESCAPE", "staging directory must be a regular directory")
manifest_path = staging_directory / "export-manifest.json"
if not manifest_path.is_file() or manifest_path.is_symlink():
raise ExportBoundaryError("MISSING_OUTPUT", "export-manifest.json is missing")
try:
manifest_stable = read_stable_file(
staging_directory, "export-manifest.json", policy.max_source_bytes
)
manifest_body = manifest_stable.body
manifest = parse_json_text(manifest_body.decode("utf-8"))
except (OSError, UnicodeError, json.JSONDecodeError, DuplicateKeyError) as exc:
raise ExportBoundaryError("OUTPUT_INVALID", "export manifest is invalid") from exc
if not isinstance(manifest, dict):
raise ExportBoundaryError("OUTPUT_INVALID", "export manifest must be an object")
validate_public_manifest(manifest, policy)
scan_public_bytes(manifest_body, policy, canaries=canaries)
expected = {"export-manifest.json"}
expected.update(entry["output_path"] for entry in manifest["entries"])
actual: set[str] = set()
collision_keys: set[str] = set()
for path in sorted(staging_directory.rglob("*")):
if path.is_symlink():
raise ExportBoundaryError("PATH_ESCAPE", "staged tree contains a symlink")
if path.is_dir():
continue
if not path.is_file():
raise ExportBoundaryError("PATH_ESCAPE", "staged tree contains a non-regular entry")
if path.stat().st_nlink != 1:
raise ExportBoundaryError("PATH_ESCAPE", "staged tree contains a hardlinked file")
relative = path.relative_to(staging_directory).as_posix()
normalized = normalize_relative_path(relative, "staged output")
collision_key = unicodedata.normalize("NFC", normalized).casefold()
if collision_key in collision_keys:
raise ExportBoundaryError("PATH_COLLISION", "staged paths collide after normalization")
collision_keys.add(collision_key)
actual.add(normalized)
extra = actual - expected
missing = expected - actual
if extra:
raise ExportBoundaryError("UNAPPROVED_OUTPUT", f"staging contains {len(extra)} extra file(s)")
if missing:
raise ExportBoundaryError("MISSING_OUTPUT", f"staging lacks {len(missing)} file(s)")
checked_outputs: list[dict[str, Any]] = []
for entry in manifest["entries"]:
path = staging_directory.joinpath(*PurePosixPath(entry["output_path"]).parts)
stable = read_stable_file(staging_directory, entry["output_path"], policy.max_source_bytes)
body = stable.body
if stable.digest != entry["output_digest"] or stable.size_bytes != entry["size_bytes"]:
raise ExportBoundaryError("DIGEST_MISMATCH", "staged output does not match manifest")
value = scan_public_bytes(body, policy, canaries=canaries, media_type=entry["media_type"])
if not isinstance(value, dict):
raise ExportBoundaryError("OUTPUT_INVALID", "current projected output must be an object")
transform = policy.transforms[entry["transformation_id"]]
allowed = set(transform["output_fields"])
if set(value) - allowed:
raise ExportBoundaryError("UNKNOWN_FIELD", "projected output has fields absent from transform")
base_required = {"schema_version", "id", "kind", "classification", "retention"}
required = base_required | set(transform["required_source_fields"])
if required - set(value):
raise ExportBoundaryError("UNKNOWN_FIELD", "projected output lacks required fields")
if value.get("id") != entry["projection_id"]:
raise ExportBoundaryError("DIGEST_MISMATCH", "projection ID disagrees with manifest")
checked_outputs.append(
{
"entry_id": entry["entry_id"],
"output_path": entry["output_path"],
"output_digest": entry["output_digest"],
}
)
return {
"schema_version": SCHEMA_VERSION,
"id": manifest["id"],
"kind": "ExportValidation",
"state": "validated",
"classification": "public_derived",
"retention": "durable",
"policy_version": policy.version,
"policy_digest": policy.digest,
"manifest_digest": sha256_bytes(manifest_body),
"outputs": checked_outputs,
"result": "pass",
}
def ensure_private_record_path(path: Path, repository_root: Path = ROOT) -> None:
absolute = path.absolute()
try:
relative = absolute.relative_to(repository_root.resolve())
except ValueError:
return
allowed = [PurePosixPath("Private"), PurePosixPath("researcher/exports/private")]
relative_posix = PurePosixPath(relative.as_posix())
if not any(relative_posix == prefix or prefix in relative_posix.parents for prefix in allowed):
raise ExportBoundaryError(
"PRIVATE_DESTINATION_REQUIRED",
"private export records inside the repository must use Private/ or researcher/exports/private/",
)
def write_private_json(path: Path, value: Mapping[str, Any]) -> None:
ensure_private_record_path(path)
atomic_write_text(path, pretty_json(dict(value)))
try:
path.chmod(0o600)
except OSError:
pass
@@ -39,6 +39,13 @@ def copy_fixture(source: Path, target: Path) -> None:
"researcher/claims/index.jsonl",
"researcher/corpus/index.json",
"researcher/corpus/inventory.schema.json",
"governance/export-policy.yaml",
"governance/export-policy.schema.json",
"researcher/exports/schemas/export-records.schema.json",
"researcher/fixtures/export/restricted-request.json",
"researcher/fixtures/export/private-root/restricted-source.json",
"researcher/exports/examples/restricted-citation-v1/export-manifest.json",
"researcher/exports/examples/restricted-citation-v1/citation/restricted-fixture.json",
"researcher/fixtures/activation-cases.jsonl",
"researcher/benchmarks/router/prompts.jsonl",
"researcher/benchmarks/scenarios/adversarial.jsonl",
@@ -51,6 +58,8 @@ def copy_fixture(source: Path, target: Path) -> None:
"researcher/benchmarks/sdk-runner/src/runEffectiveness.ts",
"researcher/scripts/validate_governance.py",
"researcher/scripts/build_inventory.py",
"researcher/scripts/validate_export.py",
"researcher/scripts/export_policy.py",
"researcher/scripts/validate_platform_compat.py",
"researcher/scripts/validate_repo.py",
"researcher/scripts/skill_health.py",
@@ -0,0 +1,410 @@
"""Adversarial tests for the allowlisted public export boundary."""
from __future__ import annotations
import base64
import copy
import json
import os
import tempfile
import unittest
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from researcher.scripts.build_inventory import pretty_json, record_digest, sha256_bytes
from researcher.scripts.export_policy import (
DEFAULT_POLICY_PATH,
ExportBoundaryError,
ExportPolicy,
check_staging,
ensure_private_record_path,
plan_export,
render_export,
)
ROOT = Path(__file__).resolve().parents[3]
def citation_source(**overrides: object) -> dict[str, object]:
value: dict[str, object] = {
"title": "Synthetic Restricted Harness Study",
"url": "https://example.invalid/restricted",
"source_type": "restricted_fixture",
"authors": ["Fixture Author"],
"published_at": "2026-08-10",
"license_note": "Metadata only",
"body": "SYNTHETIC_RESTRICTED_BODY_MUST_NOT_EXPORT_42",
}
value.update(overrides)
return value
def export_request(**entry_overrides: object) -> dict[str, object]:
entry: dict[str, object] = {
"entry_id": "entry_fixture",
"projection_id": "proj_fixture",
"source_path": "source.json",
"source_classification": "restricted_source",
"artifact_kind": "citation_metadata",
"transformation_id": "restricted_citation_v1",
"output_path": "citation/fixture.json",
}
entry.update(entry_overrides)
return {
"schema_version": "1.0.0",
"id": "export_fixture",
"kind": "ExportRequest",
"classification": "private_operational",
"retention": "review_cycle",
"entries": [entry],
}
class ExportBoundaryTests(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.policy = ExportPolicy.load(DEFAULT_POLICY_PATH)
def workspace(self, source: dict[str, object] | None = None) -> tuple[tempfile.TemporaryDirectory[str], Path]:
temporary = tempfile.TemporaryDirectory()
root = Path(temporary.name)
private = root / "private"
private.mkdir()
(private / "source.json").write_text(
pretty_json(source or citation_source()), encoding="utf-8"
)
return temporary, root
def assert_code(self, code: str, operation) -> None:
with self.assertRaises(ExportBoundaryError) as raised:
operation()
self.assertEqual(raised.exception.code, code)
def test_restricted_source_projects_metadata_without_raw_body(self) -> None:
temporary, root = self.workspace()
self.addCleanup(temporary.cleanup)
plan = plan_export(export_request(), root / "private", self.policy)
manifest, receipt = render_export(plan, root / "private", root / "staging", self.policy)
validation = check_staging(root / "staging", self.policy)
public_bytes = (root / "staging/citation/fixture.json").read_bytes()
manifest_bytes = (root / "staging/export-manifest.json").read_bytes()
self.assertNotIn(b"SYNTHETIC_RESTRICTED_BODY", public_bytes)
self.assertNotIn(b"source_path", manifest_bytes)
self.assertNotIn(b"source_digest", manifest_bytes)
self.assertEqual(manifest["classification"], "public_derived")
self.assertEqual(receipt["classification"], "private_operational")
self.assertIn("source_digest", receipt["sources"][0])
self.assertEqual(validation["result"], "pass")
def test_plan_is_private_and_binds_exact_source(self) -> None:
temporary, root = self.workspace()
self.addCleanup(temporary.cleanup)
plan = plan_export(export_request(), root / "private", self.policy)
self.assertEqual(plan["classification"], "private_operational")
self.assertRegex(plan["entries"][0]["source_digest"], r"^sha256:[0-9a-f]{64}$")
self.assertEqual(plan["entries"][0]["source_path"], "source.json")
def test_render_is_byte_deterministic_across_fresh_directories(self) -> None:
temporary, root = self.workspace()
self.addCleanup(temporary.cleanup)
plan = plan_export(export_request(), root / "private", self.policy)
first_manifest, first_receipt = render_export(
plan, root / "private", root / "staging-one", self.policy
)
second_manifest, second_receipt = render_export(
plan, root / "private", root / "staging-two", self.policy
)
self.assertEqual(first_manifest, second_manifest)
self.assertEqual(first_receipt, second_receipt)
for relative in ["export-manifest.json", "citation/fixture.json"]:
self.assertEqual(
(root / "staging-one" / relative).read_bytes(),
(root / "staging-two" / relative).read_bytes(),
)
def test_secret_reference_source_is_denied(self) -> None:
temporary, root = self.workspace()
self.addCleanup(temporary.cleanup)
request = export_request(source_classification="secret_reference")
self.assert_code(
"CLASSIFICATION_DENIED",
lambda: plan_export(request, root / "private", self.policy),
)
def test_unknown_request_field_is_denied(self) -> None:
temporary, root = self.workspace()
self.addCleanup(temporary.cleanup)
request = export_request()
request["ambient_authority"] = True
self.assert_code("UNKNOWN_FIELD", lambda: plan_export(request, root / "private", self.policy))
def test_unknown_entry_field_is_denied(self) -> None:
temporary, root = self.workspace()
self.addCleanup(temporary.cleanup)
request = export_request()
request["entries"][0]["credential"] = "fixture"
self.assert_code("UNKNOWN_FIELD", lambda: plan_export(request, root / "private", self.policy))
def test_source_path_traversal_is_denied(self) -> None:
temporary, root = self.workspace()
self.addCleanup(temporary.cleanup)
request = export_request(source_path="../source.json")
self.assert_code("PATH_ESCAPE", lambda: plan_export(request, root / "private", self.policy))
def test_output_path_traversal_is_denied(self) -> None:
temporary, root = self.workspace()
self.addCleanup(temporary.cleanup)
request = export_request(output_path="citation/../../escape.json")
self.assert_code("PATH_ESCAPE", lambda: plan_export(request, root / "private", self.policy))
def test_casefolded_output_collision_is_denied(self) -> None:
temporary, root = self.workspace()
self.addCleanup(temporary.cleanup)
request = export_request()
second = copy.deepcopy(request["entries"][0])
second.update(
{
"entry_id": "entry_second",
"projection_id": "proj_second",
"output_path": "citation/FIXTURE.json",
}
)
request["entries"].append(second)
self.assert_code("PATH_COLLISION", lambda: plan_export(request, root / "private", self.policy))
@unittest.skipIf(not hasattr(os, "symlink"), "symlink support required")
def test_symlink_source_is_denied(self) -> None:
temporary, root = self.workspace()
self.addCleanup(temporary.cleanup)
source = root / "private/source.json"
real = root / "real.json"
source.replace(real)
source.symlink_to(real)
self.assert_code(
"PATH_ESCAPE",
lambda: plan_export(export_request(), root / "private", self.policy),
)
def test_source_change_after_plan_is_denied(self) -> None:
temporary, root = self.workspace()
self.addCleanup(temporary.cleanup)
plan = plan_export(export_request(), root / "private", self.policy)
(root / "private/source.json").write_text(
pretty_json(citation_source(title="Changed")), encoding="utf-8"
)
self.assert_code(
"SOURCE_CHANGED",
lambda: render_export(plan, root / "private", root / "staging", self.policy),
)
self.assertFalse((root / "staging").exists())
def test_existing_staging_directory_is_denied(self) -> None:
temporary, root = self.workspace()
self.addCleanup(temporary.cleanup)
plan = plan_export(export_request(), root / "private", self.policy)
(root / "staging").mkdir()
self.assert_code(
"OUTPUT_EXISTS",
lambda: render_export(plan, root / "private", root / "staging", self.policy),
)
def test_concurrent_render_has_one_winner_and_one_typed_denial(self) -> None:
temporary, root = self.workspace()
self.addCleanup(temporary.cleanup)
plan = plan_export(export_request(), root / "private", self.policy)
def attempt() -> str:
try:
render_export(plan, root / "private", root / "staging", self.policy)
return "success"
except ExportBoundaryError as exc:
return exc.code
with ThreadPoolExecutor(max_workers=2) as executor:
results = list(executor.map(lambda _: attempt(), range(2)))
self.assertEqual(results.count("success"), 1)
self.assertEqual(len(set(results) & {"OUTPUT_BUSY", "OUTPUT_EXISTS"}), 1)
def test_extra_staged_file_is_denied(self) -> None:
temporary, root = self.workspace()
self.addCleanup(temporary.cleanup)
plan = plan_export(export_request(), root / "private", self.policy)
render_export(plan, root / "private", root / "staging", self.policy)
(root / "staging/extra.txt").write_text("extra", encoding="utf-8")
self.assert_code("UNAPPROVED_OUTPUT", lambda: check_staging(root / "staging", self.policy))
def test_tampered_output_is_denied(self) -> None:
temporary, root = self.workspace()
self.addCleanup(temporary.cleanup)
plan = plan_export(export_request(), root / "private", self.policy)
render_export(plan, root / "private", root / "staging", self.policy)
output = root / "staging/citation/fixture.json"
output.write_text(output.read_text(encoding="utf-8") + "\n", encoding="utf-8")
self.assert_code("DIGEST_MISMATCH", lambda: check_staging(root / "staging", self.policy))
def test_private_manifest_field_is_denied(self) -> None:
temporary, root = self.workspace()
self.addCleanup(temporary.cleanup)
plan = plan_export(export_request(), root / "private", self.policy)
render_export(plan, root / "private", root / "staging", self.policy)
manifest_path = root / "staging/export-manifest.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
manifest["source_digest"] = plan["entries"][0]["source_digest"]
manifest_path.write_text(pretty_json(manifest), encoding="utf-8")
self.assert_code("UNKNOWN_FIELD", lambda: check_staging(root / "staging", self.policy))
def test_plain_hex_and_base64_canaries_are_denied(self) -> None:
canary = "synthetic-canary-9d8547"
variants = [canary, canary.encode().hex(), base64.b64encode(canary.encode()).decode()]
for index, variant in enumerate(variants):
with self.subTest(index=index):
source = {
"title": "Private review",
"summary": variant,
"reason_code": "FIXTURE",
}
temporary, root = self.workspace(source)
try:
request = export_request(
source_classification="private_human",
artifact_kind="research_summary",
transformation_id="research_summary_v1",
output_path=f"research-summary/fixture-{index}.json",
)
plan = plan_export(request, root / "private", self.policy)
self.assert_code(
"SECRET_CANARY",
lambda: render_export(
plan,
root / "private",
root / "staging",
self.policy,
canaries=[canary],
),
)
finally:
temporary.cleanup()
def test_high_confidence_secret_structures_are_denied(self) -> None:
patterns = [
"-----BEGIN PRIVATE KEY-----",
"Authorization: Bearer fixture-token",
"https://example.invalid/?X-Amz-Signature=fixture",
"/Users/private-user/Library/fixture",
]
for index, text in enumerate(patterns):
with self.subTest(index=index):
source = {"title": "Private review", "summary": text, "reason_code": "FIXTURE"}
temporary, root = self.workspace(source)
try:
request = export_request(
source_classification="private_human",
artifact_kind="research_summary",
transformation_id="research_summary_v1",
output_path=f"research-summary/pattern-{index}.json",
)
plan = plan_export(request, root / "private", self.policy)
self.assert_code(
"SECRET_PATTERN",
lambda: render_export(plan, root / "private", root / "staging", self.policy),
)
finally:
temporary.cleanup()
def test_invalid_projected_field_type_leaves_no_staging_tree(self) -> None:
temporary, root = self.workspace(citation_source(authors=7))
self.addCleanup(temporary.cleanup)
plan = plan_export(export_request(), root / "private", self.policy)
self.assert_code(
"SOURCE_INVALID",
lambda: render_export(plan, root / "private", root / "staging", self.policy),
)
self.assertFalse((root / "staging").exists())
self.assertEqual(list(root.glob(".staging.*")), [])
def test_duplicate_json_keys_in_source_are_denied(self) -> None:
temporary, root = self.workspace()
self.addCleanup(temporary.cleanup)
(root / "private/source.json").write_text(
'{"title":"a","title":"b","url":"x","source_type":"fixture"}',
encoding="utf-8",
)
self.assert_code(
"SOURCE_INVALID",
lambda: plan_export(export_request(), root / "private", self.policy),
)
@unittest.skipIf(not hasattr(os, "symlink"), "symlink support required")
def test_symlink_in_staged_tree_is_denied(self) -> None:
temporary, root = self.workspace()
self.addCleanup(temporary.cleanup)
plan = plan_export(export_request(), root / "private", self.policy)
render_export(plan, root / "private", root / "staging", self.policy)
output = root / "staging/citation/fixture.json"
outside = root / "outside.json"
output.replace(outside)
output.symlink_to(outside)
self.assert_code("PATH_ESCAPE", lambda: check_staging(root / "staging", self.policy))
@unittest.skipIf(not hasattr(os, "link"), "hardlink support required")
def test_hardlink_in_staged_tree_is_denied(self) -> None:
temporary, root = self.workspace()
self.addCleanup(temporary.cleanup)
plan = plan_export(export_request(), root / "private", self.policy)
render_export(plan, root / "private", root / "staging", self.policy)
output = root / "staging/citation/fixture.json"
outside = root / "outside.json"
os.link(output, outside)
self.assert_code("PATH_ESCAPE", lambda: check_staging(root / "staging", self.policy))
def test_policy_digest_change_invalidates_plan(self) -> None:
temporary, root = self.workspace()
self.addCleanup(temporary.cleanup)
plan = plan_export(export_request(), root / "private", self.policy)
changed_policy = ExportPolicy(
copy.deepcopy(self.policy.document),
"sha256:" + "f" * 64,
self.policy.source,
)
self.assert_code(
"POLICY_MISMATCH",
lambda: render_export(plan, root / "private", root / "staging", changed_policy),
)
def test_unknown_projected_output_field_is_denied(self) -> None:
temporary, root = self.workspace()
self.addCleanup(temporary.cleanup)
plan = plan_export(export_request(), root / "private", self.policy)
render_export(plan, root / "private", root / "staging", self.policy)
output_path = root / "staging/citation/fixture.json"
output = json.loads(output_path.read_text(encoding="utf-8"))
output["unexpected"] = "value"
output_body = pretty_json(output).encode()
output_path.write_bytes(output_body)
manifest_path = root / "staging/export-manifest.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
manifest["entries"][0]["output_digest"] = sha256_bytes(output_body)
manifest["entries"][0]["size_bytes"] = len(output_body)
manifest_path.write_text(pretty_json(manifest), encoding="utf-8")
self.assert_code("UNKNOWN_FIELD", lambda: check_staging(root / "staging", self.policy))
def test_private_record_destination_inside_repo_is_restricted(self) -> None:
self.assert_code(
"PRIVATE_DESTINATION_REQUIRED",
lambda: ensure_private_record_path(ROOT / "researcher/exports/plan.json"),
)
ensure_private_record_path(ROOT / "researcher/exports/private/plan.json")
ensure_private_record_path(ROOT / "Private/plan.json")
def test_transformation_digest_is_policy_bound(self) -> None:
transform = self.policy.transforms["restricted_citation_v1"]
temporary, root = self.workspace()
self.addCleanup(temporary.cleanup)
plan = plan_export(export_request(), root / "private", self.policy)
manifest, _ = render_export(plan, root / "private", root / "staging", self.policy)
self.assertEqual(manifest["entries"][0]["transformation_digest"], record_digest(transform))
if __name__ == "__main__":
unittest.main()
+140
View File
@@ -0,0 +1,140 @@
#!/usr/bin/env python3
"""Plan, render, and validate allowlisted public export projections."""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
from typing import Any
try:
from build_inventory import DuplicateKeyError, parse_json_text, pretty_json
from export_policy import (
DEFAULT_POLICY_PATH,
ExportBoundaryError,
ExportPolicy,
check_staging,
plan_export,
render_export,
write_private_json,
)
except ModuleNotFoundError: # Imported as researcher.scripts.validate_export.
from researcher.scripts.build_inventory import DuplicateKeyError, parse_json_text, pretty_json
from researcher.scripts.export_policy import (
DEFAULT_POLICY_PATH,
ExportBoundaryError,
ExportPolicy,
check_staging,
plan_export,
render_export,
write_private_json,
)
def load_json_object(path: Path, label: str) -> dict[str, Any]:
try:
value = parse_json_text(path.read_text(encoding="utf-8"))
except (OSError, UnicodeError, json.JSONDecodeError, DuplicateKeyError) as exc:
raise ExportBoundaryError("PARSE_ERROR", f"{label} could not be loaded") from exc
if not isinstance(value, dict):
raise ExportBoundaryError("PARSE_ERROR", f"{label} must be a JSON object")
return value
def load_canaries(path: Path | None) -> list[str]:
if path is None:
return []
try:
return [line for line in path.read_text(encoding="utf-8").splitlines() if line]
except (OSError, UnicodeError) as exc:
raise ExportBoundaryError("PARSE_ERROR", "canary file could not be loaded") from exc
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--policy", type=Path, default=DEFAULT_POLICY_PATH)
subparsers = parser.add_subparsers(dest="command", required=True)
plan_parser = subparsers.add_parser("plan", help="create an immutable private ExportPlan")
plan_parser.add_argument("--request", type=Path, required=True)
plan_parser.add_argument("--private-root", type=Path, required=True)
plan_parser.add_argument("--plan-out", type=Path, required=True)
render_parser = subparsers.add_parser("render", help="render a plan into a fresh public staging tree")
render_parser.add_argument("--plan", type=Path, required=True)
render_parser.add_argument("--private-root", type=Path, required=True)
render_parser.add_argument("--staging-dir", type=Path, required=True)
render_parser.add_argument("--receipt-out", type=Path, required=True)
render_parser.add_argument("--canary-file", type=Path)
check_parser = subparsers.add_parser("check", help="validate an existing public staging tree")
check_parser.add_argument("--staging-dir", type=Path, required=True)
check_parser.add_argument("--canary-file", type=Path)
return parser
def main() -> int:
parser = build_parser()
args = parser.parse_args()
try:
policy = ExportPolicy.load(args.policy)
if args.command == "plan":
request = load_json_object(args.request, "ExportRequest")
plan = plan_export(request, args.private_root, policy)
write_private_json(args.plan_out, plan)
print(
json.dumps(
{
"result": "planned",
"export_id": plan["id"],
"entry_count": len(plan["entries"]),
"policy_version": policy.version,
},
sort_keys=True,
)
)
return 0
if args.command == "render":
plan = load_json_object(args.plan, "ExportPlan")
canaries = load_canaries(args.canary_file)
manifest, receipt = render_export(
plan, args.private_root, args.staging_dir, policy, canaries=canaries
)
try:
write_private_json(args.receipt_out, receipt)
except Exception:
# The staging tree was created by this command and is not yet
# externally visible. Remove it if the private audit receipt
# cannot be persisted, so a supported render is never orphaned.
import shutil
shutil.rmtree(args.staging_dir, ignore_errors=True)
raise
print(
json.dumps(
{
"result": "rendered",
"export_id": manifest["id"],
"entry_count": len(manifest["entries"]),
"policy_version": policy.version,
},
sort_keys=True,
)
)
return 0
canaries = load_canaries(args.canary_file)
validation = check_staging(args.staging_dir, policy, canaries=canaries)
print(pretty_json(validation), end="")
return 0
except ExportBoundaryError as exc:
print(f"[{exc.code}] {exc.safe_message}", file=sys.stderr)
return 1
except OSError as exc:
print(f"[IO_ERROR] {exc.__class__.__name__}", file=sys.stderr)
return 1
if __name__ == "__main__":
raise SystemExit(main())