134 lines
5.0 KiB
Python
134 lines
5.0 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
import pytest
|
|
|
|
PASSWORD = "correct horse battery staple"
|
|
|
|
|
|
async def setup_admin(client, username: str = "admin"):
|
|
return await client.post("/api/v2/setup", json={"username": username, "password": PASSWORD})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_setup_race_creates_exactly_one_admin(app_client) -> None:
|
|
client, _ = app_client
|
|
responses = await asyncio.gather(setup_admin(client, "admin-a"), setup_admin(client, "admin-b"))
|
|
assert sorted(response.status_code for response in responses) == [201, 409]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_setup_sets_secure_session_and_csrf_cookies(app_client) -> None:
|
|
client, _ = app_client
|
|
response = await setup_admin(client)
|
|
assert response.status_code == 201
|
|
cookies = response.headers.get_list("set-cookie")
|
|
session = next(value for value in cookies if value.startswith("backup_tool_session="))
|
|
csrf = next(value for value in cookies if value.startswith("backup_tool_csrf="))
|
|
assert "HttpOnly" in session and "Secure" in session and "SameSite=strict" in session
|
|
assert "HttpOnly" not in csrf and "Secure" in csrf and "SameSite=strict" in csrf
|
|
assert "password" not in response.text.lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_csrf_is_required_for_cookie_authenticated_mutation(app_client) -> None:
|
|
client, _ = app_client
|
|
assert (await setup_admin(client)).status_code == 201
|
|
denied = await client.post(
|
|
"/api/v2/admin/secrets", json={"purpose": "ssh", "value": "top-secret"}
|
|
)
|
|
assert denied.status_code == 403
|
|
assert denied.json()["code"] == "csrf_failed"
|
|
|
|
allowed = await client.post(
|
|
"/api/v2/admin/secrets",
|
|
json={"purpose": "ssh", "value": "top-secret"},
|
|
headers={"X-CSRF-Token": client.cookies["backup_tool_csrf"]},
|
|
)
|
|
assert allowed.status_code == 201
|
|
assert "top-secret" not in allowed.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_rejects_wrong_password_and_restores_session(app_client) -> None:
|
|
client, _ = app_client
|
|
assert (await setup_admin(client)).status_code == 201
|
|
await client.post(
|
|
"/api/v2/auth/logout", headers={"X-CSRF-Token": client.cookies["backup_tool_csrf"]}
|
|
)
|
|
wrong = await client.post(
|
|
"/api/v2/auth/login", json={"username": "admin", "password": "wrong-password"}
|
|
)
|
|
assert wrong.status_code == 401
|
|
logged_in = await client.post(
|
|
"/api/v2/auth/login", json={"username": "admin", "password": PASSWORD}
|
|
)
|
|
assert logged_in.status_code == 200
|
|
session = await client.get("/api/v2/auth/session")
|
|
assert session.status_code == 200
|
|
assert session.json()["username"] == "admin"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_api_token_scope_revocation_and_expiry(app_client) -> None:
|
|
client, _ = app_client
|
|
assert (await setup_admin(client)).status_code == 201
|
|
csrf = client.cookies["backup_tool_csrf"]
|
|
expiry = (datetime.now(UTC) + timedelta(hours=1)).isoformat()
|
|
created = await client.post(
|
|
"/api/v2/auth/tokens",
|
|
json={"scopes": ["audit:read"], "expires_at": expiry},
|
|
headers={"X-CSRF-Token": csrf, "Idempotency-Key": "audit-token"},
|
|
)
|
|
assert created.status_code == 201
|
|
token = created.json()["token"]
|
|
client.cookies.clear()
|
|
allowed = await client.get("/api/v2/audit", headers={"Authorization": f"Bearer {token}"})
|
|
assert allowed.status_code == 200
|
|
denied = await client.post(
|
|
"/api/v2/admin/secrets",
|
|
json={"purpose": "ssh", "value": "x"},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert denied.status_code == 403
|
|
assert denied.json()["code"] == "insufficient_scope"
|
|
|
|
# Log back in to revoke it.
|
|
assert (
|
|
await client.post("/api/v2/auth/login", json={"username": "admin", "password": PASSWORD})
|
|
).status_code == 200
|
|
csrf = client.cookies["backup_tool_csrf"]
|
|
assert (
|
|
await client.delete(
|
|
f"/api/v2/auth/tokens/{created.json()['id']}",
|
|
headers={"X-CSRF-Token": csrf},
|
|
)
|
|
).status_code == 204
|
|
client.cookies.clear()
|
|
revoked = await client.get("/api/v2/audit", headers={"Authorization": f"Bearer {token}"})
|
|
assert revoked.status_code == 401
|
|
|
|
# Expired token is accepted at creation but never at authentication.
|
|
assert (
|
|
await client.post("/api/v2/auth/login", json={"username": "admin", "password": PASSWORD})
|
|
).status_code == 200
|
|
expired = await client.post(
|
|
"/api/v2/auth/tokens",
|
|
json={
|
|
"scopes": ["audit:read"],
|
|
"expires_at": (datetime.now(UTC) - timedelta(seconds=1)).isoformat(),
|
|
},
|
|
headers={
|
|
"X-CSRF-Token": client.cookies["backup_tool_csrf"],
|
|
"Idempotency-Key": "expired-token",
|
|
},
|
|
)
|
|
assert expired.status_code == 201
|
|
expired_token = expired.json()["token"]
|
|
client.cookies.clear()
|
|
assert (
|
|
await client.get("/api/v2/audit", headers={"Authorization": f"Bearer {expired_token}"})
|
|
).status_code == 401
|