backup-tool/tests/security/test_session_security.py

from __future__ import annotations

import importlib
from datetime import UTC, datetime, timedelta

import pytest

PASSWORD = "correct horse battery staple"


@pytest.mark.asyncio
async def test_remote_setup_requires_bootstrap_secret(tmp_path) -> None:
    config = importlib.import_module("backup_tool.config")
    app_module = importlib.import_module("backup_tool.api.app")
    key = tmp_path / "master.key"
    key.write_bytes(b"m2-test-master-key-material-32-bytes-minimum")
    key.chmod(0o600)
    roots = []
    for name in ("data", "repositories", "sources", "restores"):
        path = tmp_path / name
        path.mkdir()
        roots.append(path)
    settings = config.Settings(
        data_dir=roots[0],
        database_url=f"sqlite+aiosqlite:///{roots[0] / 'metadata.db'}",
        repository_roots=(roots[1],),
        local_source_roots=(roots[2],),
        restore_roots=(roots[3],),
        master_key_file=key,
        public_base_url="https://backup.example.test",
        bootstrap_secret="bootstrap-secret-123",
    )
    cli = importlib.import_module("backup_tool.cli")
    from alembic import command
    from httpx import ASGITransport, AsyncClient

    command.upgrade(cli.build_alembic_config(settings), "head")
    app = app_module.create_app(settings)
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="https://backup.example.test"
    ) as client:
        denied = await client.post(
            "/api/v2/setup", json={"username": "admin", "password": PASSWORD}
        )
        assert denied.status_code == 403
        accepted = await client.post(
            "/api/v2/setup",
            json={
                "username": "admin",
                "password": PASSWORD,
                "bootstrap_secret": "bootstrap-secret-123",
            },
        )
        assert accepted.status_code == 201
    await app.state.engine.dispose()


@pytest.mark.asyncio
async def test_remote_setup_without_configured_bootstrap_fails_closed(tmp_path) -> None:
    config = importlib.import_module("backup_tool.config")
    app_module = importlib.import_module("backup_tool.api.app")
    cli = importlib.import_module("backup_tool.cli")
    from alembic import command
    from httpx import ASGITransport, AsyncClient

    key = tmp_path / "master.key"
    key.write_bytes(b"m2-test-master-key-material-32-bytes-minimum")
    key.chmod(0o600)
    roots = [tmp_path / name for name in ("data", "repositories", "sources", "restores")]
    for root in roots:
        root.mkdir()
    settings = config.Settings(
        data_dir=roots[0],
        database_url=f"sqlite+aiosqlite:///{roots[0] / 'metadata.db'}",
        repository_roots=(roots[1],),
        local_source_roots=(roots[2],),
        restore_roots=(roots[3],),
        master_key_file=key,
        public_base_url="https://backup.example.test",
    )
    command.upgrade(cli.build_alembic_config(settings), "head")
    app = app_module.create_app(settings)
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url=settings.public_base_url
    ) as client:
        response = await client.post(
            "/api/v2/setup", json={"username": "admin", "password": PASSWORD}
        )
    assert response.status_code == 403
    assert response.json()["code"] == "bootstrap_required"
    await app.state.engine.dispose()


def test_expired_session_is_rejected(tmp_path) -> None:
    auth = importlib.import_module("backup_tool.security.auth")
    key = tmp_path / "master.key"
    key.write_bytes(b"x" * 32)
    expired = auth.sign_session(
        "user",
        "csrf",
        key,
        expires_at=datetime.now(UTC) - timedelta(seconds=1),
        session_id="session",
    )
    assert auth.verify_session(expired, key) is None


@pytest.mark.asyncio
async def test_logout_revocation_survives_app_restart(tmp_path) -> None:
    config = importlib.import_module("backup_tool.config")
    app_module = importlib.import_module("backup_tool.api.app")
    cli = importlib.import_module("backup_tool.cli")
    from alembic import command
    from httpx import ASGITransport, AsyncClient

    key = tmp_path / "master.key"
    key.write_bytes(b"m2-test-master-key-material-32-bytes-minimum")
    key.chmod(0o600)
    roots = [tmp_path / name for name in ("data", "repositories", "sources", "restores")]
    for root in roots:
        root.mkdir()
    settings = config.Settings(
        data_dir=roots[0],
        database_url=f"sqlite+aiosqlite:///{roots[0] / 'metadata.db'}",
        repository_roots=(roots[1],),
        local_source_roots=(roots[2],),
        restore_roots=(roots[3],),
        master_key_file=key,
        public_base_url="https://127.0.0.1:8000",
    )
    command.upgrade(cli.build_alembic_config(settings), "head")
    app = app_module.create_app(settings)
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url=settings.public_base_url
    ) as client:
        setup = await client.post("/api/v2/setup", json={"username": "admin", "password": PASSWORD})
        assert setup.status_code == 201
        copied_cookie = client.cookies.get("backup_tool_session")
        csrf = client.cookies.get("backup_tool_csrf")
        assert copied_cookie is not None
        assert csrf is not None
        logout = await client.post("/api/v2/auth/logout", headers={"X-CSRF-Token": csrf})
        assert logout.status_code == 204
    await app.state.engine.dispose()

    restarted = app_module.create_app(settings)
    async with AsyncClient(
        transport=ASGITransport(restarted), base_url=settings.public_base_url
    ) as client:
        rejected = await client.get(
            "/api/v2/auth/session", headers={"Cookie": f"backup_tool_session={copied_cookie}"}
        )
    assert rejected.status_code == 401
    await restarted.state.engine.dispose()