187 lines
7.1 KiB
Python
187 lines
7.1 KiB
Python
import httpx
|
|
import pytest
|
|
|
|
from src import main
|
|
|
|
|
|
@pytest.fixture
|
|
def access_control(monkeypatch):
|
|
monkeypatch.setenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "correct horse battery staple")
|
|
monkeypatch.setenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", "a-separate-session-signing-secret-with-enough-entropy")
|
|
|
|
|
|
@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_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_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_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_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_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")
|
|
)
|