test(credentials): 登记 B-02 生产验收

scripts/acceptance_cases/b02_credentials.py

@@ -0,0 +1,89 @@
+"""B-02 Fernet-to-Stronghold migration acceptance driver."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import shutil
+import subprocess
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+TEST = "credentials::tests::b02_fernet_migration_matrix_is_atomic_verified_and_idempotent"
+
+
+def sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as source:
+        for chunk in iter(lambda: source.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--config", required=True)
+    parser.add_argument("--output", required=True)
+    args = parser.parse_args()
+    case_id = os.environ.get("OPENNEXUS_ACCEPTANCE_CASE_ID", "")
+    output = Path(args.output)
+    output.parent.mkdir(parents=True, exist_ok=True)
+    cargo = shutil.which("cargo")
+    if case_id != "B-02" or cargo is None:
+        result = {
+            "schema": 1,
+            "case_id": case_id,
+            "status": "FAILED",
+            "reason": "B-02 requires the registered case ID and Cargo.",
+            "assertions": [{"name": "driver prerequisites", "status": "FAILED", "evidence": "case ID or Cargo missing"}],
+            "metrics": {"peak_rss_bytes": None, "max_process_count": None, "denied_access_count": None},
+            "files": [],
+            "revisions": [],
+        }
+        output.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+        return 1
+    command = [
+        cargo,
+        "test",
+        "--manifest-path",
+        str(ROOT / "frontend" / "src-tauri" / "Cargo.toml"),
+        "--locked",
+        "--lib",
+        TEST,
+        "--",
+        "--exact",
+        "--nocapture",
+    ]
+    completed = subprocess.run(command, cwd=ROOT, capture_output=True, text=True, check=False)
+    print(completed.stdout, end="")
+    print(completed.stderr, end="")
+    passed = completed.returncode == 0 and "1 passed; 0 failed" in completed.stdout
+    status = "PASSED" if passed else "FAILED"
+    evidence = f"cargo exact test {TEST}"
+    assertions = [
+        {"name": "100 records preserve IDs and decrypted values", "status": status, "evidence": evidence},
+        {"name": "local and environment master keys are handled without source mutation", "status": status, "evidence": evidence},
+        {"name": "three repeat imports add no records", "status": status, "evidence": evidence},
+        {"name": "empty source switches with zero records", "status": status, "evidence": evidence},
+        {"name": "missing key and bad token do not switch or mutate source/target", "status": status, "evidence": evidence},
+        {"name": "same target ID with a different value rejects the full migration", "status": status, "evidence": evidence},
+    ]
+    fixture = ROOT / "frontend" / "src-tauri" / "tests" / "fixtures" / "fernet-python.json"
+    result = {
+        "schema": 1,
+        "case_id": case_id,
+        "status": status,
+        "reason": "" if passed else "The exact Rust B-02 acceptance oracle failed.",
+        "assertions": assertions,
+        "metrics": {"peak_rss_bytes": None, "max_process_count": None, "denied_access_count": None},
+        "files": [{"path": "frontend/src-tauri/tests/fixtures/fernet-python.json", "sha256": sha256(fixture)}],
+        "revisions": [],
+    }
+    output.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    return 0 if passed else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/phase3_acceptance.py

@@ -31,7 +31,13 @@ CASE_SUITES = {
 ALL_CASES = tuple(case for cases in CASE_SUITES.values() for case in cases)
 # A case becomes executable only when a repository-owned driver is registered here.
 # Component/unit test commands are deliberately not treated as production acceptance.
-CASE_DRIVERS: dict[str, dict[str, Any]] = {}
+CASE_DRIVERS: dict[str, dict[str, Any]] = {
+    "B-02": {
+        "driver": "scripts/acceptance_cases/b02_credentials.py",
+        "timeout_seconds": 900,
+        "required_metrics": (),
+    },
+}
 ENV_NAME = re.compile(r"[A-Z][A-Z0-9_]{2,127}")
 RUN_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{2,63}")
 SENSITIVE_KEY = re.compile(r"(?:password|passwd|secret|token|api[_-]?key|credential)", re.I)
@@ -122,7 +128,7 @@ def load_config(path: Path, environ: dict[str, str] | None = None) -> dict[str,
     if config.get("isolated") is not True or config.get("allow_destructive") is not True:
         raise AcceptanceError("CONFIG_ISOLATION_CONFIRMATION_REQUIRED")
     run_id = config.get("run_id")
-    if not isinstance(run_id, str) or not RUN_ID.fullmatch(run_id):
+    if not isinstance(run_id, str) or not RUN_ID.fullmatch(run_id) or run_id.startswith("replace-"):
         raise AcceptanceError("CONFIG_RUN_ID_INVALID")
     profile = config.get("platform_profile")
     if not isinstance(profile, str) or not profile.strip():
@@ -420,6 +426,26 @@ def _repository_evidence(config: dict[str, Any]) -> dict[str, Any]:
     return {"commit": commit or None, "lock_sha256": locks, "artifact_sha256": artifacts}
 
 
+def repository_changes() -> tuple[str, ...]:
+    paths: set[str] = set()
+    commands = (
+        ["git", "diff", "--name-only", "-z", "--", ".", ":(exclude)backend/data/vault"],
+        ["git", "diff", "--cached", "--name-only", "-z", "--", ".", ":(exclude)backend/data/vault"],
+        ["git", "ls-files", "--others", "--exclude-standard", "-z"],
+    )
+    for command in commands:
+        completed = subprocess.run(command, cwd=ROOT, capture_output=True, check=False)
+        if completed.returncode != 0:
+            raise AcceptanceError("SOURCE_STATE_UNAVAILABLE")
+        for raw in completed.stdout.split(b"\0"):
+            if not raw:
+                continue
+            path = raw.decode("utf-8", errors="replace").replace("\\", "/")
+            if path != "backend/data/vault" and not path.startswith("backend/data/vault/"):
+                paths.add(path)
+    return tuple(sorted(paths))
+
+
 def execute(args: argparse.Namespace, environ: dict[str, str] | None = None) -> int:
     environ = os.environ if environ is None else environ
     if args.list_cases:
@@ -431,6 +457,10 @@ def execute(args: argparse.Namespace, environ: dict[str, str] | None = None) ->
     selected = select_cases(args.suite, args.case)
     config_path = Path(args.config).resolve()
     config = load_config(config_path, environ)
+    changes = repository_changes()
+    if changes:
+        preview = ",".join(changes[:10])
+        raise AcceptanceError(f"SOURCE_TREE_DIRTY: {preview}")
     if args.suite == "all" and not config.get("platform_profile"):
         raise AcceptanceError("CONFIG_PLATFORM_PROFILE_REQUIRED")
     prepare_isolated_root(config)