1322 lines
50 KiB
Python
1322 lines
50 KiB
Python
import asyncio
|
|
import sqlite3
|
|
import time
|
|
from urllib.parse import parse_qs, urlsplit
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src import main
|
|
from src.session_store import SessionStoreError
|
|
from src.views import FRONTEND_BUILD
|
|
|
|
|
|
@pytest.fixture
|
|
def access_control(monkeypatch, tmp_path):
|
|
monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator")
|
|
monkeypatch.setenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "correct horse battery staple")
|
|
monkeypatch.setenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", "a-separate-session-signing-secret-with-enough-entropy")
|
|
monkeypatch.setenv("STACKCHAIN_SESSION_DB", str(tmp_path / "sessions.sqlite3"))
|
|
monkeypatch.setenv("STACKCHAIN_SECURITY_EVENT_DB", str(tmp_path / "security-events.sqlite3"))
|
|
monkeypatch.setenv("STACKCHAIN_LOGIN_ATTEMPT_DB", str(tmp_path / "login-attempts.sqlite3"))
|
|
monkeypatch.setenv("STACKCHAIN_LOGIN_MAX_FAILURES", "3")
|
|
monkeypatch.setenv("STACKCHAIN_LOGIN_WINDOW_SECONDS", "60")
|
|
|
|
|
|
async def fresh_grant(client, action: str, target: str) -> str:
|
|
response = await client.post(
|
|
"/api/v1/fresh-authorization",
|
|
json={
|
|
"access_token": "correct horse battery staple",
|
|
"action": action,
|
|
"target": target,
|
|
},
|
|
headers={
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": client.cookies["stackchain_csrf"],
|
|
},
|
|
)
|
|
assert response.status_code == 201
|
|
return response.json()["grant"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_passkey_enrollment_options_require_fresh_authorization_and_are_one_time(
|
|
access_control,
|
|
):
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
await client.post(
|
|
"/api/v1/session",
|
|
json={
|
|
"access_token": "correct horse battery staple",
|
|
"device_label": "Timmy's phone",
|
|
},
|
|
)
|
|
headers = {
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": client.cookies["stackchain_csrf"],
|
|
}
|
|
missing = await client.post("/api/v1/passkeys/registration/options", headers=headers)
|
|
grant = await fresh_grant(client, "enroll_passkey", "current_device")
|
|
created = await client.post(
|
|
"/api/v1/passkeys/registration/options",
|
|
headers={**headers, "X-Step-Up-Grant": grant},
|
|
)
|
|
replayed = await client.post(
|
|
"/api/v1/passkeys/registration/options",
|
|
headers={**headers, "X-Step-Up-Grant": grant},
|
|
)
|
|
|
|
assert missing.status_code == 428
|
|
assert missing.json()["detail"]["action"] == "enroll_passkey"
|
|
assert created.status_code == 201
|
|
assert created.headers["cache-control"] == "no-store"
|
|
options = created.json()
|
|
assert options["rp"] == {"id": "test", "name": "Stackchain Dashboard"}
|
|
assert options["user"]["name"] == "stackchain-operator"
|
|
assert options["authenticatorSelection"]["userVerification"] == "required"
|
|
assert options["challenge"]
|
|
assert replayed.status_code == 428
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_enrolled_passkey_can_sign_in_without_the_operator_token(
|
|
access_control, monkeypatch
|
|
):
|
|
class VerifiedRegistration:
|
|
credential_id = b"phone-credential"
|
|
credential_public_key = b"credential-public-key"
|
|
sign_count = 0
|
|
|
|
class VerifiedAuthentication:
|
|
new_sign_count = 1
|
|
|
|
monkeypatch.setattr(
|
|
main.passkeys,
|
|
"verify_registration",
|
|
lambda **_kwargs: VerifiedRegistration(),
|
|
raising=False,
|
|
)
|
|
monkeypatch.setattr(
|
|
main.passkeys,
|
|
"verify_authentication",
|
|
lambda **_kwargs: VerifiedAuthentication(),
|
|
raising=False,
|
|
)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as bootstrap:
|
|
await bootstrap.post(
|
|
"/api/v1/session",
|
|
json={"access_token": "correct horse battery staple", "device_label": "Phone"},
|
|
)
|
|
csrf_headers = {
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": bootstrap.cookies["stackchain_csrf"],
|
|
}
|
|
grant = await fresh_grant(bootstrap, "enroll_passkey", "current_device")
|
|
options = await bootstrap.post(
|
|
"/api/v1/passkeys/registration/options",
|
|
headers={**csrf_headers, "X-Step-Up-Grant": grant},
|
|
)
|
|
enrolled = await bootstrap.post(
|
|
"/api/v1/passkeys/registration/verify",
|
|
json={"challenge": options.json()["challenge"], "credential": {"id": "fake"}},
|
|
headers=csrf_headers,
|
|
)
|
|
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as returning:
|
|
sign_in_options = await returning.post("/api/v1/passkeys/authentication/options")
|
|
signed_in = await returning.post(
|
|
"/api/v1/passkeys/authentication/verify",
|
|
json={
|
|
"challenge": sign_in_options.json()["challenge"],
|
|
"credential": {"id": "cGhvbmUtY3JlZGVudGlhbA"},
|
|
"device_label": "Phone",
|
|
"action": "sign_in",
|
|
"target": "dashboard",
|
|
},
|
|
)
|
|
activity = await returning.get("/api/v1/security-events")
|
|
|
|
assert enrolled.status_code == 201
|
|
assert enrolled.json() == {"enrolled": True}
|
|
assert sign_in_options.status_code == 200
|
|
assert sign_in_options.json()["allowCredentials"][0]["id"] == "cGhvbmUtY3JlZGVudGlhbA"
|
|
assert signed_in.status_code == 200
|
|
assert signed_in.json() == {"authenticated": True, "method": "passkey"}
|
|
assert activity.json()["events"][0] == {
|
|
"id": activity.json()["events"][0]["id"],
|
|
"kind": "sign_in",
|
|
"method": "passkey",
|
|
"device_label": "Phone",
|
|
"target": "dashboard",
|
|
"created_at": activity.json()["events"][0]["created_at"],
|
|
"status": "completed",
|
|
}
|
|
assert "stackchain_session=" in signed_in.headers["set-cookie"]
|
|
assert "correct horse battery staple" not in signed_in.text
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_passkey_fresh_authorization_is_exact_target_bound_and_single_use(
|
|
access_control, monkeypatch
|
|
):
|
|
class VerifiedRegistration:
|
|
credential_id = b"phone-credential"
|
|
credential_public_key = b"credential-public-key"
|
|
sign_count = 0
|
|
|
|
class VerifiedAuthentication:
|
|
new_sign_count = 1
|
|
|
|
monkeypatch.setattr(
|
|
main.passkeys, "verify_registration", lambda **_kwargs: VerifiedRegistration()
|
|
)
|
|
monkeypatch.setattr(
|
|
main.passkeys, "verify_authentication", lambda **_kwargs: VerifiedAuthentication()
|
|
)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
await client.post(
|
|
"/api/v1/session",
|
|
json={"access_token": "correct horse battery staple", "device_label": "Phone"},
|
|
)
|
|
headers = {
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": client.cookies["stackchain_csrf"],
|
|
}
|
|
enrollment_grant = await fresh_grant(client, "enroll_passkey", "current_device")
|
|
registration = await client.post(
|
|
"/api/v1/passkeys/registration/options",
|
|
headers={**headers, "X-Step-Up-Grant": enrollment_grant},
|
|
)
|
|
await client.post(
|
|
"/api/v1/passkeys/registration/verify",
|
|
json={"challenge": registration.json()["challenge"], "credential": {"id": "fake"}},
|
|
headers=headers,
|
|
)
|
|
options = await client.post(
|
|
"/api/v1/passkeys/authorization/options",
|
|
json={"action": "close_issue", "target": "stackchain/api#7"},
|
|
headers=headers,
|
|
)
|
|
ceremony = {
|
|
"challenge": options.json()["challenge"],
|
|
"credential": {"id": "cGhvbmUtY3JlZGVudGlhbA"},
|
|
"action": "close_issue",
|
|
"target": "stackchain/api#7",
|
|
}
|
|
authorized = await client.post(
|
|
"/api/v1/passkeys/authorization/verify", json=ceremony, headers=headers
|
|
)
|
|
replayed = await client.post(
|
|
"/api/v1/passkeys/authorization/verify", json=ceremony, headers=headers
|
|
)
|
|
wrong_target = await client.post(
|
|
"/api/v1/passkeys/authorization/verify",
|
|
json={**ceremony, "target": "stackchain/api#8"},
|
|
headers=headers,
|
|
)
|
|
|
|
assert options.status_code == 200
|
|
assert authorized.status_code == 201
|
|
assert authorized.json()["grant"]
|
|
assert authorized.json()["expires_in"] == 90
|
|
assert replayed.status_code == 409
|
|
assert wrong_target.status_code == 409
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_revoking_an_enrolled_device_also_revokes_its_passkey(
|
|
access_control, monkeypatch
|
|
):
|
|
class VerifiedRegistration:
|
|
credential_id = b"phone-credential"
|
|
credential_public_key = b"credential-public-key"
|
|
sign_count = 0
|
|
|
|
monkeypatch.setattr(
|
|
main.passkeys, "verify_registration", lambda **_kwargs: VerifiedRegistration()
|
|
)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with (
|
|
httpx.AsyncClient(transport=transport, base_url="https://test") as phone,
|
|
httpx.AsyncClient(transport=transport, base_url="https://test") as laptop,
|
|
):
|
|
await phone.post(
|
|
"/api/v1/session",
|
|
json={"access_token": "correct horse battery staple", "device_label": "Phone"},
|
|
)
|
|
phone_headers = {
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": phone.cookies["stackchain_csrf"],
|
|
}
|
|
enrollment_grant = await fresh_grant(phone, "enroll_passkey", "current_device")
|
|
registration = await phone.post(
|
|
"/api/v1/passkeys/registration/options",
|
|
headers={**phone_headers, "X-Step-Up-Grant": enrollment_grant},
|
|
)
|
|
await phone.post(
|
|
"/api/v1/passkeys/registration/verify",
|
|
json={"challenge": registration.json()["challenge"], "credential": {"id": "fake"}},
|
|
headers=phone_headers,
|
|
)
|
|
await laptop.post(
|
|
"/api/v1/session",
|
|
json={"access_token": "correct horse battery staple", "device_label": "Laptop"},
|
|
)
|
|
devices = (await laptop.get("/api/v1/sessions")).json()["devices"]
|
|
phone_device = next(item for item in devices if item["device_label"] == "Phone")
|
|
grant = await fresh_grant(laptop, "revoke_device", phone_device["management_id"])
|
|
revoked = await laptop.delete(
|
|
f"/api/v1/sessions/{phone_device['management_id']}",
|
|
headers={
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": laptop.cookies["stackchain_csrf"],
|
|
"X-Step-Up-Grant": grant,
|
|
},
|
|
)
|
|
passkey_options = await laptop.post("/api/v1/passkeys/authentication/options")
|
|
|
|
assert revoked.status_code == 200
|
|
assert passkey_options.status_code == 404
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_default_operator_mode_fails_closed_before_gitea_when_secrets_are_missing(monkeypatch):
|
|
monkeypatch.delenv("STACKCHAIN_DASHBOARD_AUTH_MODE", raising=False)
|
|
called = False
|
|
|
|
async def user():
|
|
nonlocal called
|
|
called = True
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
health = await client.get("/healthz")
|
|
responses = [
|
|
await client.get("/readyz"),
|
|
await client.get("/login"),
|
|
await client.get("/"),
|
|
await client.get("/api/v1/context"),
|
|
]
|
|
|
|
assert health.status_code == 200
|
|
assert [response.status_code for response in responses] == [503, 503, 503, 503]
|
|
assert all(response.headers["cache-control"] == "no-store" for response in responses)
|
|
assert all(response.json() == {
|
|
"detail": "Dashboard authentication is not configured"
|
|
} for response in responses)
|
|
assert called is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_operator_mode_rejects_reused_or_incomplete_secrets(monkeypatch):
|
|
monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator")
|
|
monkeypatch.setenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "same-secret-with-at-least-thirty-two-characters")
|
|
monkeypatch.setenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", "same-secret-with-at-least-thirty-two-characters")
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
reused = await client.get("/login")
|
|
monkeypatch.setenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", "")
|
|
incomplete = await client.get("/login")
|
|
|
|
assert reused.status_code == 503
|
|
assert incomplete.status_code == 503
|
|
assert "same-secret" not in reused.text
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_insecure_local_mode_is_restricted_to_loopback(monkeypatch):
|
|
monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "insecure-local")
|
|
local_transport = httpx.ASGITransport(app=main.app, client=("127.0.0.1", 1234))
|
|
remote_transport = httpx.ASGITransport(app=main.app, client=("203.0.113.9", 1234))
|
|
async with httpx.AsyncClient(transport=local_transport, base_url="http://test") as client:
|
|
local = await client.get("/login")
|
|
async with httpx.AsyncClient(transport=remote_transport, base_url="http://test") as client:
|
|
remote = await client.get("/login")
|
|
|
|
assert local.status_code == 200
|
|
assert remote.status_code == 403
|
|
assert remote.json() == {"detail": "Insecure local mode requires a loopback client"}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_anonymous_private_request_is_rejected_before_gitea(access_control, monkeypatch):
|
|
called = False
|
|
|
|
async def user():
|
|
nonlocal called
|
|
called = True
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
response = await client.get("/api/v1/context")
|
|
|
|
assert response.status_code == 401
|
|
assert response.json() == {"detail": "Authentication required"}
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert called is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_anonymous_share_target_redirect_preserves_only_bounded_capture_fields(access_control):
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
response = await client.get(
|
|
"/",
|
|
params={
|
|
"title": "Production crash",
|
|
"text": "Steps from the mobile app",
|
|
"url": "https://example.com/incidents/42",
|
|
"next": "https://evil.example/steal",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 303
|
|
login_query = parse_qs(urlsplit(response.headers["location"]).query)
|
|
assert login_query == {
|
|
"continue": [
|
|
"./?title=Production+crash&text=Steps+from+the+mobile+app&url="
|
|
"https%3A%2F%2Fexample.com%2Fincidents%2F42"
|
|
]
|
|
}
|
|
assert response.headers["cache-control"] == "no-store"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_oversized_share_target_is_not_carried_through_login(access_control):
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
response = await client.get(
|
|
"/", params={"title": "x" * 201, "text": "keep me"}
|
|
)
|
|
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "login"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_sign_in_creates_secure_session_without_echoing_access_token(access_control):
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
response = await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {"authenticated": True}
|
|
cookies = response.headers.get_list("set-cookie")
|
|
assert any("stackchain_session=" in value and "HttpOnly" in value and "Secure" in value and "SameSite=strict" in value for value in cookies)
|
|
assert any("stackchain_csrf=" in value and "Secure" in value and "SameSite=strict" in value and "HttpOnly" not in value for value in cookies)
|
|
assert "correct horse battery staple" not in response.text
|
|
assert response.headers["cache-control"] == "no-store"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_sign_in_throttles_repeated_failures_with_retry_guidance(access_control):
|
|
transport = httpx.ASGITransport(app=main.app, client=("203.0.113.7", 1234))
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
failures = [
|
|
await client.post("/api/v1/session", json={"access_token": "wrong"})
|
|
for _ in range(3)
|
|
]
|
|
blocked = await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
|
|
assert [response.status_code for response in failures] == [401, 401, 401]
|
|
assert blocked.status_code == 429
|
|
assert blocked.json() == {"detail": "Too many sign-in attempts"}
|
|
assert blocked.headers["retry-after"].isdigit()
|
|
assert blocked.headers["cache-control"] == "no-store"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_successful_sign_in_clears_prior_failures(access_control):
|
|
transport = httpx.ASGITransport(app=main.app, client=("203.0.113.8", 1234))
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
for _ in range(2):
|
|
await client.post("/api/v1/session", json={"access_token": "wrong"})
|
|
success = await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
after_success = [
|
|
await client.post("/api/v1/session", json={"access_token": "wrong"})
|
|
for _ in range(3)
|
|
]
|
|
|
|
assert success.status_code == 200
|
|
assert [response.status_code for response in after_success] == [401, 401, 401]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_authenticated_get_reaches_private_api(access_control, monkeypatch):
|
|
async def user():
|
|
return {"id": 1, "login": "timmy", "full_name": "", "email": ""}
|
|
|
|
async def empty():
|
|
return []
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "repos", empty)
|
|
monkeypatch.setattr(main, "issues", empty)
|
|
monkeypatch.setattr(main, "pull_requests", empty)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
signed_in = await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
response = await client.get("/api/v1/context")
|
|
|
|
assert signed_in.status_code == 200
|
|
assert response.status_code == 200
|
|
assert response.json()["user"]["login"] == "timmy"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_session_status_reuses_the_middleware_validation(access_control, monkeypatch):
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
original_store = main.dashboard_auth._session_store()
|
|
lookups = 0
|
|
|
|
class CountingStore:
|
|
def status(self, session_id, expires_at, *, idle_timeout_seconds):
|
|
nonlocal lookups
|
|
lookups += 1
|
|
return original_store.status(
|
|
session_id,
|
|
expires_at,
|
|
idle_timeout_seconds=idle_timeout_seconds,
|
|
)
|
|
|
|
monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: CountingStore())
|
|
response = await client.get("/api/v1/session")
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["authenticated"] is True
|
|
assert isinstance(payload["expires_at"], int)
|
|
assert payload["expires_at"] > int(time.time())
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert lookups == 1
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_authenticated_session_status_exposes_csrf_proof_and_offline_lease(access_control):
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
response = await client.get("/api/v1/session")
|
|
|
|
payload = response.json()
|
|
assert response.status_code == 200
|
|
assert payload["authenticated"] is True
|
|
assert payload["csrf_token"] == client.cookies["stackchain_csrf"]
|
|
assert payload["expires_at"] > int(time.time())
|
|
assert "correct horse battery staple" not in response.text
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_anonymous_session_status_discloses_no_offline_lease(access_control):
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
response = await client.get("/api/v1/session")
|
|
|
|
assert response.status_code == 401
|
|
assert response.json() == {"detail": "Authentication required"}
|
|
assert "expires_at" not in response.text
|
|
assert response.headers["cache-control"] == "no-store"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_idle_session_is_rejected_with_distinct_api_and_page_recovery(access_control, monkeypatch):
|
|
monkeypatch.setenv("STACKCHAIN_DASHBOARD_IDLE_TIMEOUT_SECONDS", "900")
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
with sqlite3.connect(main.dashboard_auth._session_store().path) as connection:
|
|
connection.execute(
|
|
"UPDATE active_sessions SET last_active_at = ?", (int(time.time()) - 900,)
|
|
)
|
|
|
|
api_response = await client.get("/api/v1/session")
|
|
page_response = await client.get("/")
|
|
|
|
assert api_response.status_code == 401
|
|
assert api_response.json() == {
|
|
"detail": "Authentication required",
|
|
"code": "session_idle",
|
|
}
|
|
assert page_response.status_code == 303
|
|
assert page_response.headers["location"] == "login?reason=session-idle"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_activity_heartbeat_extends_the_server_idle_deadline(access_control, monkeypatch):
|
|
monkeypatch.setenv("STACKCHAIN_DASHBOARD_IDLE_TIMEOUT_SECONDS", "900")
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
heartbeat = await client.post(
|
|
"/api/v1/session/activity",
|
|
headers={
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": client.cookies["stackchain_csrf"],
|
|
},
|
|
)
|
|
|
|
assert heartbeat.status_code == 200
|
|
assert heartbeat.json()["active"] is True
|
|
assert heartbeat.json()["idle_expires_at"] >= int(time.time()) + 899
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_merge_requires_single_use_fresh_authorization_bound_to_exact_target(
|
|
access_control, monkeypatch
|
|
):
|
|
merge_calls = []
|
|
|
|
async def assigned(repository, number):
|
|
return True
|
|
|
|
async def merge(repository, number, expected_head_sha):
|
|
merge_calls.append((repository, number, expected_head_sha))
|
|
return {"number": number, "merged": True, "state": "closed"}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned)
|
|
monkeypatch.setattr(main.gitea_proxy, "merge_assigned_pull", merge)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
csrf_headers = {
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": client.cookies["stackchain_csrf"],
|
|
}
|
|
missing = await client.post(
|
|
"/api/v1/repos/stackchain/api/pulls/7/merge",
|
|
json={"expected_head_sha": "abc123"},
|
|
headers=csrf_headers,
|
|
)
|
|
authorized = await client.post(
|
|
"/api/v1/fresh-authorization",
|
|
json={
|
|
"access_token": "correct horse battery staple",
|
|
"action": "merge_pull",
|
|
"target": "stackchain/api#7",
|
|
},
|
|
headers=csrf_headers,
|
|
)
|
|
grant_headers = {**csrf_headers, "X-Step-Up-Grant": authorized.json()["grant"]}
|
|
merged = await client.post(
|
|
"/api/v1/repos/stackchain/api/pulls/7/merge",
|
|
json={"expected_head_sha": "abc123"},
|
|
headers=grant_headers,
|
|
)
|
|
replayed = await client.post(
|
|
"/api/v1/repos/stackchain/api/pulls/7/merge",
|
|
json={"expected_head_sha": "abc123"},
|
|
headers=grant_headers,
|
|
)
|
|
|
|
assert missing.status_code == 428
|
|
assert missing.json() == {
|
|
"detail": {
|
|
"detail": "Fresh authorization required",
|
|
"code": "step_up_required",
|
|
"action": "merge_pull",
|
|
"target": "stackchain/api#7",
|
|
}
|
|
}
|
|
assert authorized.status_code == 201
|
|
assert authorized.json()["expires_in"] == 90
|
|
assert merged.status_code == 200
|
|
assert replayed.status_code == 428
|
|
assert merge_calls == [("stackchain/api", 7, "abc123")]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_other_high_impact_routes_require_fresh_authorization_before_mutation(
|
|
access_control, monkeypatch
|
|
):
|
|
close_calls = []
|
|
|
|
async def assigned(repository, number):
|
|
return True
|
|
|
|
async def close(repository, number):
|
|
close_calls.append((repository, number))
|
|
return {"number": number, "state": "closed"}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
|
|
monkeypatch.setattr(main.gitea_proxy, "close_issue", close)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with (
|
|
httpx.AsyncClient(transport=transport, base_url="https://test") as phone,
|
|
httpx.AsyncClient(transport=transport, base_url="https://test") as laptop,
|
|
):
|
|
await phone.post(
|
|
"/api/v1/session",
|
|
json={"access_token": "correct horse battery staple", "device_label": "Phone"},
|
|
)
|
|
await laptop.post(
|
|
"/api/v1/session",
|
|
json={"access_token": "correct horse battery staple", "device_label": "Laptop"},
|
|
)
|
|
remote = next(
|
|
item
|
|
for item in (await phone.get("/api/v1/sessions")).json()["devices"]
|
|
if not item["current"]
|
|
)
|
|
headers = {
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": phone.cookies["stackchain_csrf"],
|
|
}
|
|
closed = await phone.patch(
|
|
"/api/v1/repos/stackchain/api/issues/7/close", headers=headers
|
|
)
|
|
revoked = await phone.delete(
|
|
f"/api/v1/sessions/{remote['management_id']}", headers=headers
|
|
)
|
|
revoked_all = await phone.delete("/api/v1/sessions", headers=headers)
|
|
laptop_still_active = await laptop.get("/api/v1/session")
|
|
|
|
assert [closed.status_code, revoked.status_code, revoked_all.status_code] == [428, 428, 428]
|
|
assert [closed.json()["detail"]["action"], revoked.json()["detail"]["action"], revoked_all.json()["detail"]["action"]] == [
|
|
"close_issue", "revoke_device", "revoke_all_sessions"
|
|
]
|
|
assert close_calls == []
|
|
assert laptop_still_active.status_code == 200
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_operator_can_review_and_revoke_one_remote_device(access_control):
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with (
|
|
httpx.AsyncClient(transport=transport, base_url="https://test") as phone,
|
|
httpx.AsyncClient(transport=transport, base_url="https://test") as laptop,
|
|
):
|
|
await phone.post(
|
|
"/api/v1/session",
|
|
json={
|
|
"access_token": "correct horse battery staple",
|
|
"device_label": "Pixel <script>",
|
|
},
|
|
)
|
|
await laptop.post(
|
|
"/api/v1/session",
|
|
json={
|
|
"access_token": "correct horse battery staple",
|
|
"device_label": "Work laptop",
|
|
},
|
|
)
|
|
|
|
listed = await laptop.get("/api/v1/sessions")
|
|
devices = listed.json()["devices"]
|
|
phone_device = next(device for device in devices if device["device_label"] == "Pixel <script>")
|
|
current_device = next(device for device in devices if device["current"])
|
|
|
|
missing_csrf = await laptop.delete(
|
|
f"/api/v1/sessions/{phone_device['management_id']}"
|
|
)
|
|
grant = await fresh_grant(
|
|
laptop, "revoke_device", phone_device["management_id"]
|
|
)
|
|
revoked = await laptop.delete(
|
|
f"/api/v1/sessions/{phone_device['management_id']}",
|
|
headers={
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": laptop.cookies["stackchain_csrf"],
|
|
"X-Step-Up-Grant": grant,
|
|
},
|
|
)
|
|
phone_status = await phone.get("/api/v1/session")
|
|
phone_page = await phone.get("/", follow_redirects=False)
|
|
laptop_status = await laptop.get("/api/v1/session")
|
|
|
|
assert listed.status_code == 200
|
|
assert listed.headers["cache-control"] == "no-store"
|
|
assert current_device["device_label"] == "Work laptop"
|
|
assert missing_csrf.status_code == 403
|
|
assert revoked.json() == {"revoked": True, "current_session": False}
|
|
assert phone_status.status_code == 401
|
|
assert phone_status.json() == {
|
|
"detail": "Authentication required",
|
|
"code": "session_revoked",
|
|
}
|
|
assert phone_page.status_code == 303
|
|
assert phone_page.headers["location"] == "login?reason=session-revoked"
|
|
assert laptop_status.status_code == 200
|
|
assert "session_hash" not in listed.text
|
|
assert "csrf" not in listed.text
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_active_device_labels_are_bounded_at_sign_in(access_control):
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
response = await client.post(
|
|
"/api/v1/session",
|
|
json={
|
|
"access_token": "correct horse battery staple",
|
|
"device_label": "x" * 65,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 422
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_mutation_requires_same_origin_and_session_csrf(access_control, monkeypatch):
|
|
calls = 0
|
|
|
|
async def mark_read(notification_id):
|
|
nonlocal calls
|
|
calls += 1
|
|
return {"id": notification_id, "read": True}
|
|
|
|
monkeypatch.setattr(main, "mark_notification_read", mark_read)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
missing = await client.patch("/api/v1/notifications/7/read")
|
|
foreign = await client.patch(
|
|
"/api/v1/notifications/7/read",
|
|
headers={
|
|
"Origin": "https://evil.example",
|
|
"X-CSRF-Token": client.cookies["stackchain_csrf"],
|
|
},
|
|
)
|
|
valid = await client.patch(
|
|
"/api/v1/notifications/7/read",
|
|
headers={
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": client.cookies["stackchain_csrf"],
|
|
},
|
|
)
|
|
|
|
assert missing.status_code == 403
|
|
assert foreign.status_code == 403
|
|
assert valid.status_code == 200
|
|
assert calls == 1
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_logout_clears_session_and_blocks_private_routes(access_control):
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
csrf = client.cookies["stackchain_csrf"]
|
|
response = await client.delete(
|
|
"/api/v1/session",
|
|
headers={"Origin": "https://test", "X-CSRF-Token": csrf},
|
|
)
|
|
private = await client.get("/api/v1/background-identity")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {"authenticated": False, "clear_private_device_data": True}
|
|
assert private.status_code == 401
|
|
assert all("Max-Age=0" in value for value in response.headers.get_list("set-cookie"))
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_sign_out_all_devices_revokes_every_existing_session(access_control):
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with (
|
|
httpx.AsyncClient(transport=transport, base_url="https://test") as phone,
|
|
httpx.AsyncClient(transport=transport, base_url="https://test") as laptop,
|
|
):
|
|
await phone.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
await laptop.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
|
|
grant = await fresh_grant(phone, "revoke_all_sessions", "all")
|
|
response = await phone.delete(
|
|
"/api/v1/sessions",
|
|
headers={
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": phone.cookies["stackchain_csrf"],
|
|
"X-Step-Up-Grant": grant,
|
|
},
|
|
)
|
|
phone_private = await phone.get("/api/v1/background-identity")
|
|
laptop_private = await laptop.get("/api/v1/background-identity")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"authenticated": False,
|
|
"all_sessions_revoked": True,
|
|
"clear_private_device_data": True,
|
|
}
|
|
assert phone_private.status_code == 401
|
|
assert laptop_private.status_code == 401
|
|
assert all("Max-Age=0" in value for value in response.headers.get_list("set-cookie"))
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_sign_out_all_devices_rejects_cross_site_requests_without_revoking(access_control):
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
response = await client.delete(
|
|
"/api/v1/sessions",
|
|
headers={
|
|
"Origin": "https://evil.example",
|
|
"X-CSRF-Token": client.cookies["stackchain_csrf"],
|
|
},
|
|
)
|
|
session = await client.get("/api/v1/session")
|
|
|
|
assert response.status_code == 403
|
|
assert response.headers.get_list("set-cookie") == []
|
|
assert session.status_code == 200
|
|
assert session.json()["authenticated"] is True
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_sign_out_all_devices_registry_failure_sets_no_cookies(access_control, monkeypatch):
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
csrf = client.cookies["stackchain_csrf"]
|
|
grant = await fresh_grant(client, "revoke_all_sessions", "all")
|
|
|
|
class BrokenStore:
|
|
def status(self, session_id, expires_at, *, idle_timeout_seconds):
|
|
return "active"
|
|
|
|
def consume_step_up(self, grant, session_id, *, action, target):
|
|
return True
|
|
|
|
def revoke_all(self):
|
|
raise SessionStoreError("database path and secret details")
|
|
|
|
monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: BrokenStore())
|
|
response = await client.delete(
|
|
"/api/v1/sessions",
|
|
headers={
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": csrf,
|
|
"X-Step-Up-Grant": grant,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 503
|
|
assert response.json() == {"detail": "Session registry is temporarily unavailable"}
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.headers.get_list("set-cookie") == []
|
|
assert "database path" not in response.text
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_logout_revokes_a_captured_cookie_before_gitea(access_control, monkeypatch):
|
|
calls = 0
|
|
|
|
async def user():
|
|
nonlocal calls
|
|
calls += 1
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
captured = client.cookies["stackchain_session"]
|
|
csrf = client.cookies["stackchain_csrf"]
|
|
|
|
async with httpx.AsyncClient(
|
|
transport=transport,
|
|
base_url="https://test",
|
|
headers={"Cookie": f"stackchain_session={captured}"},
|
|
) as replay:
|
|
before = await replay.get("/api/v1/background-identity")
|
|
|
|
await client.delete(
|
|
"/api/v1/session",
|
|
headers={"Origin": "https://test", "X-CSRF-Token": csrf},
|
|
)
|
|
|
|
async with httpx.AsyncClient(
|
|
transport=transport,
|
|
base_url="https://test",
|
|
headers={"Cookie": f"stackchain_session={captured}"},
|
|
) as replay:
|
|
after = await replay.get("/api/v1/background-identity")
|
|
|
|
assert before.status_code == 200
|
|
assert after.status_code == 401
|
|
assert after.headers["cache-control"] == "no-store"
|
|
assert calls == 1
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_session_registry_latency_does_not_block_the_event_loop(access_control, monkeypatch):
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
|
|
class SlowStore:
|
|
def status(self, session_id, expires_at, *, idle_timeout_seconds):
|
|
time.sleep(0.15)
|
|
return "active"
|
|
|
|
monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: SlowStore())
|
|
private_request = asyncio.create_task(client.get("/api/v1/session"))
|
|
await asyncio.sleep(0)
|
|
started = time.perf_counter()
|
|
await asyncio.sleep(0.01)
|
|
heartbeat_elapsed = time.perf_counter() - started
|
|
response = await private_request
|
|
|
|
assert response.status_code == 200
|
|
assert heartbeat_elapsed < 0.08
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_sign_in_throttle_lookup_does_not_block_the_event_loop(
|
|
access_control, monkeypatch
|
|
):
|
|
class SlowAttempts:
|
|
def retry_after(self, source):
|
|
time.sleep(0.15)
|
|
return 17
|
|
|
|
monkeypatch.setattr(main, "_login_attempt_store", lambda: SlowAttempts())
|
|
transport = httpx.ASGITransport(app=main.app, client=("203.0.113.10", 1234))
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
sign_in = asyncio.create_task(
|
|
client.post("/api/v1/session", json={"access_token": "wrong"})
|
|
)
|
|
await asyncio.sleep(0)
|
|
started = time.perf_counter()
|
|
await asyncio.sleep(0.01)
|
|
heartbeat_elapsed = time.perf_counter() - started
|
|
response = await sign_in
|
|
|
|
assert response.status_code == 429
|
|
assert response.headers["retry-after"] == "17"
|
|
assert heartbeat_elapsed < 0.08
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_failed_sign_in_recording_does_not_block_the_event_loop(
|
|
access_control, monkeypatch
|
|
):
|
|
class SlowAttempts:
|
|
def retry_after(self, source):
|
|
time.sleep(0.02)
|
|
return 0
|
|
|
|
def record_failure(self, source):
|
|
time.sleep(0.15)
|
|
|
|
monkeypatch.setattr(main, "_login_attempt_store", lambda: SlowAttempts())
|
|
transport = httpx.ASGITransport(app=main.app, client=("203.0.113.11", 1234))
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
sign_in = asyncio.create_task(
|
|
client.post("/api/v1/session", json={"access_token": "wrong"})
|
|
)
|
|
await asyncio.sleep(0)
|
|
started = time.perf_counter()
|
|
await asyncio.sleep(0.04)
|
|
heartbeat_elapsed = time.perf_counter() - started
|
|
response = await sign_in
|
|
|
|
assert response.status_code == 401
|
|
assert heartbeat_elapsed < 0.08
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_successful_sign_in_throttle_clear_does_not_block_the_event_loop(
|
|
access_control, monkeypatch
|
|
):
|
|
class SlowAttempts:
|
|
def retry_after(self, source):
|
|
time.sleep(0.02)
|
|
return 0
|
|
|
|
def clear(self, source):
|
|
time.sleep(0.15)
|
|
|
|
monkeypatch.setattr(main, "_login_attempt_store", lambda: SlowAttempts())
|
|
transport = httpx.ASGITransport(app=main.app, client=("203.0.113.12", 1234))
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
sign_in = asyncio.create_task(
|
|
client.post(
|
|
"/api/v1/session",
|
|
json={"access_token": "correct horse battery staple"},
|
|
)
|
|
)
|
|
await asyncio.sleep(0)
|
|
started = time.perf_counter()
|
|
await asyncio.sleep(0.04)
|
|
heartbeat_elapsed = time.perf_counter() - started
|
|
response = await sign_in
|
|
|
|
assert response.status_code == 200
|
|
assert heartbeat_elapsed < 0.08
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_session_activation_does_not_block_the_event_loop(
|
|
access_control, monkeypatch
|
|
):
|
|
class Attempts:
|
|
def retry_after(self, source):
|
|
time.sleep(0.02)
|
|
return 0
|
|
|
|
def clear(self, source):
|
|
return None
|
|
|
|
class SlowStore:
|
|
def activate(self, session_id, expires_at, **kwargs):
|
|
time.sleep(0.15)
|
|
|
|
monkeypatch.setattr(main, "_login_attempt_store", lambda: Attempts())
|
|
monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: SlowStore())
|
|
transport = httpx.ASGITransport(app=main.app, client=("203.0.113.13", 1234))
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
sign_in = asyncio.create_task(
|
|
client.post(
|
|
"/api/v1/session",
|
|
json={"access_token": "correct horse battery staple"},
|
|
)
|
|
)
|
|
await asyncio.sleep(0)
|
|
started = time.perf_counter()
|
|
await asyncio.sleep(0.04)
|
|
heartbeat_elapsed = time.perf_counter() - started
|
|
response = await sign_in
|
|
|
|
assert response.status_code == 200
|
|
assert any(
|
|
"stackchain_session=" in value
|
|
for value in response.headers.get_list("set-cookie")
|
|
)
|
|
assert heartbeat_elapsed < 0.08
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_single_session_revocation_does_not_block_the_event_loop(
|
|
access_control, monkeypatch
|
|
):
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
await client.post(
|
|
"/api/v1/session",
|
|
json={"access_token": "correct horse battery staple"},
|
|
)
|
|
csrf = client.cookies["stackchain_csrf"]
|
|
|
|
class SlowStore:
|
|
def status(self, session_id, expires_at, *, idle_timeout_seconds):
|
|
time.sleep(0.02)
|
|
return "active"
|
|
|
|
def revoke(self, session_id):
|
|
time.sleep(0.15)
|
|
|
|
monkeypatch.setattr(
|
|
main.dashboard_auth, "_session_store", lambda now=None: SlowStore()
|
|
)
|
|
sign_out = asyncio.create_task(
|
|
client.delete(
|
|
"/api/v1/session",
|
|
headers={"Origin": "https://test", "X-CSRF-Token": csrf},
|
|
)
|
|
)
|
|
await asyncio.sleep(0)
|
|
started = time.perf_counter()
|
|
await asyncio.sleep(0.04)
|
|
heartbeat_elapsed = time.perf_counter() - started
|
|
response = await sign_out
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["authenticated"] is False
|
|
assert heartbeat_elapsed < 0.08
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_session_registry_read_failure_fails_closed_before_gitea(
|
|
access_control, monkeypatch
|
|
):
|
|
called = False
|
|
|
|
async def user():
|
|
nonlocal called
|
|
called = True
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
|
|
class BrokenStore:
|
|
def status(self, session_id, expires_at, *, idle_timeout_seconds):
|
|
raise SessionStoreError("database path and secret details")
|
|
|
|
monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: BrokenStore())
|
|
response = await client.get("/api/v1/background-identity")
|
|
health = await client.get("/healthz")
|
|
|
|
assert response.status_code == 503
|
|
assert response.json() == {"detail": "Session registry is temporarily unavailable"}
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert "database path" not in response.text
|
|
assert called is False
|
|
assert health.status_code == 200
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_sign_in_registry_write_failure_issues_no_cookie(access_control, monkeypatch):
|
|
class BrokenStore:
|
|
def activate(self, session_id, expires_at, **kwargs):
|
|
raise SessionStoreError("database path and secret details")
|
|
|
|
monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: BrokenStore())
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
response = await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
|
|
assert response.status_code == 503
|
|
assert response.json() == {"detail": "Session registry is temporarily unavailable"}
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.headers.get_list("set-cookie") == []
|
|
assert "database path" not in response.text
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_logout_registry_failure_does_not_claim_revocation(access_control, monkeypatch):
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
csrf = client.cookies["stackchain_csrf"]
|
|
|
|
class BrokenStore:
|
|
def status(self, session_id, expires_at, *, idle_timeout_seconds):
|
|
return "active"
|
|
|
|
def revoke(self, session_id):
|
|
raise SessionStoreError("database path and secret details")
|
|
|
|
monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: BrokenStore())
|
|
response = await client.delete(
|
|
"/api/v1/session",
|
|
headers={"Origin": "https://test", "X-CSRF-Token": csrf},
|
|
)
|
|
|
|
assert response.status_code == 503
|
|
assert response.json() == {"detail": "Session registry is temporarily unavailable"}
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.headers.get_list("set-cookie") == []
|
|
assert "database path" not in response.text
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_public_routes_skip_session_registry_validation(access_control, monkeypatch):
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
signed_in = await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
|
|
class BrokenStore:
|
|
def status(self, session_id, expires_at, *, idle_timeout_seconds):
|
|
raise SessionStoreError("public routes must not read the registry")
|
|
|
|
monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: BrokenStore())
|
|
login = await client.get("/login")
|
|
manifest = await client.get("/manifest.webmanifest")
|
|
static = await client.get("/static/session.js")
|
|
runtime = await client.get("/" + FRONTEND_BUILD.runtime_name)
|
|
|
|
assert signed_in.status_code == 200
|
|
assert [login.status_code, manifest.status_code, static.status_code, runtime.status_code] == [
|
|
200,
|
|
200,
|
|
200,
|
|
200,
|
|
]
|
|
assert runtime.headers["cache-control"] == "public, max-age=31536000, immutable"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_public_routes_remain_available_and_readiness_hides_identity(access_control, monkeypatch):
|
|
async def user():
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
health = await client.get("/healthz")
|
|
login = await client.get("/login")
|
|
ready = await client.get("/readyz")
|
|
|
|
assert health.status_code == 200
|
|
assert login.status_code == 200
|
|
assert 'name="access_token"' in login.text
|
|
assert ready.status_code == 200
|
|
assert ready.json() == {"status": "ready", "service": "stackchain-dashboard"}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_subpath_deployment_scopes_routes_and_session_cookies(access_control):
|
|
transport = httpx.ASGITransport(app=main.app, root_path="/dashboard")
|
|
async with httpx.AsyncClient(
|
|
transport=transport, base_url="https://test/dashboard/"
|
|
) as client:
|
|
login = await client.get("login")
|
|
signed_in = await client.post(
|
|
"api/v1/session",
|
|
json={"access_token": "correct horse battery staple"},
|
|
)
|
|
status = await client.get("api/v1/session")
|
|
|
|
assert login.status_code == 200
|
|
assert signed_in.status_code == 200
|
|
assert status.status_code == 200
|
|
assert any(
|
|
"stackchain_session=" in value and "Path=/dashboard" in value
|
|
for value in signed_in.headers.get_list("set-cookie")
|
|
)
|