test(v2): define secure control-plane contracts

tests/security/__init__.py

@@ -0,0 +1 @@
+"""Security tests."""

tests/security/test_auth.py

@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+import importlib
+
+import pytest
+
+CANARY_PASSWORD = "password-canary-4dd52a"
+CANARY_SECRET = "secret-canary-9a73ce"
+
+
+@pytest.mark.asyncio
+async def test_password_hash_secret_and_token_are_never_disclosed(app_client) -> None:
+    client, settings = app_client
+    setup = await client.post(
+        "/api/v2/setup", json={"username": "admin", "password": CANARY_PASSWORD}
+    )
+    assert setup.status_code == 201
+    csrf = client.cookies["backup_tool_csrf"]
+    secret = await client.post(
+        "/api/v2/admin/secrets",
+        json={"purpose": "database", "value": CANARY_SECRET},
+        headers={"X-CSRF-Token": csrf},
+    )
+    assert secret.status_code == 201
+    token_response = await client.post(
+        "/api/v2/auth/tokens",
+        json={"scopes": ["audit:read"], "expires_at": None},
+        headers={"X-CSRF-Token": csrf, "Idempotency-Key": "canary-token"},
+    )
+    token = token_response.json()["token"]
+
+    responses = [
+        setup,
+        secret,
+        await client.get("/api/v2/auth/session"),
+        await client.get("/api/v2/admin/secrets"),
+        await client.get("/api/v2/audit", params={"limit": 100}),
+        await client.get("/openapi.json"),
+    ]
+    combined = "\n".join(response.text for response in responses)
+    assert CANARY_PASSWORD not in combined
+    assert CANARY_SECRET not in combined
+    assert token not in combined
+
+    models = importlib.import_module("backup_tool.db.models")
+    engine_module = importlib.import_module("backup_tool.db.engine")
+    sqlalchemy = importlib.import_module("sqlalchemy")
+    async_sessionmaker = importlib.import_module("sqlalchemy.ext.asyncio").async_sessionmaker
+    engine = engine_module.create_engine(settings)
+    sessions = async_sessionmaker(engine, expire_on_commit=False)
+    try:
+        async with sessions() as session:
+            user = await session.scalar(sqlalchemy.select(models.User))
+            stored_secret = await session.scalar(sqlalchemy.select(models.Secret))
+            stored_token = await session.scalar(sqlalchemy.select(models.ApiToken))
+            audits = list((await session.scalars(sqlalchemy.select(models.AuditEvent))).all())
+        assert user is not None and user.password_hash != CANARY_PASSWORD
+        assert user.password_hash.startswith("$argon2id$")
+        assert stored_secret is not None and CANARY_SECRET.encode() not in stored_secret.ciphertext
+        assert stored_token is not None and stored_token.token_hash != token
+        assert all(CANARY_PASSWORD not in str(audit.details) for audit in audits)
+        assert all(CANARY_SECRET not in str(audit.details) for audit in audits)
+        assert all(token not in str(audit.details) for audit in audits)
+    finally:
+        await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_master_key_file_never_appears_in_problem_detail(app_client) -> None:
+    client, settings = app_client
+    response = await client.post(
+        "/api/v2/auth/login", json={"username": "missing", "password": "bad"}
+    )
+    assert response.status_code == 401
+    assert str(settings.master_key_file) not in response.text

tests/security/test_leakage_scan.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+import subprocess
+import sys
+
+
+def test_leakage_scan_rejects_canary_and_accepts_clean_file(tmp_path) -> None:
+    path = tmp_path / "output.txt"
+    path.write_text("safe output", encoding="utf-8")
+    clean = subprocess.run(
+        [sys.executable, "tools/leakage_scan.py", "--canary", "secret-canary", str(path)],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert clean.returncode == 0
+
+    path.write_text("oops secret-canary escaped", encoding="utf-8")
+    leaked = subprocess.run(
+        [sys.executable, "tools/leakage_scan.py", "--canary", "secret-canary", str(path)],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert leaked.returncode == 1
+    assert "output.txt" in leaked.stdout