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"]),
|
)
|
