test(auth): require durable sessions and safe remote setup
This commit is contained in:
@@ -55,6 +55,38 @@ async def test_remote_setup_requires_bootstrap_secret(tmp_path) -> None:
|
||||
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"
|
||||
@@ -67,3 +99,45 @@ def test_expired_session_is_rejected(tmp_path) -> None:
|
||||
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,
|
||||
)
|
||||
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()
|
||||
|
||||
Reference in New Issue
Block a user