from __future__ import annotations import hashlib import os import re import stat from dataclasses import dataclass from pathlib import Path from typing import Any, Mapping from .constants import ( EXTENSION_ID, EXTENSION_NAME, EXTENSION_ORIGIN, EXTENSION_VERSION, HOST_NAME, SOURCE_MANIFEST_NAME, TASK_ID, ) from .strict_json import canonical_bytes, loads LOWER_SHA256 = re.compile(r"^[0-9a-f]{64}$") class IdentityError(RuntimeError): def __init__(self, code: str, message: str) -> None: super().__init__(message) self.code = code @dataclass(frozen=True) class VerifiedLocalIdentity: source_root: Path source_manifest_bytes: int source_manifest_sha256: str payload_tree_sha256: str extension_id: str host_install_receipt_sha256: str chrome_parent_pid: int def _exact_keys(value: Any, expected: set[str], field: str) -> Mapping[str, Any]: if not isinstance(value, dict) or set(value) != expected: raise IdentityError("E_LOCAL_EXTENSION_IDENTITY", f"{field} keys differ") return value def _is_reparse(path: Path) -> bool: info = path.lstat() attrs = getattr(info, "st_file_attributes", 0) return stat.S_ISLNK(info.st_mode) or bool(attrs & getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0x400)) def _safe_root(path: Path) -> Path: absolute = Path(os.path.abspath(path)) current = Path(absolute.anchor) for part in absolute.parts[1:]: current /= part if not current.exists() or _is_reparse(current): raise IdentityError("E_LOCAL_EXTENSION_IDENTITY", "source root is missing or reparse-backed") if not absolute.is_dir(): raise IdentityError("E_LOCAL_EXTENSION_IDENTITY", "source root is not a directory") return absolute def _tree_hash(entries: list[Mapping[str, Any]]) -> str: material = bytearray() for entry in sorted(entries, key=lambda item: str(item["path"])): material.extend(str(entry["path"]).encode("utf-8")) material.extend(b"\0") material.extend(str(entry["bytes"]).encode("ascii")) material.extend(b"\0") material.extend(str(entry["sha256"]).encode("ascii")) material.extend(b"\n") return hashlib.sha256(bytes(material)).hexdigest() def verify_source_tree(source_root: Path) -> dict[str, Any]: root = _safe_root(source_root) manifest_path = root / SOURCE_MANIFEST_NAME if not manifest_path.is_file() or _is_reparse(manifest_path): raise IdentityError("E_LOCAL_EXTENSION_IDENTITY", "source manifest is missing or unsafe") raw = manifest_path.read_bytes() value = loads(raw) manifest = _exact_keys( value, {"schema", "extension_id", "root", "entries", "payload_tree_sha256", "manifest_semantic_sha256"}, "source manifest", ) if manifest["schema"] != 1 or manifest["extension_id"] != EXTENSION_ID: raise IdentityError("E_LOCAL_EXTENSION_IDENTITY", "source manifest identity differs") entries = manifest["entries"] if not isinstance(entries, list) or not entries: raise IdentityError("E_LOCAL_EXTENSION_IDENTITY", "source manifest entries are invalid") expected_paths: set[str] = set() for item in entries: entry = _exact_keys(item, {"path", "bytes", "sha256"}, "source entry") path = entry["path"] if not isinstance(path, str) or not path or "\\" in path or path.startswith("/") or ".." in Path(path).parts: raise IdentityError("E_LOCAL_EXTENSION_IDENTITY", "source manifest path is unsafe") if path in expected_paths or not isinstance(entry["bytes"], int) or not LOWER_SHA256.fullmatch(str(entry["sha256"])): raise IdentityError("E_LOCAL_EXTENSION_IDENTITY", "source manifest entry is invalid") expected_paths.add(path) actual_paths: set[str] = set() actual_entries: list[dict[str, Any]] = [] for candidate in root.rglob("*"): relative = candidate.relative_to(root).as_posix() if relative == SOURCE_MANIFEST_NAME: continue if _is_reparse(candidate) or not candidate.is_file(): raise IdentityError("E_LOCAL_EXTENSION_IDENTITY", "source tree contains a non-regular entry") actual_paths.add(relative) payload = candidate.read_bytes() actual_entries.append({"path": relative, "bytes": len(payload), "sha256": hashlib.sha256(payload).hexdigest()}) if actual_paths != expected_paths: raise IdentityError("E_LOCAL_EXTENSION_IDENTITY", "source tree exact set differs") expected_by_path = {str(item["path"]): item for item in entries} for actual in actual_entries: if actual != expected_by_path[actual["path"]]: raise IdentityError("E_LOCAL_EXTENSION_IDENTITY", "source file identity differs") if _tree_hash(actual_entries) != manifest["payload_tree_sha256"]: raise IdentityError("E_LOCAL_EXTENSION_IDENTITY", "payload tree identity differs") semantic = dict(manifest) semantic["manifest_semantic_sha256"] = None if hashlib.sha256(canonical_bytes(semantic)).hexdigest() != manifest["manifest_semantic_sha256"]: raise IdentityError("E_LOCAL_EXTENSION_IDENTITY", "manifest semantic identity differs") return { "root": root, "manifest_bytes": len(raw), "manifest_sha256": hashlib.sha256(raw).hexdigest(), "payload_tree_sha256": manifest["payload_tree_sha256"], } def verify_local_identity(source_root: Path, facts: Mapping[str, Any]) -> VerifiedLocalIdentity: facts = _exact_keys(facts, {"source_approval", "load_approval", "host"}, "identity facts") source = verify_source_tree(source_root) source_approval = _exact_keys( facts["source_approval"], { "schema", "task_id", "scope", "source_manifest_path", "source_manifest_bytes", "source_manifest_sha256", "payload_tree_sha256", "manifest_bytes", "manifest_sha256", "extension_id", "approved_by_role", "review_audit_id", "created_at", }, "source approval", ) if ( source_approval["schema"] != 1 or source_approval["task_id"] != TASK_ID or source_approval["scope"] != "controlled-local-unpacked-source-manifest" or source_approval["approved_by_role"] != "dev.reviewer.project" or Path(str(source_approval["source_manifest_path"])) != source["root"] / SOURCE_MANIFEST_NAME ): raise IdentityError("E_TRUSTED_ADAPTER_UNAVAILABLE", "reviewer source approval is unavailable") manifest_payload = (source["root"] / "manifest.json").read_bytes() manifest_sha = hashlib.sha256(manifest_payload).hexdigest() required_source = { "source_manifest_bytes": source["manifest_bytes"], "source_manifest_sha256": source["manifest_sha256"], "payload_tree_sha256": source["payload_tree_sha256"], "manifest_bytes": len(manifest_payload), "manifest_sha256": manifest_sha, "extension_id": EXTENSION_ID, } if any(source_approval[key] != value for key, value in required_source.items()): raise IdentityError("E_LOCAL_EXTENSION_IDENTITY", "reviewer source approval binding differs") load = _exact_keys( facts["load_approval"], { "schema", "task_id", "scope", "account_holder_confirmed", "observed_extension_id", "observed_name", "observed_version", "observed_enabled", "observed_error_count", "source_root_absolute", "source_manifest_bytes", "source_manifest_sha256", "payload_tree_sha256", "host_install_receipt_bytes", "host_install_receipt_sha256", "host_manifest_sha256", "chrome_parent_pid", "chrome_parent_path_sha256", "chrome_parent_signature_verified", "chrome_parent_started_at", "observed_at", "approved_by_role", "approval_reason", }, "load approval", ) if ( load["schema"] != 1 or load["task_id"] != TASK_ID or load["scope"] != "local-unpacked-load" or load["approved_by_role"] not in {"project.admin", "case_analysis.video_downloader"} or load["account_holder_confirmed"] is not True or load["observed_enabled"] is not True or load["observed_error_count"] != 0 or load["observed_extension_id"] != EXTENSION_ID or load["observed_name"] != EXTENSION_NAME or load["observed_version"] != EXTENSION_VERSION or Path(str(load["source_root_absolute"])) != source["root"] or load["source_manifest_bytes"] != source["manifest_bytes"] or load["source_manifest_sha256"] != source["manifest_sha256"] or load["payload_tree_sha256"] != source["payload_tree_sha256"] or load["chrome_parent_signature_verified"] is not True ): raise IdentityError("E_LOCAL_EXTENSION_IDENTITY", "local unpacked load approval differs") host = _exact_keys( facts["host"], { "host_name", "allowed_origins", "install_receipt_bytes", "install_receipt_sha256", "host_manifest_sha256", "chrome_parent_pid", "chrome_parent_path_sha256", "chrome_parent_signature_verified", "chrome_parent_started_at", }, "host identity", ) if ( host["host_name"] != HOST_NAME or host["allowed_origins"] != [EXTENSION_ORIGIN] or host["install_receipt_bytes"] != load["host_install_receipt_bytes"] or host["install_receipt_sha256"] != load["host_install_receipt_sha256"] or host["host_manifest_sha256"] != load["host_manifest_sha256"] or host["chrome_parent_pid"] != load["chrome_parent_pid"] or host["chrome_parent_path_sha256"] != load["chrome_parent_path_sha256"] or host["chrome_parent_signature_verified"] is not True or host["chrome_parent_started_at"] != load["chrome_parent_started_at"] ): raise IdentityError("E_LOCAL_EXTENSION_IDENTITY", "host or current Chrome identity differs") return VerifiedLocalIdentity( source_root=source["root"], source_manifest_bytes=source["manifest_bytes"], source_manifest_sha256=source["manifest_sha256"], payload_tree_sha256=source["payload_tree_sha256"], extension_id=EXTENSION_ID, host_install_receipt_sha256=str(load["host_install_receipt_sha256"]), chrome_parent_pid=int(load["chrome_parent_pid"]), )