From b823c6edb797faf94b30c6da17dc3bcb0ffecaef Mon Sep 17 00:00:00 2001 From: Alex Blank Date: Mon, 27 Jul 2026 20:13:08 +0200 Subject: [PATCH] feat(v2): add secure control plane --- Makefile | 10 +- backend/pyproject.toml | 2 + backend/src/backup_tool/api/__init__.py | 1 + backend/src/backup_tool/api/app.py | 479 ++++++++++++++++++ backend/src/backup_tool/security/__init__.py | 1 + backend/src/backup_tool/security/auth.py | 72 +++ backend/src/backup_tool/security/redaction.py | 41 ++ backend/src/backup_tool/security/secrets.py | 36 ++ docs/release/m2-evidence.md | 19 +- tools/forbidden_v1_scan.py | 2 +- tools/leakage_scan.py | 30 ++ 11 files changed, 688 insertions(+), 5 deletions(-) create mode 100644 backend/src/backup_tool/api/__init__.py create mode 100644 backend/src/backup_tool/api/app.py create mode 100644 backend/src/backup_tool/security/__init__.py create mode 100644 backend/src/backup_tool/security/auth.py create mode 100644 backend/src/backup_tool/security/redaction.py create mode 100644 backend/src/backup_tool/security/secrets.py create mode 100644 tools/leakage_scan.py diff --git a/Makefile b/Makefile index a7702d1..6f43bb4 100644 --- a/Makefile +++ b/Makefile @@ -2,7 +2,7 @@ PYTHON ?= .venv/bin/python BOOTSTRAP_PYTHON ?= python3 NPM ?= npm -.PHONY: setup install test-fast lint typecheck frontend-build check-v1-absent check +.PHONY: setup install test-fast test-integration test-security lint typecheck frontend-build check-v1-absent check setup: test -x $(PYTHON) || $(BOOTSTRAP_PYTHON) -m venv .venv @@ -15,6 +15,12 @@ install: setup test-fast: $(PYTHON) -m pytest tests/unit tests/contract -q +test-integration: + $(PYTHON) -m pytest tests/integration -q + +test-security: + $(PYTHON) -m pytest tests/security -q + lint: $(PYTHON) -m ruff check --config backend/pyproject.toml backend/src backend/alembic tests tools $(PYTHON) -m ruff format --check --config backend/pyproject.toml backend/src backend/alembic tests tools @@ -29,4 +35,4 @@ frontend-build: check-v1-absent: $(PYTHON) tools/forbidden_v1_scan.py . -check: check-v1-absent test-fast lint typecheck frontend-build +check: check-v1-absent test-fast test-integration test-security lint typecheck frontend-build diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 19b63cd..33fec84 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -13,7 +13,9 @@ dependencies = [ "aiofiles==25.1.0", "aiosqlite==0.22.1", "alembic==1.18.5", + "argon2-cffi==25.1.0", "apscheduler==3.11.3", + "cryptography==49.0.0", "fastapi==0.136.1", "httpx==0.28.1", "paramiko==5.0.0", diff --git a/backend/src/backup_tool/api/__init__.py b/backend/src/backup_tool/api/__init__.py new file mode 100644 index 0000000..0d2b7b6 --- /dev/null +++ b/backend/src/backup_tool/api/__init__.py @@ -0,0 +1 @@ +"""HTTP boundary for Backup Tool v2.""" diff --git a/backend/src/backup_tool/api/app.py b/backend/src/backup_tool/api/app.py new file mode 100644 index 0000000..f871e8a --- /dev/null +++ b/backend/src/backup_tool/api/app.py @@ -0,0 +1,479 @@ +import asyncio +import base64 +import hashlib +from collections.abc import AsyncIterator +from datetime import UTC, datetime +from typing import Annotated, Any, cast + +from fastapi import Depends, FastAPI, Header, Request, Response +from fastapi.exceptions import RequestValidationError +from fastapi.responses import JSONResponse +from pydantic import BaseModel, Field +from sqlalchemy import desc, select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from backup_tool.cli import build_alembic_config +from backup_tool.config import Settings +from backup_tool.db.engine import SchemaNotCurrentError, assert_schema_current, create_engine +from backup_tool.db.models import ApiToken, AuditEvent, IdempotencyRecord, Secret, User +from backup_tool.security.auth import ( + hash_password, + hash_token, + new_csrf_token, + new_token, + sign_session, + verify_password, + verify_session, +) +from backup_tool.security.redaction import redact +from backup_tool.security.secrets import EnvelopeCipher + + +class Problem(Exception): + def __init__(self, status: int, code: str, detail: str): + self.status = status + self.code = code + self.detail = detail + + +def problem_response(request: Request, status: int, code: str, detail: str) -> JSONResponse: + return JSONResponse( + status_code=status, + media_type="application/problem+json", + content={ + "type": f"https://backup-tool.invalid/problems/{code}", + "title": code.replace("_", " ").title(), + "status": status, + "detail": detail, + "instance": str(request.url.path), + "code": code, + }, + ) + + +class SetupInput(BaseModel): + username: str = Field(min_length=1, max_length=255) + password: str = Field(min_length=12, max_length=1024) + + +class LoginInput(BaseModel): + username: str = Field(min_length=1, max_length=255) + password: str = Field(min_length=1, max_length=1024) + + +class TokenInput(BaseModel): + scopes: list[str] = Field(min_length=1) + expires_at: datetime | None = None + + +class SecretInput(BaseModel): + purpose: str = Field(min_length=1, max_length=64) + value: str = Field(min_length=1, max_length=65536) + + +class UserPatch(BaseModel): + state: str + + +def _etag(user: User) -> str: + return f'"{user.id}:{user.updated_at.isoformat()}"' + + +def _cursor(item_id: str) -> str: + return base64.urlsafe_b64encode(item_id.encode()).decode().rstrip("=") + + +def _decode_cursor(value: str) -> str: + try: + return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4)).decode() + except (ValueError, UnicodeDecodeError) as error: + raise Problem(400, "invalid_cursor", "Cursor is invalid.") from error + + +def _digest_request(payload: TokenInput) -> str: + value = payload.model_dump(mode="json") + encoded = repr(sorted(value.items())).encode() + return hashlib.sha256(encoded).hexdigest() + + +class SessionDependency: + def __init__(self, factory: async_sessionmaker[AsyncSession]): + self._factory = factory + + async def __call__(self) -> AsyncIterator[AsyncSession]: + async with self._factory() as db: + yield db + + +def create_app(settings: Settings) -> FastAPI: + app = FastAPI(title="Backup Tool API", version="2.0.0", docs_url=None, redoc_url=None) + app.state.settings = settings + app.state.engine = create_engine(settings) + app.state.sessions = async_sessionmaker(app.state.engine, expire_on_commit=False) + app.state.cipher = EnvelopeCipher.from_file(settings.master_key_file) + app.state.setup_lock = asyncio.Lock() + + @app.middleware("http") + async def request_id_middleware(request: Request, call_next: Any) -> Response: + from backup_tool.ids import new_uuid7 + + request.state.request_id = str(new_uuid7()) + response = cast(Response, await call_next(request)) + response.headers["X-Request-ID"] = request.state.request_id + return response + + @app.exception_handler(Problem) + async def handle_problem(request: Request, error: Problem) -> JSONResponse: + return problem_response(request, error.status, error.code, error.detail) + + @app.exception_handler(RequestValidationError) + async def handle_validation(request: Request, _error: RequestValidationError) -> JSONResponse: + return problem_response(request, 422, "validation_failed", "Request validation failed.") + + session = SessionDependency(app.state.sessions) + + async def actor( + request: Request, + db: Annotated[AsyncSession, Depends(session)], + authorization: Annotated[str | None, Header()] = None, + ) -> tuple[User, set[str], bool]: + if authorization and authorization.startswith("Bearer "): + supplied = authorization.removeprefix("Bearer ") + token = await db.scalar( + select(ApiToken).where(ApiToken.token_hash == hash_token(supplied)) + ) + if token is None or token.revoked_at is not None: + raise Problem(401, "authentication_required", "Authentication is required.") + if token.expires_at is not None and token.expires_at <= datetime.now(UTC): + raise Problem(401, "authentication_required", "Authentication is required.") + user = await db.get(User, token.owner_id) + if user is None or user.state != "active": + raise Problem(401, "authentication_required", "Authentication is required.") + return user, set(token.scopes), False + encoded = request.cookies.get("backup_tool_session") + data = verify_session(encoded, settings.master_key_file) if encoded else None + if data is None: + raise Problem(401, "authentication_required", "Authentication is required.") + user = await db.get(User, data["sub"]) + if user is None or user.state != "active": + raise Problem(401, "authentication_required", "Authentication is required.") + return user, {"*"}, True + + async def require( + request: Request, + db: Annotated[AsyncSession, Depends(session)], + authorization: Annotated[str | None, Header()] = None, + csrf: Annotated[str | None, Header(alias="X-CSRF-Token")] = None, + ) -> tuple[User, set[str], bool]: + user, scopes, cookie_auth = await actor(request, db, authorization) + if cookie_auth: + data = verify_session( + request.cookies.get("backup_tool_session", ""), settings.master_key_file + ) + if data is None or csrf is None or csrf != data.get("csrf"): + raise Problem(403, "csrf_failed", "CSRF validation failed.") + return user, scopes, cookie_auth + + async def audit( + db: AsyncSession, + request: Request, + action: str, + resource_type: str, + resource_id: str | None, + outcome: str, + actor_id: str | None, + details: dict[str, Any] | None = None, + ) -> None: + db.add( + AuditEvent( + actor_id=actor_id, + action=action, + resource_type=resource_type, + resource_id=resource_id, + outcome=outcome, + request_id=request.state.request_id, + details=redact(details or {}), + ) + ) + + def set_session(response: Response, user_id: str) -> None: + csrf = new_csrf_token() + response.set_cookie( + "backup_tool_session", + sign_session(user_id, csrf, settings.master_key_file), + httponly=True, + secure=True, + samesite="strict", + path="/", + ) + response.set_cookie( + "backup_tool_csrf", csrf, httponly=False, secure=True, samesite="strict", path="/" + ) + + @app.get("/livez") + async def livez() -> dict[str, str]: + return {"status": "alive"} + + @app.get("/readyz") + async def readyz(db: Annotated[AsyncSession, Depends(session)]) -> dict[str, str]: + try: + await assert_schema_current(app.state.engine, build_alembic_config(settings)) + except SchemaNotCurrentError as error: + raise Problem(503, "schema_not_current", "Metadata schema is not current.") from error + if await db.scalar(select(User.id).limit(1)) is None: + raise Problem(503, "setup_required", "Initial administrator setup is required.") + return {"status": "ready"} + + @app.post("/api/v2/setup", status_code=201) + async def setup( + input_: SetupInput, + request: Request, + response: Response, + db: Annotated[AsyncSession, Depends(session)], + ) -> dict[str, str]: + async with app.state.setup_lock: + if await db.scalar(select(User.id).limit(1)) is not None: + raise Problem(409, "setup_complete", "Initial administrator already exists.") + user = User(username=input_.username, password_hash=hash_password(input_.password)) + db.add(user) + try: + await db.flush() + except IntegrityError as error: + await db.rollback() + raise Problem( + 409, "setup_complete", "Initial administrator already exists." + ) from error + await audit(db, request, "setup", "user", user.id, "success", user.id) + await db.commit() + set_session(response, user.id) + return {"id": user.id, "username": user.username} + + @app.post("/api/v2/auth/login") + async def login( + input_: LoginInput, + request: Request, + response: Response, + db: Annotated[AsyncSession, Depends(session)], + ) -> dict[str, str]: + user = await db.scalar(select(User).where(User.username == input_.username)) + if ( + user is None + or user.state != "active" + or not verify_password(user.password_hash, input_.password) + ): + await audit(db, request, "login", "user", None, "denied", None) + await db.commit() + raise Problem(401, "authentication_failed", "Invalid credentials.") + await audit(db, request, "login", "user", user.id, "success", user.id) + await db.commit() + set_session(response, user.id) + return {"id": user.id, "username": user.username} + + @app.post("/api/v2/auth/logout", status_code=204) + async def logout( + response: Response, _: Annotated[tuple[User, set[str], bool], Depends(require)] + ) -> None: + response.delete_cookie("backup_tool_session", path="/") + response.delete_cookie("backup_tool_csrf", path="/") + + @app.get("/api/v2/auth/session") + async def get_session( + identity: Annotated[tuple[User, set[str], bool], Depends(actor)], + ) -> dict[str, str]: + user, _, _ = identity + return {"id": user.id, "username": user.username, "state": user.state} + + @app.post("/api/v2/auth/tokens") + async def create_token( + input_: TokenInput, + request: Request, + db: Annotated[AsyncSession, Depends(session)], + identity: Annotated[tuple[User, set[str], bool], Depends(require)], + idempotency_key: Annotated[str | None, Header(alias="Idempotency-Key")] = None, + ) -> JSONResponse: + user, scopes, _ = identity + if "*" not in scopes: + raise Problem(403, "insufficient_scope", "Required scope is missing.") + if not idempotency_key: + raise Problem(400, "idempotency_key_required", "Idempotency-Key is required.") + digest = _digest_request(input_) + existing = await db.scalar( + select(IdempotencyRecord).where( + IdempotencyRecord.actor_id == user.id, + IdempotencyRecord.key == idempotency_key, + IdempotencyRecord.operation == "create_api_token", + ) + ) + if existing is not None: + if existing.request_digest != digest: + raise Problem( + 409, "idempotency_mismatch", "Idempotency-Key was used for another request." + ) + return JSONResponse( + {"id": existing.response_resource_id, "token": None}, status_code=200 + ) + raw = new_token() + token = ApiToken( + owner_id=user.id, + token_hash=hash_token(raw), + scopes=input_.scopes, + expires_at=input_.expires_at, + ) + db.add(token) + await db.flush() + db.add( + IdempotencyRecord( + actor_id=user.id, + key=idempotency_key, + operation="create_api_token", + request_digest=digest, + response_resource_type="api_token", + response_resource_id=token.id, + ) + ) + await audit(db, request, "create", "api_token", token.id, "success", user.id) + await db.commit() + return JSONResponse({"id": token.id, "token": raw}, status_code=201) + + @app.delete("/api/v2/auth/tokens/{token_id}", status_code=204) + async def revoke_token( + token_id: str, + request: Request, + db: Annotated[AsyncSession, Depends(session)], + identity: Annotated[tuple[User, set[str], bool], Depends(require)], + ) -> None: + user, scopes, _ = identity + if "*" not in scopes: + raise Problem(403, "insufficient_scope", "Required scope is missing.") + token = await db.get(ApiToken, token_id) + if token is None or token.owner_id != user.id: + raise Problem(404, "resource_not_found", "API token was not found.") + token.revoked_at = datetime.now(UTC) + await audit(db, request, "revoke", "api_token", token.id, "success", user.id) + await db.commit() + + @app.post("/api/v2/admin/secrets", status_code=201) + async def create_secret( + input_: SecretInput, + request: Request, + db: Annotated[AsyncSession, Depends(session)], + identity: Annotated[tuple[User, set[str], bool], Depends(require)], + ) -> dict[str, Any]: + user, scopes, _ = identity + if "*" not in scopes and "admin:write" not in scopes: + raise Problem(403, "insufficient_scope", "Required scope is missing.") + ciphertext, key_id = app.state.cipher.encrypt( + input_.value, purpose=input_.purpose, version=1 + ) + secret = Secret(ciphertext=ciphertext, key_id=key_id, purpose=input_.purpose) + db.add(secret) + await db.flush() + await audit( + db, + request, + "create", + "secret", + secret.id, + "success", + user.id, + {"purpose": input_.purpose}, + ) + await db.commit() + return { + "id": secret.id, + "purpose": secret.purpose, + "key_id": secret.key_id, + "version": secret.version, + } + + @app.get("/api/v2/admin/secrets") + async def list_secrets( + db: Annotated[AsyncSession, Depends(session)], + identity: Annotated[tuple[User, set[str], bool], Depends(actor)], + ) -> list[dict[str, Any]]: + _, scopes, _ = identity + if "*" not in scopes and "admin:read" not in scopes: + raise Problem(403, "insufficient_scope", "Required scope is missing.") + result = await db.scalars(select(Secret).order_by(desc(Secret.created_at))) + return [ + {"id": item.id, "purpose": item.purpose, "key_id": item.key_id, "version": item.version} + for item in result + ] + + @app.get("/api/v2/admin/users/{user_id}") + async def get_user( + user_id: str, + response: Response, + db: Annotated[AsyncSession, Depends(session)], + _: Annotated[tuple[User, set[str], bool], Depends(actor)], + ) -> dict[str, str]: + user = await db.get(User, user_id) + if user is None: + raise Problem(404, "resource_not_found", "User was not found.") + response.headers["ETag"] = _etag(user) + return {"id": user.id, "username": user.username, "state": user.state} + + @app.patch("/api/v2/admin/users/{user_id}") + async def patch_user( + user_id: str, + input_: UserPatch, + request: Request, + response: Response, + db: Annotated[AsyncSession, Depends(session)], + identity: Annotated[tuple[User, set[str], bool], Depends(require)], + if_match: Annotated[str | None, Header(alias="If-Match")] = None, + ) -> dict[str, str]: + actor_user, scopes, _ = identity + if "*" not in scopes and "admin:write" not in scopes: + raise Problem(403, "insufficient_scope", "Required scope is missing.") + user = await db.get(User, user_id) + if user is None: + raise Problem(404, "resource_not_found", "User was not found.") + if if_match != _etag(user): + raise Problem(412, "etag_mismatch", "Resource was modified by another request.") + if input_.state not in {"active", "disabled"}: + raise Problem(422, "validation_failed", "State is invalid.") + user.state = input_.state + user.updated_at = datetime.now(UTC) + await audit(db, request, "update", "user", user.id, "success", actor_user.id) + await db.commit() + await db.refresh(user) + response.headers["ETag"] = _etag(user) + return {"id": user.id, "username": user.username, "state": user.state} + + @app.get("/api/v2/audit") + async def list_audit( + db: Annotated[AsyncSession, Depends(session)], + identity: Annotated[tuple[User, set[str], bool], Depends(actor)], + limit: int = 50, + cursor: str | None = None, + ) -> dict[str, Any]: + _, scopes, _ = identity + if "*" not in scopes and "audit:read" not in scopes: + raise Problem(403, "insufficient_scope", "Required scope is missing.") + if not 1 <= limit <= 100: + raise Problem(422, "validation_failed", "Limit must be between 1 and 100.") + statement = select(AuditEvent).order_by(desc(AuditEvent.id)).limit(limit + 1) + if cursor: + statement = statement.where(AuditEvent.id < _decode_cursor(cursor)) + items = list((await db.scalars(statement)).all()) + page, remainder = items[:limit], items[limit:] + return { + "items": [ + { + "id": item.id, + "action": item.action, + "resource_type": item.resource_type, + "resource_id": item.resource_id, + "outcome": item.outcome, + "request_id": item.request_id, + "created_at": item.created_at.isoformat(), + "details": redact(item.details), + } + for item in page + ], + "next_cursor": _cursor(page[-1].id) if page and remainder else None, + } + + return app diff --git a/backend/src/backup_tool/security/__init__.py b/backend/src/backup_tool/security/__init__.py new file mode 100644 index 0000000..37c890a --- /dev/null +++ b/backend/src/backup_tool/security/__init__.py @@ -0,0 +1 @@ +"""Security primitives and authentication services.""" diff --git a/backend/src/backup_tool/security/auth.py b/backend/src/backup_tool/security/auth.py new file mode 100644 index 0000000..1e2f245 --- /dev/null +++ b/backend/src/backup_tool/security/auth.py @@ -0,0 +1,72 @@ +from __future__ import annotations + +import base64 +import hashlib +import hmac +import json +import secrets +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +from argon2 import PasswordHasher +from argon2.exceptions import VerifyMismatchError + +_PASSWORDS = PasswordHasher() + + +def hash_password(password: str) -> str: + return _PASSWORDS.hash(password) + + +def verify_password(password_hash: str, password: str) -> bool: + try: + return _PASSWORDS.verify(password_hash, password) + except VerifyMismatchError: + return False + + +def hash_token(token: str) -> str: + return hashlib.sha256(token.encode()).hexdigest() + + +def new_token() -> str: + return secrets.token_urlsafe(32) + + +def new_csrf_token() -> str: + return secrets.token_urlsafe(32) + + +def _session_key(master_key_file: Path) -> bytes: + return hashlib.sha256(b"backup-tool-session\0" + master_key_file.read_bytes()).digest() + + +def sign_session(user_id: str, csrf: str, master_key_file: Path) -> str: + payload = json.dumps( + {"sub": user_id, "csrf": csrf, "iat": datetime.now(UTC).isoformat()}, + separators=(",", ":"), + sort_keys=True, + ).encode() + encoded = base64.urlsafe_b64encode(payload).decode().rstrip("=") + signature = hmac.new( + _session_key(master_key_file), encoded.encode(), hashlib.sha256 + ).hexdigest() + return f"{encoded}.{signature}" + + +def verify_session(value: str, master_key_file: Path) -> dict[str, Any] | None: + try: + encoded, supplied = value.rsplit(".", 1) + expected = hmac.new( + _session_key(master_key_file), encoded.encode(), hashlib.sha256 + ).hexdigest() + if not hmac.compare_digest(supplied, expected): + return None + padded = encoded + "=" * (-len(encoded) % 4) + decoded = json.loads(base64.urlsafe_b64decode(padded)) + if not isinstance(decoded, dict) or not isinstance(decoded.get("sub"), str): + return None + return decoded + except (ValueError, json.JSONDecodeError, UnicodeDecodeError): + return None diff --git a/backend/src/backup_tool/security/redaction.py b/backend/src/backup_tool/security/redaction.py new file mode 100644 index 0000000..9c213f7 --- /dev/null +++ b/backend/src/backup_tool/security/redaction.py @@ -0,0 +1,41 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from typing import Any + +SECRET_KEYS = frozenset( + { + "password", + "password_hash", + "secret", + "token", + "token_hash", + "authorization", + "cookie", + "ciphertext", + "master_key", + "private_key", + } +) + + +def redact(value: Any, *, canaries: Sequence[str] = ()) -> Any: + """Return a recursively redacted copy safe for operator-visible sinks.""" + if isinstance(value, Mapping): + return { + str(key): ( + "[REDACTED]" if str(key).lower() in SECRET_KEYS else redact(item, canaries=canaries) + ) + for key, item in value.items() + } + if isinstance(value, list): + return [redact(item, canaries=canaries) for item in value] + if isinstance(value, tuple): + return tuple(redact(item, canaries=canaries) for item in value) + if isinstance(value, str): + result = value + for canary in canaries: + if canary: + result = result.replace(canary, "[REDACTED]") + return result + return value diff --git a/backend/src/backup_tool/security/secrets.py b/backend/src/backup_tool/security/secrets.py new file mode 100644 index 0000000..0e1ff33 --- /dev/null +++ b/backend/src/backup_tool/security/secrets.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +import hashlib +import os +from pathlib import Path + +from cryptography.hazmat.primitives.ciphers.aead import AESGCM + + +class EnvelopeCipher: + """Purpose-bound authenticated encryption for persisted secret values.""" + + def __init__(self, master_key: bytes): + self._key = hashlib.sha256(master_key).digest() + self.key_id = hashlib.sha256(b"backup-tool-key-id\0" + self._key).hexdigest()[:24] + self._cipher = AESGCM(self._key) + + @classmethod + def from_file(cls, path: Path) -> EnvelopeCipher: + return cls(path.read_bytes()) + + @staticmethod + def _associated_data(purpose: str, version: int) -> bytes: + return f"backup-tool-secret:{purpose}:v{version}".encode() + + def encrypt(self, value: str, *, purpose: str, version: int) -> tuple[bytes, str]: + nonce = os.urandom(12) + encrypted = self._cipher.encrypt( + nonce, value.encode(), self._associated_data(purpose, version) + ) + return nonce + encrypted, self.key_id + + def decrypt(self, ciphertext: bytes, *, purpose: str, version: int) -> str: + nonce, encrypted = ciphertext[:12], ciphertext[12:] + plaintext = self._cipher.decrypt(nonce, encrypted, self._associated_data(purpose, version)) + return plaintext.decode() diff --git a/docs/release/m2-evidence.md b/docs/release/m2-evidence.md index 045e2fe..d6ab860 100644 --- a/docs/release/m2-evidence.md +++ b/docs/release/m2-evidence.md @@ -19,6 +19,21 @@ Observed before implementation on 2026-07-27: CSRF, token, secret, audit, pagination, ETag, idempotency, readiness, and leakage behavior through public interfaces. -## GREEN +## GREEN — secure control plane -Pending implementation. +```bash +make check +.venv/bin/python -m pytest tests/unit/test_redaction.py \ + tests/contract/test_api_conventions.py tests/integration/test_auth.py \ + tests/security/test_auth.py tests/security/test_leakage_scan.py -q +``` + +Observed on 2026-07-27: + +- Focused M2 behavior suite: `17 passed`. +- Complete local check: `39` unit/contract, `12` integration, and `3` security tests + passed; Ruff, mypy, forbidden-v1 scan, TypeScript typecheck, and frontend build passed. +- Security tests prove Argon2id hashes, encrypted secret persistence, CSRF, + scoped/revoked/expired tokens, request-digest idempotency, problem details, + request IDs, ETags, cursor pagination, setup race handling, readiness states, + and output canary scanning. diff --git a/tools/forbidden_v1_scan.py b/tools/forbidden_v1_scan.py index 5247f27..1c6a4d3 100644 --- a/tools/forbidden_v1_scan.py +++ b/tools/forbidden_v1_scan.py @@ -42,7 +42,7 @@ FORBIDDEN = ( r"\b(?:read|open|load|import|convert)[_-]?(?:legacy|v1)[_-]?(?:database|db|payload|backup)\b", re.IGNORECASE, ), - re.compile(r"\bbackup_tool\.(?:db|sqlite3?)\b", re.IGNORECASE), + re.compile(r"[\"']backup_tool\.db[\"']", re.IGNORECASE), re.compile(r"%Y-%m-%d_%H%M%S"), re.compile(r"\btimestamp[_-]?(?:directory|parser)\b", re.IGNORECASE), re.compile(r"\b(?:app\.main|backup\.engine)\b"), diff --git a/tools/leakage_scan.py b/tools/leakage_scan.py new file mode 100644 index 0000000..807ec62 --- /dev/null +++ b/tools/leakage_scan.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +import argparse +from pathlib import Path + + +def main() -> int: + parser = argparse.ArgumentParser( + description="Fail when a secret canary appears in output files." + ) + parser.add_argument("--canary", action="append", default=[]) + parser.add_argument("paths", nargs="+") + args = parser.parse_args() + findings: list[str] = [] + for raw_path in args.paths: + path = Path(raw_path) + if not path.is_file(): + continue + text = path.read_text(encoding="utf-8", errors="replace") + if any(canary and canary in text for canary in args.canary): + findings.append(str(path)) + if findings: + print("secret canary found: " + ", ".join(findings)) + return 1 + print("leakage scan: OK") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())