test(sidecar): 登记 A-02 生产验收

backend/tests/test_sidecar_auth.py

@@ -3,6 +3,7 @@ import json
 import os
 from pathlib import Path
 import queue
+import secrets
 import subprocess
 import sys
 import threading
@@ -51,6 +52,42 @@ def test_session_auth_covers_every_route_and_rejects_duplicate_headers():
     assert len(calls) == 6
 
 
+def test_session_auth_rejects_missing_wrong_and_old_generation_100_times_without_side_effects():
+    calls = []
+
+    async def app(scope, receive, send):
+        calls.append(scope["path"])
+        await send({"type": "http.response.start", "status": 204, "headers": []})
+
+    secret = "ab" * 32
+    generation = "cd" * 32
+    auth = SessionAuth(app, secret, generation, 4567)
+    host = (b"host", b"127.0.0.1:4567")
+    authorization = (b"authorization", f"Bearer {secret}".encode())
+    current = (b"x-core-generation", generation.encode())
+    cases = {
+        "missing": [host, current],
+        "wrong": [host, (b"authorization", ("Bearer " + "ef" * 32).encode()), current],
+        "old": [host, authorization, (b"x-core-generation", ("01" * 32).encode())],
+    }
+
+    async def request(headers):
+        messages = []
+
+        async def send(message):
+            messages.append(message)
+
+        await auth({"type": "http", "headers": headers, "path": "/api/notes"}, None, send)
+        return messages
+
+    for name, headers in cases.items():
+        for _ in range(100):
+            messages = asyncio.run(request(headers))
+            assert messages[0]["status"] == 401, name
+            assert json.loads(messages[1]["body"])["error"]["code"] == "AUTH_REQUIRED"
+    assert calls == []
+
+
 def test_handshake_proof_binds_port_pid_generation_and_challenge():
     args = ["01" * 32, "02" * 32, "03" * 32, 123, 4567, 123]
     expected = proof(*args)
@@ -62,8 +99,8 @@ def test_handshake_proof_binds_port_pid_generation_and_challenge():
 
 
 def test_real_sidecar_bootstrap_auth_and_parent_eof(tmp_path):
-    config = dict(protocol=1, secret="01" * 32, challenge="02" * 32,
+    config = dict(protocol=1, secret=secrets.token_hex(32), challenge=secrets.token_hex(32),
-                  generation="03" * 32, data_dir=str(tmp_path / "core"))
+                  generation=secrets.token_hex(32), data_dir=str(tmp_path / "core"))
     executable = os.environ.get("OPENNEXUS_CORE_TEST_BINARY")
     command = [executable] if executable else [sys.executable, "-m", "app.sidecar"]
     diagnostics = (tmp_path / "core-stderr.log").open("wb")
@@ -73,6 +110,8 @@ def test_real_sidecar_bootstrap_auth_and_parent_eof(tmp_path):
                                stderr=diagnostics,
                                creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
     try:
+        assert config["secret"] not in "\0".join(command)
+        assert config["secret"] not in "\0".join(os.environ.values())
         config["launcher_pid"] = process.pid
         process.stdin.write(json.dumps(config).encode() + b"\n")
         process.stdin.flush()
@@ -96,8 +135,22 @@ def test_real_sidecar_bootstrap_auth_and_parent_eof(tmp_path):
         })
         with opener.open(request, timeout=5) as response:
             assert json.load(response)["status"] == "ok"
+        for disabled in ("/docs", "/redoc", "/openapi.json", "/docs/oauth2-redirect"):
+            request = urllib.request.Request(url + disabled, headers={
+                "Authorization": "Bearer " + config["secret"],
+                "X-Core-Generation": config["generation"],
+            })
+            with pytest.raises(urllib.error.HTTPError) as error:
+                opener.open(request, timeout=5)
+            assert error.value.code == 404
         process.stdin.close()
         assert process.wait(timeout=10) == 0
+        diagnostics.flush()
+        planted = config["secret"].encode()
+        assert planted not in (tmp_path / "core-stderr.log").read_bytes()
+        for artifact in (tmp_path / "core").rglob("*"):
+            if artifact.is_file():
+                assert planted not in artifact.read_bytes(), artifact.name
     finally:
         if not process.stdin.closed:
             process.stdin.close()

scripts/acceptance_cases/a02_sidecar.py

@@ -0,0 +1,118 @@
+"""A-02 authenticated Core transport and secret-exposure 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]
+BACKEND = ROOT / "backend"
+MANIFEST = ROOT / "frontend" / "src-tauri" / "Cargo.toml"
+PYTHON = BACKEND / (".venv/Scripts/python.exe" if os.name == "nt" else ".venv/bin/python")
+PYTHON_TESTS = (
+    "tests/test_sidecar_auth.py::test_session_auth_covers_every_route_and_rejects_duplicate_headers",
+    "tests/test_sidecar_auth.py::test_session_auth_rejects_missing_wrong_and_old_generation_100_times_without_side_effects",
+    "tests/test_sidecar_auth.py::test_real_sidecar_bootstrap_auth_and_parent_eof",
+)
+RUST_TESTS = (
+    (
+        "--test",
+        "core_process",
+        "real_python_core_authenticates_and_rotates_generation",
+    ),
+    (
+        "--bin",
+        "notesagent-desktop",
+        "core_proxy_tests::request_dto_accepts_camel_case_and_rejects_unowned_headers",
+    ),
+)
+
+
+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 run(command: list[str], cwd: Path, expected: str) -> bool:
+    completed = subprocess.run(command, cwd=cwd, capture_output=True, text=True, check=False)
+    print(completed.stdout, end="")
+    print(completed.stderr, end="")
+    return completed.returncode == 0 and expected in completed.stdout
+
+
+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)
+    assertions = [
+        {"name": "outermost authentication covers health, API, SSE, binary, docs, and unknown routes"},
+        {"name": "one hundred missing, wrong, and old-generation requests each return 401 with zero business calls"},
+        {"name": "duplicate authorization, browser Origin, and wrong Host are denied"},
+        {"name": "real Core binds a random port, rotates generations, and keeps docs disabled under valid auth"},
+        {"name": "session material is absent from URL, argv, environment, logs, data files, and WebView-owned DTO fields"},
+    ]
+    cargo = shutil.which(os.environ.get("CARGO", "cargo"))
+    passed = case_id == "A-02" and PYTHON.is_file() and cargo is not None
+    if passed:
+        passed = run([str(PYTHON), "-m", "pytest", *PYTHON_TESTS, "-q"], BACKEND, "3 passed")
+        for selector, target, test in RUST_TESTS:
+            command = [
+                cargo,
+                "test",
+                "--manifest-path",
+                str(MANIFEST),
+                "--locked",
+                "--features",
+                "desktop",
+                selector,
+                target,
+                test,
+                "--",
+                "--exact",
+                "--nocapture",
+            ]
+            passed = run(command, ROOT, "1 passed; 0 failed") and passed
+    status = "PASSED" if passed else "FAILED"
+    evidence = "three exact Python Sidecar tests and two exact Rust Host transport tests"
+    for assertion in assertions:
+        assertion.update({"status": status, "evidence": evidence})
+    files = []
+    for relative in (
+        "backend/app/sidecar.py",
+        "backend/tests/test_sidecar_auth.py",
+        "frontend/src-tauri/src/core.rs",
+        "frontend/src-tauri/src/main.rs",
+        "frontend/src-tauri/tests/core_process.rs",
+    ):
+        files.append({"path": relative, "sha256": sha256(ROOT / relative)})
+    payload = {
+        "schema": 1,
+        "case_id": case_id,
+        "status": status,
+        "reason": "" if passed else "An A-02 Sidecar authentication or Host transport oracle failed.",
+        "assertions": assertions,
+        "metrics": {"peak_rss_bytes": None, "max_process_count": None, "denied_access_count": None},
+        "files": files,
+        "revisions": [
+            {"scope": "unauthenticated rejection matrix", "missing": 100, "wrong": 100, "old_generation": 100},
+            {"scope": "disabled documentation routes", "route_count": 4},
+        ],
+    }
+    output.write_text(json.dumps(payload, 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

@@ -32,6 +32,11 @@ 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]] = {
+    "A-02": {
+        "driver": "scripts/acceptance_cases/a02_sidecar.py",
+        "timeout_seconds": 900,
+        "required_metrics": (),
+    },
     "B-01": {
         "driver": "scripts/acceptance_cases/b01_credentials.py",
         "timeout_seconds": 900,