Files
aios/tools/personal_weixin/pwx_b3_evidence_tool.py
T

1055 lines
41 KiB
Python

#!/usr/bin/env python3
"""Fail-closed Personal-Weixin B3 owner-evidence verifier.
The tool never captures network traffic and never accepts credential plaintext.
It hashes restricted raw files in streaming mode, validates commit-safe redacted
views, and requires a byte-exact independent review before emitting the single
R10 evidence manifest.
"""
from __future__ import annotations
import argparse
from datetime import datetime
import hashlib
import json
import os
from pathlib import Path
import re
import stat
import sys
import tempfile
import subprocess
import unicodedata
from typing import Any, NoReturn
SCHEMA_VERSION = 1
FACTS_CONTRACT = "personal-weixin-b3-owner-evidence-facts"
MANIFEST_CONTRACT = "personal-weixin-b3-owner-evidence-manifest"
ENVELOPE_CONTRACT = "personal-weixin-b3-owner-evidence"
FINAL_RELATIVE_PATH = Path("release/v2r1c0-rc1/weixin-bridge-evidence.json")
BLOCKED_STATUS = "BLOCKED_EVIDENCE_REQUIRED"
MAX_FACTS_BYTES = 256 * 1024
MAX_RAW_BYTES = 16 * 1024 * 1024
MAX_REDACTED_BYTES = 64 * 1024
MAX_REVIEW_BYTES = 1024 * 1024
HEX64 = re.compile(r"^[0-9a-f]{64}$")
SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$")
SAFE_OPERATION = re.compile(r"^[a-z][a-z0-9_]{0,63}$")
JSON_TYPES = {"string", "integer", "number", "boolean", "null", "object", "array"}
MODE_NAMES = {"sandbox", "shadow", "restricted-capture"}
EXPECTED_LINEAGE = {
"foundation_implementation_commit": "2fcab6f5833a405a9a26683811600b6e5f1aa7b1",
"foundation_result_commit": "64a71b6199a8e2bd609ba145349e67b874888505",
"r10_lineage_commit": "74e603c0c943501110577dbe3c7f6a0b18b8ec7d",
"evidence_gate_commit": "62a4067cf7a462ebe7cf29524238bc5d0a70b813",
"upstream_audit_commit": "b2f5f531019a70801444987dbb97dc1f69cf47de",
"authorization_commit": "a1f933e500151dc8" "afc1b25b30448a31899a8f64",
"b4_amendment_commit": "1d334b1ccaac54c3c3ec4dd84693ab68fc67cef3",
"authorization_result_commit": "8af40af62927caf631ea6fc682155f0f65cbd1a6",
}
EXPECTED_AUTHORITY = {
"vendor_message_id_field": "$.msgs[*].message_id",
"client_id_field": "$.msgs[*].client_id",
"client_id_authoritative": False,
"mapping_verified": True,
"authoritative": ["verified inbound vendor message_id"],
"non_authoritative": [
"local client_id",
"outbound correlation ID",
"local sequence",
"timestamp",
"content hash",
"outbound receipt ID",
],
}
EXPECTED_PRODUCTION = {
"baseline": "V1R4C6",
"changed": False,
"release_status": "paused",
}
class Blocked(Exception):
"""A sanitized, expected fail-closed decision."""
def __init__(self, code: str, field: str = "", capture_id: str = "") -> None:
super().__init__(code)
self.code = code if SAFE_OPERATION.fullmatch(code.lower()) else "validation_blocked"
self.field = field if re.fullmatch(r"[A-Za-z0-9_.\[\]-]{0,128}", field) else ""
self.capture_id = capture_id if SAFE_ID.fullmatch(capture_id) else ""
class SafeParser(argparse.ArgumentParser):
def error(self, message: str) -> NoReturn:
del message
raise Blocked("cli_invalid", "cli")
def _blocked(code: str, field: str = "", capture_id: str = "") -> NoReturn:
raise Blocked(code, field, capture_id)
def _emit_blocked(error: Blocked) -> int:
parts = ["ERROR", error.code.upper()]
if error.field:
parts.append(f"field={error.field}")
if error.capture_id:
parts.append(f"capture_id={error.capture_id}")
print(" ".join(parts), file=sys.stderr)
print(BLOCKED_STATUS)
return 2
def _canonical_bytes(value: Any) -> bytes:
try:
encoded = json.dumps(
value,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
allow_nan=False,
)
except (TypeError, ValueError):
_blocked("canonical_encoding", "manifest")
return encoded.encode("utf-8") + b"\n"
def _reject_float(value: str) -> NoReturn:
del value
_blocked("float_not_allowed", "json")
def _pairs_no_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
result: dict[str, Any] = {}
for key, value in pairs:
if key in result:
_blocked("duplicate_json_key", "json")
result[key] = value
return result
def _decode_json(data: bytes, field: str) -> Any:
try:
text = data.decode("utf-8", errors="strict")
return json.loads(
text,
object_pairs_hook=_pairs_no_duplicates,
parse_float=_reject_float,
parse_constant=_reject_float,
)
except Blocked:
raise
except (UnicodeDecodeError, json.JSONDecodeError):
_blocked("invalid_json", field)
def _read_bounded(path: Path, limit: int, field: str) -> bytes:
path = _reject_symlink_chain(path, field, include_leaf=False)
try:
metadata = path.stat()
if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > limit:
_blocked("file_bounds", field)
return path.read_bytes()
except Blocked:
raise
except OSError:
_blocked("file_unavailable", field)
def _load_json(path: Path, limit: int, field: str) -> Any:
path = _reject_symlink_chain(path, field, include_leaf=True)
return _decode_json(_read_bounded(path, limit, field), field)
def _strict_dict(value: Any, expected: set[str], field: str) -> dict[str, Any]:
if type(value) is not dict or set(value) != expected:
_blocked("schema_keys", field)
return value
def _strict_list(value: Any, field: str) -> list[Any]:
if type(value) is not list:
_blocked("schema_type", field)
return value
def _strict_string(value: Any, field: str, *, nonempty: bool = True) -> str:
if type(value) is not str or (nonempty and not value.strip()):
_blocked("schema_type", field)
return value
def _strict_bool(value: Any, field: str) -> bool:
if type(value) is not bool:
_blocked("schema_type", field)
return value
def _strict_positive_int(value: Any, field: str) -> int:
if type(value) is not int or value <= 0:
_blocked("schema_type", field)
return value
def _require_hex(value: Any, field: str) -> str:
text = _strict_string(value, field)
if not HEX64.fullmatch(text):
_blocked("sha256_invalid", field)
return text
def _parse_rfc3339(value: Any, field: str) -> datetime:
text = _strict_string(value, field)
candidate = text[:-1] + "+00:00" if text.endswith("Z") else text
try:
parsed = datetime.fromisoformat(candidate)
except ValueError:
_blocked("timestamp_invalid", field)
if parsed.tzinfo is None or parsed.utcoffset() is None:
_blocked("timestamp_timezone_required", field)
return parsed
def _reject_symlink_chain(
path: Path,
field: str, *,
include_leaf: bool,
) -> Path:
absolute = Path(os.path.abspath(os.fspath(path)))
if not absolute.is_absolute():
_blocked("absolute_path_required", field)
current = Path(absolute.anchor)
parts = absolute.parts[1:]
count = len(parts) if include_leaf else max(len(parts) - 1, 0)
for index, part in enumerate(parts[:count]):
current = current / part
try:
metadata = os.lstat(current)
except OSError:
_blocked("path_component_unavailable", field)
if stat.S_ISLNK(metadata.st_mode):
_blocked("ancestor_symlink_rejected", field)
if index < count - 1 and not stat.S_ISDIR(metadata.st_mode):
_blocked("ancestor_not_directory", field)
return absolute
def _absolute(path: Path, field: str) -> Path:
return _reject_symlink_chain(path, field, include_leaf=True)
def _is_within(path: Path, root: Path) -> bool:
try:
path.relative_to(root)
return True
except ValueError:
return False
def _validate_root(path: Path, field: str) -> Path:
root = _absolute(path, field)
try:
info = os.lstat(root)
except OSError:
_blocked("root_unavailable", field)
if not stat.S_ISDIR(info.st_mode):
_blocked("root_invalid", field)
return root
def _regular_beneath(
supplied: Any,
restricted_root: Path,
repo_root: Path,
field: str,
) -> tuple[Path, os.stat_result]:
path = _absolute(Path(_strict_string(supplied, field)), field)
if not _is_within(path, restricted_root) or _is_within(path, repo_root):
_blocked("restricted_path_boundary", field)
relative = path.relative_to(restricted_root)
current = restricted_root
try:
for part in relative.parts:
current = current / part
metadata = os.lstat(current)
if stat.S_ISLNK(metadata.st_mode):
_blocked("symlink_rejected", field)
if not stat.S_ISREG(metadata.st_mode):
_blocked("regular_file_required", field)
except Blocked:
raise
except OSError:
_blocked("file_unavailable", field)
return path, metadata
def _stream_hash(
path: Path,
initial: os.stat_result,
limit: int,
field: str,
) -> tuple[str, int]:
if initial.st_size <= 0 or initial.st_size > limit:
_blocked("file_bounds", field)
flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
try:
descriptor = os.open(path, flags)
except OSError:
_blocked("file_unavailable", field)
digest = hashlib.sha256()
count = 0
try:
before = os.fstat(descriptor)
identity = (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns)
if identity != (initial.st_dev, initial.st_ino, initial.st_size, initial.st_mtime_ns):
_blocked("file_changed", field)
while True:
chunk = os.read(descriptor, 1024 * 1024)
if not chunk:
break
count += len(chunk)
if count > limit:
_blocked("file_bounds", field)
digest.update(chunk)
after = os.fstat(descriptor)
if identity != (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns):
_blocked("file_changed", field)
finally:
os.close(descriptor)
if count != initial.st_size:
_blocked("file_changed", field)
return digest.hexdigest(), count
def _read_redacted(
path: Path,
initial: os.stat_result,
field: str,
) -> tuple[bytes, str, int]:
if initial.st_size <= 0 or initial.st_size > MAX_REDACTED_BYTES:
_blocked("file_bounds", field)
flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
try:
descriptor = os.open(path, flags)
except OSError:
_blocked("file_unavailable", field)
try:
before = os.fstat(descriptor)
identity = (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns)
if identity != (initial.st_dev, initial.st_ino, initial.st_size, initial.st_mtime_ns):
_blocked("file_changed", field)
data = bytearray()
while True:
chunk = os.read(descriptor, 8192)
if not chunk:
break
data.extend(chunk)
if len(data) > MAX_REDACTED_BYTES:
_blocked("file_bounds", field)
after = os.fstat(descriptor)
if identity != (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns):
_blocked("file_changed", field)
finally:
os.close(descriptor)
raw = bytes(data)
return raw, hashlib.sha256(raw).hexdigest(), len(raw)
def _validate_redacted_view(
value: Any,
capture_id: str,
direction: str,
credential_label: str,
) -> dict[str, Any]:
view = _strict_dict(
value,
{"schema_version", "capture_id", "direction", "credential_label", "observed_fields"},
"redacted_view",
)
if view["schema_version"] != SCHEMA_VERSION:
_blocked("schema_version", "redacted_view.schema_version", capture_id)
if view["capture_id"] != capture_id or view["direction"] != direction:
_blocked("redacted_identity", "redacted_view", capture_id)
if view["credential_label"] != credential_label:
_blocked("redacted_credential", "redacted_view.credential_label", capture_id)
observed = _strict_list(view["observed_fields"], "redacted_view.observed_fields")
if not observed:
_blocked("redacted_fields_missing", "redacted_view.observed_fields", capture_id)
normalized: list[dict[str, Any]] = []
seen: set[str] = set()
for index, item in enumerate(observed):
entry = _strict_dict(
item,
{"path", "json_type", "present", "value"},
f"redacted_view.observed_fields[{index}]",
)
json_path = _strict_string(entry["path"], f"redacted_view.observed_fields[{index}].path")
if not re.fullmatch(r"\$[A-Za-z0-9_.\[\]*-]{1,126}", json_path) or json_path in seen:
_blocked("redacted_path_invalid", f"redacted_view.observed_fields[{index}].path", capture_id)
seen.add(json_path)
json_type = _strict_string(entry["json_type"], f"redacted_view.observed_fields[{index}].json_type")
if json_type not in JSON_TYPES:
_blocked("redacted_type_invalid", f"redacted_view.observed_fields[{index}].json_type", capture_id)
present = _strict_bool(entry["present"], f"redacted_view.observed_fields[{index}].present")
expected_value = "<REDACTED>" if present else "<ABSENT>"
if entry["value"] != expected_value:
_blocked("redacted_value_invalid", f"redacted_view.observed_fields[{index}].value", capture_id)
normalized.append(
{"path": json_path, "json_type": json_type, "present": present, "value": expected_value}
)
if direction == "inbound":
required = {EXPECTED_AUTHORITY["vendor_message_id_field"], EXPECTED_AUTHORITY["client_id_field"]}
if not required.issubset(seen):
_blocked("authority_fields_missing", "redacted_view.observed_fields", capture_id)
return {
"schema_version": SCHEMA_VERSION,
"capture_id": capture_id,
"direction": direction,
"credential_label": credential_label,
"observed_fields": sorted(normalized, key=lambda item: item["path"]),
}
def _validate_raw_wire(
value: Any,
restricted_root: Path,
repo_root: Path,
) -> tuple[list[dict[str, Any]], dict[str, tuple[str, str]], datetime]:
entries = _strict_list(value, "raw_wire")
if len(entries) < 2:
_blocked("raw_wire_missing", "raw_wire")
normalized: list[dict[str, Any]] = []
capture_index: dict[str, tuple[str, str]] = {}
directions: set[str] = set()
credential_labels: set[str] = set()
latest: datetime | None = None
expected_keys = {
"capture_id", "direction", "credential_label", "captured_at",
"restricted_path", "restricted_sha256", "restricted_bytes",
"redacted_source_path", "redacted_sha256", "redacted_bytes",
}
for index, item in enumerate(entries):
field = f"raw_wire[{index}]"
entry = _strict_dict(item, expected_keys, field)
capture_id = _strict_string(entry["capture_id"], f"{field}.capture_id")
if not SAFE_ID.fullmatch(capture_id) or capture_id in capture_index:
_blocked("capture_id_invalid", f"{field}.capture_id")
direction = _strict_string(entry["direction"], f"{field}.direction")
if direction not in {"inbound", "outbound"}:
_blocked("direction_invalid", f"{field}.direction", capture_id)
credential_label = _strict_string(entry["credential_label"], f"{field}.credential_label")
if credential_label not in {"credential-a", "credential-b"}:
_blocked("credential_label_invalid", f"{field}.credential_label", capture_id)
captured_at = _parse_rfc3339(entry["captured_at"], f"{field}.captured_at")
raw_path, raw_info = _regular_beneath(
entry["restricted_path"], restricted_root, repo_root, f"{field}.restricted_path"
)
actual_sha, actual_bytes = _stream_hash(raw_path, raw_info, MAX_RAW_BYTES, f"{field}.restricted_path")
if _require_hex(entry["restricted_sha256"], f"{field}.restricted_sha256") != actual_sha:
_blocked("raw_sha_mismatch", f"{field}.restricted_sha256", capture_id)
if _strict_positive_int(entry["restricted_bytes"], f"{field}.restricted_bytes") != actual_bytes:
_blocked("raw_bytes_mismatch", f"{field}.restricted_bytes", capture_id)
redacted_path, redacted_info = _regular_beneath(
entry["redacted_source_path"], restricted_root, repo_root, f"{field}.redacted_source_path"
)
redacted_bytes, redacted_sha, redacted_size = _read_redacted(
redacted_path, redacted_info, f"{field}.redacted_source_path"
)
if _require_hex(entry["redacted_sha256"], f"{field}.redacted_sha256") != redacted_sha:
_blocked("redacted_sha_mismatch", f"{field}.redacted_sha256", capture_id)
if _strict_positive_int(entry["redacted_bytes"], f"{field}.redacted_bytes") != redacted_size:
_blocked("redacted_bytes_mismatch", f"{field}.redacted_bytes", capture_id)
redacted_view = _validate_redacted_view(
_decode_json(redacted_bytes, f"{field}.redacted_source_path"),
capture_id,
direction,
credential_label,
)
normalized.append(
{
"capture_id": capture_id,
"direction": direction,
"credential_label": credential_label,
"captured_at": entry["captured_at"],
"restricted_path": str(raw_path),
"restricted_sha256": actual_sha,
"restricted_bytes": actual_bytes,
"redacted_path": (
f"{FINAL_RELATIVE_PATH.as_posix()}#/manifest/raw_wire/"
f"{capture_id}/redacted_view"
),
"redacted_sha256": redacted_sha,
"redacted_bytes": redacted_size,
"redacted_view": redacted_view,
}
)
capture_index[capture_id] = (direction, credential_label)
directions.add(direction)
credential_labels.add(credential_label)
latest = captured_at if latest is None or captured_at > latest else latest
if directions != {"inbound", "outbound"}:
_blocked("raw_direction_coverage", "raw_wire")
if credential_labels != {"credential-a", "credential-b"}:
_blocked("raw_credential_coverage", "raw_wire")
assert latest is not None
return sorted(normalized, key=lambda item: item["capture_id"]), capture_index, latest
def _validate_mode(value: Any) -> dict[str, Any]:
mode = _strict_dict(
value,
{"name", "production_traffic", "writes_production", "capture_only"},
"mode",
)
name = _strict_string(mode["name"], "mode.name")
if name not in MODE_NAMES:
_blocked("mode_invalid", "mode.name")
if _strict_bool(mode["production_traffic"], "mode.production_traffic") is not False:
_blocked("production_traffic_forbidden", "mode.production_traffic")
if _strict_bool(mode["writes_production"], "mode.writes_production") is not False:
_blocked("production_write_forbidden", "mode.writes_production")
if _strict_bool(mode["capture_only"], "mode.capture_only") is not True:
_blocked("capture_only_required", "mode.capture_only")
return {
"name": name,
"production_traffic": False,
"writes_production": False,
"capture_only": True,
}
def _validate_credentials(
value: Any,
capture_index: dict[str, tuple[str, str]],
mode_name: str,
) -> list[dict[str, Any]]:
credentials = _strict_list(value, "credentials")
if len(credentials) != 2:
_blocked("credential_count", "credentials")
normalized: list[dict[str, Any]] = []
labels: set[str] = set()
fingerprints: set[str] = set()
expected_keys = {"label", "fingerprint_sha256", "secret_included", "scope", "test_result"}
for index, item in enumerate(credentials):
field = f"credentials[{index}]"
credential = _strict_dict(item, expected_keys, field)
label = _strict_string(credential["label"], f"{field}.label")
if label not in {"credential-a", "credential-b"} or label in labels:
_blocked("credential_label_invalid", f"{field}.label")
fingerprint = _require_hex(credential["fingerprint_sha256"], f"{field}.fingerprint_sha256")
if fingerprint in fingerprints:
_blocked("credential_fingerprint_duplicate", f"{field}.fingerprint_sha256")
if _strict_bool(credential["secret_included"], f"{field}.secret_included") is not False:
_blocked("credential_secret_forbidden", f"{field}.secret_included")
scope = _strict_dict(
credential["scope"],
{"mode", "operations", "evidence_capture_ids"},
f"{field}.scope",
)
if scope["mode"] != mode_name:
_blocked("credential_mode_mismatch", f"{field}.scope.mode")
operations = _strict_list(scope["operations"], f"{field}.scope.operations")
if not operations or len(operations) != len(set(operations)):
_blocked("credential_operations", f"{field}.scope.operations")
for operation in operations:
if type(operation) is not str or not SAFE_OPERATION.fullmatch(operation):
_blocked("credential_operations", f"{field}.scope.operations")
capture_ids = _strict_list(scope["evidence_capture_ids"], f"{field}.scope.evidence_capture_ids")
if not capture_ids or len(capture_ids) != len(set(capture_ids)):
_blocked("credential_capture_ids", f"{field}.scope.evidence_capture_ids")
expected_capture_ids = {
capture_id for capture_id, (_, capture_label) in capture_index.items() if capture_label == label
}
if set(capture_ids) != expected_capture_ids:
_blocked("credential_capture_coverage", f"{field}.scope.evidence_capture_ids")
if credential["test_result"] != "PASS":
_blocked("credential_test_required", f"{field}.test_result")
labels.add(label)
fingerprints.add(fingerprint)
normalized.append(
{
"label": label,
"fingerprint_sha256": fingerprint,
"secret_included": False,
"scope": {
"mode": mode_name,
"operations": sorted(operations),
"evidence_capture_ids": sorted(capture_ids),
},
"test_result": "PASS",
}
)
if labels != {"credential-a", "credential-b"}:
_blocked("credential_coverage", "credentials")
return sorted(normalized, key=lambda item: item["label"])
def _validate_authority(value: Any) -> dict[str, Any]:
authority = _strict_dict(value, set(EXPECTED_AUTHORITY), "authority_map")
if authority != EXPECTED_AUTHORITY:
_blocked("authority_map_mismatch", "authority_map")
return {
**EXPECTED_AUTHORITY,
"authoritative": sorted(EXPECTED_AUTHORITY["authoritative"]),
"non_authoritative": sorted(EXPECTED_AUTHORITY["non_authoritative"]),
}
def _validate_production(value: Any) -> dict[str, Any]:
production = _strict_dict(value, set(EXPECTED_PRODUCTION), "production")
if production != EXPECTED_PRODUCTION:
_blocked("production_boundary", "production")
return dict(EXPECTED_PRODUCTION)
def _build_manifest(facts: Any, restricted_root: Path, repo_root: Path) -> tuple[dict[str, Any], datetime]:
root = _strict_dict(
facts,
{
"schema_version", "contract", "executor", "lineage", "raw_wire",
"credentials", "authority_map", "mode", "review", "production",
},
"facts",
)
if root["schema_version"] != SCHEMA_VERSION or root["contract"] != FACTS_CONTRACT:
_blocked("facts_contract", "facts")
executor = _strict_string(root["executor"], "executor")
if len(executor) > 128:
_blocked("executor_invalid", "executor")
lineage = _strict_dict(root["lineage"], set(EXPECTED_LINEAGE), "lineage")
if lineage != EXPECTED_LINEAGE:
_blocked("lineage_mismatch", "lineage")
mode = _validate_mode(root["mode"])
raw_wire, capture_index, latest_capture = _validate_raw_wire(
root["raw_wire"], restricted_root, repo_root
)
credentials = _validate_credentials(root["credentials"], capture_index, mode["name"])
authority = _validate_authority(root["authority_map"])
production = _validate_production(root["production"])
manifest = {
"schema_version": SCHEMA_VERSION,
"contract": MANIFEST_CONTRACT,
"executor": executor,
"lineage": dict(EXPECTED_LINEAGE),
"raw_wire": raw_wire,
"credentials": credentials,
"authority_map": authority,
"mode": mode,
"production": production,
"secrets_committed": False,
"gates": {
"raw_wire_inbound": "PASS",
"raw_wire_outbound": "PASS",
"dual_credential": "PASS",
"authority_map": "PASS",
"mode_non_production": "PASS",
"reviewer_independent": "PASS",
"reviewer_decision": "PASS",
"secrets_committed_false": "PASS",
"production_changed_false": "PASS",
},
}
return manifest, latest_capture
def _validate_review(value: Any, executor: str, manifest_sha: str, latest_capture: datetime) -> dict[str, Any]:
review = _strict_dict(
value,
{"reviewer", "independent_from_executor", "decision", "reviewed_manifest_sha256", "reviewed_at"},
"review",
)
reviewer = _strict_string(review["reviewer"], "review.reviewer")
if len(reviewer) > 128:
_blocked("reviewer_invalid", "review.reviewer")
normalize = lambda text: unicodedata.normalize("NFKC", text).strip().casefold()
if normalize(reviewer) == normalize(executor):
_blocked("reviewer_not_independent", "review.reviewer")
if _strict_bool(review["independent_from_executor"], "review.independent_from_executor") is not True:
_blocked("reviewer_not_independent", "review.independent_from_executor")
if review["decision"] != "PASS":
_blocked("review_pass_required", "review.decision")
if _require_hex(review["reviewed_manifest_sha256"], "review.reviewed_manifest_sha256") != manifest_sha:
_blocked("review_hash_mismatch", "review.reviewed_manifest_sha256")
reviewed_at = _parse_rfc3339(review["reviewed_at"], "review.reviewed_at")
if reviewed_at < latest_capture:
_blocked("review_timestamp_invalid", "review.reviewed_at")
return {
"reviewer": reviewer,
"independent_from_executor": True,
"decision": "PASS",
"reviewed_manifest_sha256": manifest_sha,
"reviewed_at": review["reviewed_at"],
}
def _safe_external_output(path: Path, repo_root: Path, restricted_root: Path, field: str) -> Path:
output = _absolute(path, field)
if _is_within(output, repo_root) or _is_within(output, restricted_root):
_blocked("output_boundary", field)
parent = _reject_symlink_chain(output.parent, f"{field}.parent", include_leaf=True)
try:
metadata = os.lstat(parent)
except OSError:
_blocked("output_parent_unavailable", field)
if not stat.S_ISDIR(metadata.st_mode):
_blocked("output_parent_invalid", field)
return output
def _dirfd_atomic_replace(path: Path, data: bytes, mode: int = 0o600) -> None:
parent = _reject_symlink_chain(path.parent, f"{path.name}.parent", include_leaf=True)
directory_fd = os.open(
parent,
os.O_RDONLY
| getattr(os, "O_DIRECTORY", 0)
| getattr(os, "O_NOFOLLOW", 0),
)
opened = os.fstat(directory_fd)
observed = os.lstat(parent)
if opened.st_dev != observed.st_dev or opened.st_ino != observed.st_ino:
os.close(directory_fd)
_blocked("directory_identity_changed", f"{path.name}.parent")
temporary_name = f".{path.name}.{os.urandom(4).hex()}"
temporary_fd = os.open(
temporary_name,
os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0),
mode,
dir_fd=directory_fd,
)
try:
with os.fdopen(temporary_fd, "wb", closefd=True) as stream:
stream.write(data)
stream.flush()
os.fsync(stream.fileno())
os.replace(
temporary_name,
path.name,
src_dir_fd=directory_fd,
dst_dir_fd=directory_fd,
)
os.fsync(directory_fd)
except Exception:
try:
os.unlink(temporary_name, dir_fd=directory_fd)
except OSError:
pass
raise
finally:
os.close(directory_fd)
def _validate_trust_policy(policy: dict[str, Any], field: str) -> None:
"""Validate trust policy structure and permissions."""
if policy.get("schema_version") != 1:
_blocked("trust_policy_schema", field)
if policy.get("contract") != "personal-weixin-b3-trust-policy":
_blocked("trust_policy_contract", field)
for role_key in ("capture", "credential", "reviewer"):
role_field = f"{field}.{role_key}"
role = _strict_dict(
policy.get(role_key, {}),
{"allowed_signers", "allowed_signers_sha256", "namespace"},
role_field,
)
for file_key in ("allowed_signers",):
file_path = _reject_symlink_chain(Path(role[file_key]), f"{role_field}.{file_key}", include_leaf=True)
try:
metadata = os.lstat(file_path)
if metadata.st_uid != 0:
_blocked("trust_policy_owner_not_root", f"{role_field}.{file_key}")
if metadata.st_mode & (stat.S_IRWXG | stat.S_IRWXO):
_blocked("trust_policy_permissive", f"{role_field}.{file_key}")
except OSError:
_blocked("trust_policy_file_unavailable", f"{role_field}.{file_key}")
sha256 = role.get("allowed_signers_sha256", "")
if not HEX64.fullmatch(sha256):
_blocked("trust_policy_invalid_sha256", role_field)
namespace = role.get("namespace", "")
if not namespace.startswith("aios-pwx-b3-"):
_blocked("trust_policy_bad_namespace", role_field)
def _verify_ssh_signature(
signature_path: Path,
payload_path: Path,
allowed_signers: Path,
principal: str,
namespace: str,
field: str,
) -> None:
"""Verify an SSH signature using ssh-keygen."""
allowed_signers = _reject_symlink_chain(allowed_signers, f"{field}.allowed_signers", include_leaf=True)
signature_path = _reject_symlink_chain(signature_path, f"{field}.signature" , include_leaf=True)
payload_path = _reject_symlink_chain(payload_path, f"{field}.payload", include_leaf=True)
try:
result = subprocess.run(
[
"ssh-keygen",
"-Y", "verify",
"-f", os.fspath(allowed_signers),
"-I", principal,
"-n", namespace,
"-s", os.fspath(signature_path),
],
input=payload_path.read_bytes(),
capture_output=True,
timeout=30,
)
if result.returncode != 0:
_blocked("ssh_signature_verification_failed", field)
except (OSError, subprocess.TimeoutExpired):
_blocked("ssh_signature_verification_error", field)
def _prepare_review(args: argparse.Namespace) -> int:
repo_root = _validate_root(Path(args.repo_root), "repo_root")
restricted_root = _validate_root(Path(args.restricted_root), "restricted_root")
if _is_within(restricted_root, repo_root) or _is_within(repo_root, restricted_root):
_blocked("root_overlap", "restricted_root")
trust_policy = _load_json(Path(args.trust_policy), MAX_FACTS_BYTES, "trust_policy")
_validate_trust_policy(trust_policy, "trust_policy")
capture_policy = trust_policy["capture"]
credential_policy = trust_policy["credential"]
facts = _load_json(Path(args.facts), MAX_FACTS_BYTES, "facts")
if "capture_attestation" in facts:
_verify_ssh_signature(
Path(facts["capture_attestation"]),
Path(args.facts),
Path(capture_policy["allowed_signers"]),
capture_policy["namespace"],
capture_policy["namespace"],
"capture_attestation",
)
if "credential_attestation" in facts:
_verify_ssh_signature(
Path(facts["credential_attestation"]),
Path(args.facts),
Path(credential_policy["allowed_signers"]),
credential_policy["namespace"],
credential_policy["namespace"],
"credential_attestation",
)
manifest, _ = _build_manifest(facts, restricted_root, repo_root)
payload_bytes = _canonical_bytes(manifest)
digest = hashlib.sha256(payload_bytes).hexdigest()
payload_out = _safe_external_output(
Path(args.review_payload_out), repo_root, restricted_root, "review_payload_out"
)
hash_out = _safe_external_output(
Path(args.review_hash_out), repo_root, restricted_root, "review_hash_out"
)
if payload_out == hash_out:
_blocked("output_collision", "review_hash_out")
_dirfd_atomic_replace(payload_out, payload_bytes)
_dirfd_atomic_replace(hash_out, (digest + "\n").encode("ascii"))
return _emit_blocked(Blocked("review_required", "review"))
def _read_external_review_payload(
path: Path,
repo_root: Path,
restricted_root: Path,
) -> bytes:
review_path = _safe_external_output(path, repo_root, restricted_root, "review_payload")
try:
metadata = os.lstat(review_path)
except OSError:
_blocked("review_payload_unavailable", "review_payload")
if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode):
_blocked("review_payload_invalid", "review_payload")
return _read_bounded(review_path, MAX_REVIEW_BYTES, "review_payload")
def _ensure_directory_chain(root: Path, parent: Path, field: str) -> None:
try:
relative = parent.relative_to(root)
except ValueError:
_blocked("output_boundary", field)
current = root
for part in relative.parts:
current = current / part
try:
metadata = os.lstat(current)
except FileNotFoundError:
try:
os.mkdir(current, 0o775)
metadata = os.lstat(current)
except OSError:
_blocked("manifest_parent_unavailable", field)
except OSError:
_blocked("manifest_parent_unavailable", field)
if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
_blocked("manifest_parent_symlink", field)
try:
if parent.resolve(strict=True) != parent:
_blocked("manifest_parent_symlink", field)
except OSError:
_blocked("manifest_parent_unavailable", field)
def _atomic_create_or_match(path: Path, data: bytes, repo_root: Path) -> None:
_ensure_directory_chain(repo_root, path.parent, "manifest_out")
if os.path.lexists(path):
try:
metadata = os.lstat(path)
except OSError:
_blocked("manifest_unavailable", "manifest_out")
if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode):
_blocked("manifest_symlink", "manifest_out")
path = _reject_symlink_chain(path, "manifest_out", include_leaf=True)
if _read_bounded(path, MAX_REVIEW_BYTES, "manifest_out") == data:
return
_blocked("manifest_conflict", "manifest_out")
parent = _reject_symlink_chain(path.parent, "manifest_out.parent", include_leaf=True)
directory_fd = os.open(
parent,
os.O_RDONLY
| getattr(os, "O_DIRECTORY", 0)
| getattr(os, "O_NOFOLLOW", 0),
)
opened = os.fstat(directory_fd)
observed = os.lstat(parent)
if opened.st_dev != observed.st_dev or opened.st_ino != observed.st_ino:
os.close(directory_fd)
_blocked("directory_identity_changed", "manifest_out.parent")
temporary_name = f".{path.name}.{os.urandom(4).hex()}"
temporary_fd = os.open(
temporary_name,
os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0),
0o644,
dir_fd=directory_fd,
)
try:
with os.fdopen(temporary_fd, "wb", closefd=True) as stream:
stream.write(data)
stream.flush()
os.fsync(stream.fileno())
try:
os.link(temporary_name, path.name, src_dir_fd=directory_fd, dst_dir_fd=directory_fd)
except FileExistsError:
destination = _reject_symlink_chain(path, "manifest_out", include_leaf=True)
if _read_bounded(destination, MAX_REVIEW_BYTES, "manifest_out") != data:
_blocked("manifest_conflict", "manifest_out")
os.fsync(directory_fd)
finally:
try:
os.unlink(temporary_name, dir_fd=directory_fd)
except OSError:
pass
os.close(directory_fd)
def _finalize(args: argparse.Namespace) -> int:
repo_root = _validate_root(Path(args.repo_root), "repo_root")
restricted_root = _validate_root(Path(args.restricted_root), "restricted_root")
if _is_within(restricted_root, repo_root) or _is_within(repo_root, restricted_root):
_blocked("root_overlap", "restricted_root")
trust_policy = _load_json(Path(args.trust_policy), MAX_FACTS_BYTES, "trust_policy")
_validate_trust_policy(trust_policy, "trust_policy")
capture_policy = trust_policy["capture"]
credential_policy = trust_policy["credential"]
reviewer_policy = trust_policy["reviewer"]
facts = _load_json(Path(args.facts), MAX_FACTS_BYTES, "facts")
if "capture_attestation" in facts:
_verify_ssh_signature(
Path(facts["capture_attestation"]),
Path(args.facts),
Path(capture_policy["allowed_signers"]),
capture_policy["namespace"],
capture_policy["namespace"],
"capture_attestation",
)
if "credential_attestation" in facts:
_verify_ssh_signature(
Path(facts["credential_attestation"]),
Path(args.facts),
Path(credential_policy["allowed_signers"]),
credential_policy["namespace"],
credential_policy["namespace"],
"credential_attestation",
)
manifest, latest_capture = _build_manifest(facts, restricted_root, repo_root)
manifest_bytes = _canonical_bytes(manifest)
reviewed_bytes = _read_external_review_payload(
Path(args.review_payload), repo_root, restricted_root
)
if reviewed_bytes != manifest_bytes:
_blocked("review_payload_stale", "review_payload")
manifest_sha = hashlib.sha256(manifest_bytes).hexdigest()
review = _validate_review(facts["review"], manifest["executor"], manifest_sha, latest_capture)
if review is not None:
review_path = _reject_symlink_chain(Path(facts["review"]), "review", include_leaf=True)
_verify_ssh_signature(
review_path,
Path(args.facts),
Path(reviewer_policy["allowed_signers"]),
reviewer_policy["namespace"],
reviewer_policy["namespace"],
"review",
)
manifest_out = _absolute(Path(args.manifest_out), "manifest_out")
expected_out = repo_root / FINAL_RELATIVE_PATH
if manifest_out != expected_out:
_blocked("manifest_path_forbidden", "manifest_out")
envelope = {
"schema_version": SCHEMA_VERSION,
"contract": ENVELOPE_CONTRACT,
"status": "PASS",
"manifest": manifest,
"review": review,
}
_atomic_create_or_match(manifest_out, _canonical_bytes(envelope), repo_root)
print("PASS")
return 0
def _parser() -> SafeParser:
parser = SafeParser(description="Verify Personal-Weixin B3 owner evidence without exposing raw data.")
subparsers = parser.add_subparsers(dest="command", required=True, parser_class=SafeParser)
prepare = subparsers.add_parser("prepare-review", help="build the canonical reviewer payload")
prepare.add_argument("--facts", required=True)
prepare.add_argument("--restricted-root", required=True)
prepare.add_argument("--repo-root", required=True)
prepare.add_argument("--review-payload-out", required=True)
prepare.add_argument("--review-hash-out", required=True)
prepare.add_argument("--trust-policy", required=True)
prepare.set_defaults(handler=_prepare_review)
finalize = subparsers.add_parser("finalize", help="verify review and emit the R10 evidence manifest")
finalize.add_argument("--facts", required=True)
finalize.add_argument("--restricted-root", required=True)
finalize.add_argument("--repo-root", required=True)
finalize.add_argument("--review-payload", required=True)
finalize.add_argument("--manifest-out", required=True)
finalize.add_argument("--trust-policy", required=True)
finalize.set_defaults(handler=_finalize)
return parser
def main(argv: list[str] | None = None) -> int:
try:
args = _parser().parse_args(argv)
return int(args.handler(args))
except Blocked as error:
return _emit_blocked(error)
except Exception:
return _emit_blocked(Blocked("internal_blocked", "internal"))
if __name__ == "__main__":
raise SystemExit(main())