fix(sync): verify threshold human attestations

pdd/commands/sync_core.py

@@ -187,6 +187,19 @@ def _protected_playwright_command(
     return command
 
 
+def _protected_external_path(value: Path | None, option: str, cwd: Path) -> Path | None:
+    """Resolve an external protected path and reject candidate-local storage."""
+    if value is None:
+        return None
+    root = cwd.resolve()
+    resolved = value.expanduser().resolve()
+    try:
+        resolved.relative_to(root)
+    except ValueError:
+        return resolved
+    raise click.ClickException(f"{option} must be outside the candidate checkout")
+
+
 def _runner_config_from_options(
     options: dict[str, object], cwd: Path
 ) -> RunnerConfig:
@@ -195,6 +208,8 @@ def _runner_config_from_options(
     vitest_command = options.get("vitest_command")
     playwright_command = options.get("playwright_command")
     playwright_manifest = options.get("playwright_toolchain_manifest")
+    human_store = options.get("human_attestation_store")
+    human_replay = options.get("human_attestation_replay_ledger")
     return RunnerConfig(
         jest_command=_protected_command(
             jest_command if isinstance(jest_command, str) else None,
@@ -212,6 +227,16 @@ def _runner_config_from_options(
         ),
         playwright_toolchain_manifest=Path(playwright_manifest).expanduser().resolve()
         if isinstance(playwright_manifest, str) and playwright_manifest else None,
+        human_attestation_store=_protected_external_path(
+            human_store if isinstance(human_store, Path) else None,
+            "--human-attestation-store",
+            cwd,
+        ),
+        human_attestation_replay_ledger=_protected_external_path(
+            human_replay if isinstance(human_replay, Path) else None,
+            "--human-attestation-replay-ledger",
+            cwd,
+        ),
     )
 
 
@@ -467,6 +492,18 @@ def baseline(ctx: click.Context, module: str, reviewed_by: str, reason: str) ->
     type=click.Path(path_type=Path),
     help="Protected external Playwright toolchain manifest.",
 )
+@click.option(
+    "--human-attestation-store",
+    type=click.Path(path_type=Path),
+    envvar="PDD_SYNC_HUMAN_ATTESTATION_STORE",
+    help="Protected external human approval store path.",
+)
+@click.option(
+    "--human-attestation-replay-ledger",
+    type=click.Path(path_type=Path),
+    envvar="PDD_SYNC_HUMAN_ATTESTATION_REPLAY_LEDGER",
+    help="Protected external human approval replay ledger path.",
+)
 @click.pass_context
 def validate(
     ctx: click.Context,
@@ -477,26 +514,31 @@ def validate(
     vitest_command: str | None,
     playwright_command: str | None,
     playwright_toolchain_manifest: Path | None,
+    human_attestation_store: Path | None,
+    human_attestation_replay_ledger: Path | None,
 ) -> None:
     """Run protected obligations and transactionally finalize trusted evidence."""
     ctx.ensure_object(dict)
     root = Path.cwd().resolve()
+    config = _runner_config_from_options(
+        {
+            "jest_command": jest_command,
+            "vitest_command": vitest_command,
+            "playwright_command": playwright_command,
+            "playwright_toolchain_manifest": str(playwright_toolchain_manifest)
+            if playwright_toolchain_manifest else None,
+            "human_attestation_store": human_attestation_store,
+            "human_attestation_replay_ledger": human_attestation_replay_ledger,
+        },
+        root,
+    )
     result = finalize_unit(
         root,
         PurePosixPath(module),
         base_ref=base_ref,
         head_ref=head_ref,
         signer=attestation_signer_from_environment(),
-        config=_runner_config_from_options(
-            {
-                "jest_command": jest_command,
-                "vitest_command": vitest_command,
-                "playwright_command": playwright_command,
-                "playwright_toolchain_manifest": str(playwright_toolchain_manifest)
-                if playwright_toolchain_manifest else None,
-            },
-            root,
-        ),
+        config=config,
     )
     click.echo(
         json.dumps(

pdd/sync_core/__init__.py

@@ -29,6 +29,13 @@
     initialize_repository_identity,
     read_repository_identity,
 )
+from .human_attestation import (
+    HumanAttestationError,
+    HumanAttestationPolicy,
+    HumanAttestationRequest,
+    HumanAttestationVerifier,
+    load_human_attestation_policy,
+)
 from .fingerprint_store import (
     CorruptFingerprintError,
     FingerprintStore,
@@ -42,6 +49,7 @@
     attestation_signer_from_environment,
     canonical_root_for_paths,
     finalize_legacy_paths,
+    protected_runner_config_from_environment,
     finalize_unit,
 )
 from .evidence_store import (
@@ -164,6 +172,10 @@
     "FingerprintProvenance",
     "FingerprintStore",
     "FingerprintStoreError",
+    "HumanAttestationError",
+    "HumanAttestationPolicy",
+    "HumanAttestationRequest",
+    "HumanAttestationVerifier",
     "FileReplayStore",
     "FinalizeResult",
     "GlobalCertificateOptions",
@@ -227,6 +239,7 @@
     "evidence_relpath",
     "finalize_unit",
     "finalize_legacy_paths",
+    "protected_runner_config_from_environment",
     "build_include_closure",
     "build_canonical_report",
     "build_global_certificate",
@@ -240,6 +253,7 @@
     "load_verification_profiles",
     "load_attestation",
     "load_candidate_artifact_provenance",
+    "load_human_attestation_policy",
     "load_trust_policy",
     "load_sync_waivers",
     "initialize_repository_identity",

pdd/sync_core/descriptor_store.py

@@ -33,15 +33,19 @@ def _unlock(descriptor: int) -> None:
     fcntl.flock(descriptor, fcntl.LOCK_UN)
 
 
-def _safe_directory(metadata: os.stat_result) -> bool:
-    """Return whether metadata describes a private checker-owned directory."""
+def _owned_directory(metadata: os.stat_result) -> bool:
+    """Return whether metadata describes a checker-owned directory."""
     return (
         stat.S_ISDIR(metadata.st_mode)
         and (not hasattr(os, "getuid") or metadata.st_uid == os.getuid())
-        and not stat.S_IMODE(metadata.st_mode) & 0o077
     )
 
 
+def _safe_directory(metadata: os.stat_result) -> bool:
+    """Return whether metadata describes a private checker-owned directory."""
+    return _owned_directory(metadata) and not stat.S_IMODE(metadata.st_mode) & 0o077
+
+
 def _legacy_safe_parent(path: Path) -> tuple[int, os.stat_result]:
     """Preserve strict filesystem-root validation for unanchored callers."""
     flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
@@ -75,6 +79,28 @@ def _legacy_safe_parent(path: Path) -> tuple[int, os.stat_result]:
     return descriptor, opened
 
 
+def _open_trust_root(trust_root: Path, flags: int) -> int:
+    """Open a new private root or reject compromised pre-existing state."""
+    try:
+        os.lstat(trust_root)
+        existed = True
+    except FileNotFoundError:
+        existed = False
+        trust_root.mkdir(mode=0o700, parents=True, exist_ok=True)
+    descriptor = os.open(trust_root, flags)
+    metadata = os.fstat(descriptor)
+    if not _owned_directory(metadata) or (
+        existed and stat.S_IMODE(metadata.st_mode) & 0o022
+    ):
+        os.close(descriptor)
+        raise DescriptorStoreError("replay ledger root is unsafe")
+    os.fchmod(descriptor, 0o700)
+    if not _safe_directory(os.fstat(descriptor)):
+        os.close(descriptor)
+        raise DescriptorStoreError("replay ledger root is unsafe")
+    return descriptor
+
+
 def _safe_parent(
     path: Path, trust_root: Path | None
 ) -> tuple[int, os.stat_result]:
@@ -92,11 +118,7 @@ def _safe_parent(
     flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
     descriptor = -1
     try:
-        trust_root.mkdir(mode=0o700, parents=True, exist_ok=True)
-        descriptor = os.open(trust_root, flags)
-        root_metadata = os.fstat(descriptor)
-        if not _safe_directory(root_metadata):
-            raise DescriptorStoreError("replay ledger root is unsafe")
+        descriptor = _open_trust_root(trust_root, flags)
         for component in relative_parent.parts:
             if component in {"", ".", ".."}:
                 raise DescriptorStoreError("replay ledger parent is unsafe")
@@ -147,7 +169,11 @@ def _read_json(parent_fd: int, name: str, empty: Any) -> Any:
         descriptor = _open_relative(parent_fd, name, os.O_RDONLY)
     try:
         metadata = os.fstat(descriptor)
-        if not stat.S_ISREG(metadata.st_mode) or stat.S_IMODE(metadata.st_mode) != 0o600:
+        if (
+            not stat.S_ISREG(metadata.st_mode)
+            or stat.S_IMODE(metadata.st_mode) != 0o600
+            or (hasattr(os, "getuid") and metadata.st_uid != os.getuid())
+        ):
             raise DescriptorStoreError("replay ledger is unsafe")
         with os.fdopen(descriptor, "r", encoding="utf-8", closefd=False) as handle:
             return json.load(handle)
@@ -194,7 +220,10 @@ def update_json(
     lock_fd = -1
     try:
         lock_fd = _open_relative(parent_fd, lock_name, os.O_RDWR | os.O_CREAT, 0o600)
-        if not stat.S_ISREG(os.fstat(lock_fd).st_mode):
+        lock_metadata = os.fstat(lock_fd)
+        if not stat.S_ISREG(lock_metadata.st_mode) or (
+            hasattr(os, "getuid") and lock_metadata.st_uid != os.getuid()
+        ):
             raise DescriptorStoreError("replay ledger lock is unsafe")
         os.fchmod(lock_fd, 0o600)
         _lock(lock_fd)

pdd/sync_core/evidence_store.py

@@ -50,6 +50,7 @@ def attestation_payload(envelope: AttestationEnvelope) -> dict[str, Any]:
     """Convert a signed envelope to stable JSON data without altering its payload."""
     binding = envelope.binding
     payload = {
+        "payload_version": envelope.payload_version,
         "attestation_id": envelope.attestation_id,
         "issuer": envelope.issuer,
         "binding": {
@@ -64,6 +65,7 @@ def attestation_payload(envelope: AttestationEnvelope) -> dict[str, Any]:
             "tool_version": binding.tool_version,
             "base_sha": binding.base_sha,
             "checked_sha": binding.checked_sha,
+            "artifact_closure_digest": binding.artifact_closure_digest,
         },
         "results": [
             {
@@ -85,6 +87,8 @@ def attestation_payload(envelope: AttestationEnvelope) -> dict[str, Any]:
         payload["binding"]["playwright_toolchain_manifest"] = (
             binding.playwright_toolchain_manifest
         )
+    if envelope.payload_version == 1:
+        payload["binding"].pop("artifact_closure_digest")
     return payload
 
 
@@ -106,6 +110,27 @@ def decode_attestation(payload: Mapping[str, Any]) -> AttestationEnvelope:
     """Decode untrusted JSON into a typed envelope without granting trust."""
     try:
         binding_data = payload["binding"]
+        version_data = payload.get("payload_version")
+        if version_data is None:
+            payload_version = 1 if "artifact_closure_digest" not in binding_data else 2
+        elif (
+            isinstance(version_data, int)
+            and not isinstance(version_data, bool)
+            and version_data in {1, 2}
+        ):
+            payload_version = version_data
+        else:
+            raise ValueError("attestation payload version is unsupported")
+        if payload_version == 1:
+            if "artifact_closure_digest" in binding_data:
+                raise ValueError(
+                    "v1 binding must not contain artifact_closure_digest"
+                )
+            artifact_closure_digest = ""
+        else:
+            artifact_closure_digest = _string(
+                binding_data, "artifact_closure_digest"
+            )
         subject_data = binding_data["subject"]
         validity_data = payload["validity"]
         results_data = payload["results"]
@@ -133,6 +158,7 @@ def decode_attestation(payload: Mapping[str, Any]) -> AttestationEnvelope:
             _string(binding_data, "checked_sha"),
             tuple(command_data) if command_data is not None else None,
             binding_data.get("playwright_toolchain_manifest"),
+            artifact_closure_digest,
         )
         results = tuple(
             ObligationEvidence(
@@ -153,6 +179,7 @@ def decode_attestation(payload: Mapping[str, Any]) -> AttestationEnvelope:
             results,
             validity,
             _string(payload, "signature"),
+            payload_version,
         )
     except (KeyError, TypeError, ValueError) as exc:
         raise EvidenceStoreError(f"attestation envelope is malformed: {exc}") from exc