852 lines
32 KiB
Python
852 lines
32 KiB
Python
import asyncio
|
|
import time
|
|
from urllib.parse import parse_qs, urlsplit
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src import main
|
|
from src.session_store import SessionStoreError
|
|
|
|
|
|
@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_LOGIN_ATTEMPT_DB", str(tmp_path / "login-attempts.sqlite3"))
|
|
monkeypatch.setenv("STACKCHAIN_LOGIN_MAX_FAILURES", "3")
|
|
monkeypatch.setenv("STACKCHAIN_LOGIN_WINDOW_SECONDS", "60")
|
|
|
|
|
|
@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 is_active(self, session_id, expires_at):
|
|
nonlocal lookups
|
|
lookups += 1
|
|
return original_store.is_active(session_id, expires_at)
|
|
|
|
monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: CountingStore())
|
|
response = await client.get("/api/v1/session")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["authenticated"] is True
|
|
assert lookups == 1
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_authenticated_session_status_exposes_only_csrf_proof(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")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"authenticated": True,
|
|
"csrf_token": client.cookies["stackchain_csrf"],
|
|
}
|
|
assert "correct horse battery staple" not in response.text
|
|
|
|
|
|
@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']}"
|
|
)
|
|
revoked = await laptop.delete(
|
|
f"/api/v1/sessions/{phone_device['management_id']}",
|
|
headers={
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": laptop.cookies["stackchain_csrf"],
|
|
},
|
|
)
|
|
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"}
|
|
)
|
|
|
|
response = await phone.delete(
|
|
"/api/v1/sessions",
|
|
headers={
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": phone.cookies["stackchain_csrf"],
|
|
},
|
|
)
|
|
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"]
|
|
|
|
class BrokenStore:
|
|
def is_active(self, session_id, expires_at):
|
|
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},
|
|
)
|
|
|
|
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 is_active(self, session_id, expires_at):
|
|
time.sleep(0.15)
|
|
return True
|
|
|
|
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 is_active(self, session_id, expires_at):
|
|
time.sleep(0.02)
|
|
return True
|
|
|
|
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 is_active(self, session_id, expires_at):
|
|
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 is_active(self, session_id, expires_at):
|
|
return True
|
|
|
|
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 is_active(self, session_id, expires_at):
|
|
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")
|
|
|
|
assert signed_in.status_code == 200
|
|
assert [login.status_code, manifest.status_code, static.status_code] == [200, 200, 200]
|
|
|
|
|
|
@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")
|
|
)
|