1047 lines
38 KiB
Python
1047 lines
38 KiB
Python
import sqlite3
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src import main
|
|
from src.security_event_store import SecurityEventStore, SecurityEventStoreError
|
|
|
|
|
|
@pytest.fixture
|
|
def security_access(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 / "attempts.sqlite3"))
|
|
monkeypatch.setenv("STACKCHAIN_SECURITY_EVENT_DB", str(tmp_path / "security.sqlite3"))
|
|
|
|
async def upstream_user():
|
|
return {"id": 42, "login": "timmy"}
|
|
|
|
monkeypatch.setattr(main, "current_user", upstream_user)
|
|
return tmp_path
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_authenticated_security_activity_lists_private_sign_in_history(security_access):
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as anonymous:
|
|
rejected = await anonymous.get("/api/v1/security-events")
|
|
|
|
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",
|
|
"device_label": "Timmy's phone",
|
|
},
|
|
)
|
|
activity = await client.get("/api/v1/security-events", params={"limit": 25})
|
|
|
|
assert rejected.status_code == 401
|
|
assert signed_in.status_code == 200
|
|
assert activity.status_code == 200
|
|
assert activity.headers["cache-control"] == "no-store"
|
|
assert activity.json() == {
|
|
"events": [
|
|
{
|
|
"id": 1,
|
|
"kind": "sign_in",
|
|
"method": "token",
|
|
"device_label": "Timmy's phone",
|
|
"target": "dashboard",
|
|
"created_at": activity.json()["events"][0]["created_at"],
|
|
"status": "completed",
|
|
}
|
|
],
|
|
"authentication_alerts": [],
|
|
"next_cursor": None,
|
|
}
|
|
persisted = (security_access / "security.sqlite3").read_bytes()
|
|
assert b"correct horse battery staple" not in persisted
|
|
assert b"stackchain_session" not in persisted
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_security_activity_does_not_follow_a_previous_upstream_identity(
|
|
security_access, monkeypatch
|
|
):
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as timmy:
|
|
await timmy.post(
|
|
"/api/v1/session",
|
|
json={"access_token": "correct horse battery staple", "device_label": "Timmy phone"},
|
|
)
|
|
|
|
async def other_user():
|
|
return {"id": 84, "login": "other"}
|
|
|
|
monkeypatch.setattr(main, "current_user", other_user)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as other:
|
|
signed_in = await other.post(
|
|
"/api/v1/session",
|
|
json={"access_token": "correct horse battery staple", "device_label": "Other phone"},
|
|
)
|
|
activity = await other.get("/api/v1/security-events")
|
|
|
|
assert signed_in.status_code == 200
|
|
assert [event["device_label"] for event in activity.json()["events"]] == ["Other phone"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_tampered_security_activity_returns_no_partial_history(security_access):
|
|
transport = httpx.ASGITransport(app=main.app, raise_app_exceptions=False)
|
|
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",
|
|
},
|
|
)
|
|
with sqlite3.connect(security_access / "security.sqlite3") as connection:
|
|
payload = connection.execute(
|
|
"SELECT payload FROM security_events WHERE id = 1"
|
|
).fetchone()[0]
|
|
replacement = "A" if payload[-1] != "A" else "B"
|
|
connection.execute(
|
|
"UPDATE security_events SET payload = ? WHERE id = 1",
|
|
(payload[:-1] + replacement,),
|
|
)
|
|
|
|
activity = await client.get("/api/v1/security-events")
|
|
|
|
assert activity.status_code == 503
|
|
assert activity.json() == {
|
|
"detail": "Security activity is temporarily unavailable"
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_security_activity_limit_is_validated(security_access):
|
|
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"},
|
|
)
|
|
too_large = await client.get("/api/v1/security-events", params={"limit": 101})
|
|
invalid_cursor = await client.get("/api/v1/security-events", params={"cursor": 0})
|
|
|
|
assert too_large.status_code == 422
|
|
assert invalid_cursor.status_code == 422
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_passkey_enrollment_reservation_failure_preserves_registry(
|
|
security_access, monkeypatch
|
|
):
|
|
class VerifiedRegistration:
|
|
credential_id = b"phone-credential"
|
|
credential_public_key = b"credential-public-key"
|
|
sign_count = 0
|
|
|
|
class UnavailableJournal:
|
|
def reserve(self, *_args, **_kwargs):
|
|
raise SecurityEventStoreError("unavailable")
|
|
|
|
monkeypatch.setattr(
|
|
main.passkeys, "verify_registration", lambda **_kwargs: VerifiedRegistration()
|
|
)
|
|
transport = httpx.ASGITransport(app=main.app, raise_app_exceptions=False)
|
|
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"],
|
|
}
|
|
grant = await client.post(
|
|
"/api/v1/fresh-authorization",
|
|
json={
|
|
"access_token": "correct horse battery staple",
|
|
"action": "enroll_passkey",
|
|
"target": "current_device",
|
|
},
|
|
headers=headers,
|
|
)
|
|
options = await client.post(
|
|
"/api/v1/passkeys/registration/options",
|
|
headers={**headers, "X-Step-Up-Grant": grant.json()["grant"]},
|
|
)
|
|
monkeypatch.setattr(main, "_security_event_store", lambda: UnavailableJournal())
|
|
enrolled = await client.post(
|
|
"/api/v1/passkeys/registration/verify",
|
|
json={"challenge": options.json()["challenge"], "credential": {"id": "fake"}},
|
|
headers=headers,
|
|
)
|
|
|
|
assert enrolled.status_code == 503
|
|
assert enrolled.json() == {"detail": "Security activity is temporarily unavailable"}
|
|
assert main._passkey_store().all(principal_id=42) == []
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_passkey_enrollment_registry_failure_discards_reserved_event(
|
|
security_access, 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, raise_app_exceptions=False)
|
|
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"],
|
|
}
|
|
grant = await client.post(
|
|
"/api/v1/fresh-authorization",
|
|
json={
|
|
"access_token": "correct horse battery staple",
|
|
"action": "enroll_passkey",
|
|
"target": "current_device",
|
|
},
|
|
headers=headers,
|
|
)
|
|
options = await client.post(
|
|
"/api/v1/passkeys/registration/options",
|
|
headers={**headers, "X-Step-Up-Grant": grant.json()["grant"]},
|
|
)
|
|
store = main._passkey_store()
|
|
|
|
def fail_registration(**_kwargs):
|
|
raise main.dashboard_auth.SessionStoreError("unavailable")
|
|
|
|
monkeypatch.setattr(store, "register", fail_registration)
|
|
monkeypatch.setattr(main, "_passkey_store", lambda: store)
|
|
enrolled = await client.post(
|
|
"/api/v1/passkeys/registration/verify",
|
|
json={"challenge": options.json()["challenge"], "credential": {"id": "fake"}},
|
|
headers=headers,
|
|
)
|
|
|
|
events = SecurityEventStore(
|
|
security_access / "security.sqlite3", clock=lambda: 0
|
|
).list(principal_id=42, limit=10).events
|
|
assert enrolled.status_code == 503
|
|
assert enrolled.json() == {"detail": "Passkey registry is temporarily unavailable"}
|
|
assert all(event.kind != "passkey_enrolled" for event in events)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_passkey_enrollment_finalization_failure_retains_pending_event(
|
|
security_access, 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, raise_app_exceptions=False)
|
|
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"],
|
|
}
|
|
grant = await client.post(
|
|
"/api/v1/fresh-authorization",
|
|
json={
|
|
"access_token": "correct horse battery staple",
|
|
"action": "enroll_passkey",
|
|
"target": "current_device",
|
|
},
|
|
headers=headers,
|
|
)
|
|
options = await client.post(
|
|
"/api/v1/passkeys/registration/options",
|
|
headers={**headers, "X-Step-Up-Grant": grant.json()["grant"]},
|
|
)
|
|
journal = main._security_event_store()
|
|
|
|
def fail_finalization(_operation_id):
|
|
raise SecurityEventStoreError("unavailable")
|
|
|
|
monkeypatch.setattr(journal, "finalize", fail_finalization)
|
|
monkeypatch.setattr(main, "_security_event_store", lambda: journal)
|
|
enrolled = await client.post(
|
|
"/api/v1/passkeys/registration/verify",
|
|
json={"challenge": options.json()["challenge"], "credential": {"id": "fake"}},
|
|
headers=headers,
|
|
)
|
|
activity = await client.get("/api/v1/security-events")
|
|
|
|
event = activity.json()["events"][0]
|
|
assert enrolled.status_code == 201
|
|
assert enrolled.json() == {"enrolled": True}
|
|
assert event["kind"] == "passkey_enrolled"
|
|
assert event["status"] == "pending"
|
|
assert event["device_label"] == "Phone"
|
|
assert event["method"] == "passkey"
|
|
assert event["target"] == "passkey"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_revoked_device_activity_survives_live_session_removal(security_access):
|
|
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"},
|
|
)
|
|
devices = (await laptop.get("/api/v1/sessions")).json()["devices"]
|
|
phone_device = next(device for device in devices if device["device_label"] == "Phone")
|
|
csrf = laptop.cookies["stackchain_csrf"]
|
|
grant = await laptop.post(
|
|
"/api/v1/fresh-authorization",
|
|
json={
|
|
"access_token": "correct horse battery staple",
|
|
"action": "revoke_device",
|
|
"target": phone_device["management_id"],
|
|
},
|
|
headers={"Origin": "https://test", "X-CSRF-Token": csrf},
|
|
)
|
|
revoked = await laptop.delete(
|
|
f"/api/v1/sessions/{phone_device['management_id']}",
|
|
headers={
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": csrf,
|
|
"X-Step-Up-Grant": grant.json()["grant"],
|
|
},
|
|
)
|
|
active = (await laptop.get("/api/v1/sessions")).json()["devices"]
|
|
history = (await laptop.get("/api/v1/security-events")).json()["events"]
|
|
|
|
assert revoked.status_code == 200
|
|
assert [device["device_label"] for device in active] == ["Laptop"]
|
|
assert [(event["kind"], event["device_label"]) for event in history] == [
|
|
("device_revoked", "Phone"),
|
|
("sign_in", "Laptop"),
|
|
("sign_in", "Phone"),
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_remote_revocation_reservation_failure_preserves_the_device(
|
|
security_access, monkeypatch
|
|
):
|
|
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"},
|
|
)
|
|
devices = (await laptop.get("/api/v1/sessions")).json()["devices"]
|
|
target = next(device for device in devices if device["device_label"] == "Phone")
|
|
headers = {
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": laptop.cookies["stackchain_csrf"],
|
|
}
|
|
grant = await laptop.post(
|
|
"/api/v1/fresh-authorization",
|
|
json={
|
|
"access_token": "correct horse battery staple",
|
|
"action": "revoke_device",
|
|
"target": target["management_id"],
|
|
},
|
|
headers=headers,
|
|
)
|
|
|
|
class UnavailableJournal:
|
|
def reserve(self, *_args, **_kwargs):
|
|
raise SecurityEventStoreError("unavailable")
|
|
|
|
monkeypatch.setattr(main, "_security_event_store", lambda: UnavailableJournal())
|
|
response = await laptop.delete(
|
|
f"/api/v1/sessions/{target['management_id']}",
|
|
headers={**headers, "X-Step-Up-Grant": grant.json()["grant"]},
|
|
)
|
|
remaining = (await laptop.get("/api/v1/sessions")).json()["devices"]
|
|
|
|
assert response.status_code == 503
|
|
assert {device["device_label"] for device in remaining} == {"Phone", "Laptop"}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_successful_protected_actions_record_only_bounded_targets(
|
|
security_access, monkeypatch
|
|
):
|
|
async def yes(*_args):
|
|
return True
|
|
|
|
async def close(*_args):
|
|
return {"number": 7, "state": "closed"}
|
|
|
|
async def merge(*_args):
|
|
return {"number": 9, "merged": True, "state": "closed"}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", yes)
|
|
monkeypatch.setattr(main.gitea_proxy, "close_issue", close)
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", yes)
|
|
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", "device_label": "Phone"},
|
|
)
|
|
csrf_headers = {
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": client.cookies["stackchain_csrf"],
|
|
}
|
|
|
|
async def grant(action, target):
|
|
response = await client.post(
|
|
"/api/v1/fresh-authorization",
|
|
json={
|
|
"access_token": "correct horse battery staple",
|
|
"action": action,
|
|
"target": target,
|
|
},
|
|
headers=csrf_headers,
|
|
)
|
|
return response.json()["grant"]
|
|
|
|
issue_grant = await grant("close_issue", "stackchain/api#7")
|
|
closed = await client.patch(
|
|
"/api/v1/repos/stackchain/api/issues/7/close",
|
|
headers={**csrf_headers, "X-Step-Up-Grant": issue_grant},
|
|
)
|
|
pull_grant = await grant("merge_pull", "stackchain/app#9")
|
|
merged = await client.post(
|
|
"/api/v1/repos/stackchain/app/pulls/9/merge",
|
|
json={"expected_head_sha": "abc123"},
|
|
headers={**csrf_headers, "X-Step-Up-Grant": pull_grant},
|
|
)
|
|
events = (await client.get("/api/v1/security-events")).json()["events"]
|
|
|
|
assert closed.status_code == 200
|
|
assert merged.status_code == 200
|
|
assert [(event["kind"], event["target"]) for event in events[:2]] == [
|
|
("pull_merged", "stackchain/app#9"),
|
|
("issue_closed", "stackchain/api#7"),
|
|
]
|
|
persisted = (security_access / "security.sqlite3").read_bytes()
|
|
assert b"abc123" not in persisted
|
|
|
|
|
|
@pytest.mark.anyio
|
|
@pytest.mark.parametrize(
|
|
"path",
|
|
[
|
|
"/api/v1/repos/stackchain/api/issues/7/comments/42",
|
|
"/api/v1/repos/stackchain/api/pulls/7/comments/42",
|
|
"/api/v1/notifications/91/comments/42",
|
|
],
|
|
)
|
|
async def test_comment_deletion_requires_exact_fresh_authorization_before_gitea(
|
|
security_access, monkeypatch, path
|
|
):
|
|
calls = []
|
|
|
|
async def assigned(*_args):
|
|
return True
|
|
|
|
async def notification_target(thread_id):
|
|
assert thread_id == 91
|
|
return "stackchain/api", 7
|
|
|
|
async def delete(*args):
|
|
calls.append(args)
|
|
return {"id": 42, "deleted": True}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned)
|
|
monkeypatch.setattr(main.gitea_proxy, "notification_conversation_target", notification_target)
|
|
monkeypatch.setattr(main.gitea_proxy, "delete_owned_comment", delete)
|
|
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"],
|
|
}
|
|
response = await client.delete(path, headers=headers)
|
|
|
|
assert response.status_code == 428
|
|
assert response.json()["detail"] == {
|
|
"detail": "Fresh authorization required",
|
|
"code": "step_up_required",
|
|
"action": "delete_comment",
|
|
"target": "stackchain/api#7:42",
|
|
}
|
|
assert calls == []
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_authorized_comment_deletion_records_only_privacy_safe_target(
|
|
security_access, monkeypatch
|
|
):
|
|
async def assigned(*_args):
|
|
return True
|
|
|
|
async def delete(*_args):
|
|
return {"id": 42, "deleted": True}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
|
|
monkeypatch.setattr(main.gitea_proxy, "delete_owned_comment", delete)
|
|
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"],
|
|
}
|
|
grant = await client.post(
|
|
"/api/v1/fresh-authorization",
|
|
json={
|
|
"access_token": "correct horse battery staple",
|
|
"action": "delete_comment",
|
|
"target": "stackchain/api#7:42",
|
|
},
|
|
headers=headers,
|
|
)
|
|
response = await client.delete(
|
|
"/api/v1/repos/stackchain/api/issues/7/comments/42",
|
|
headers={**headers, "X-Step-Up-Grant": grant.json()["grant"]},
|
|
)
|
|
events = (await client.get("/api/v1/security-events")).json()["events"]
|
|
|
|
assert response.status_code == 200
|
|
assert (events[0]["kind"], events[0]["target"], events[0]["status"]) == (
|
|
"comment_deleted", "stackchain/api#7:42", "completed"
|
|
)
|
|
persisted = (security_access / "security.sqlite3").read_bytes()
|
|
assert b"private comment body" not in persisted
|
|
assert b"correct horse battery staple" not in persisted
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_comment_deletion_reservation_failure_prevents_gitea_mutation(
|
|
security_access, monkeypatch
|
|
):
|
|
calls = []
|
|
|
|
async def assigned(*_args):
|
|
return True
|
|
|
|
async def delete(*args):
|
|
calls.append(args)
|
|
return {"id": 42, "deleted": True}
|
|
|
|
class UnavailableJournal:
|
|
def reserve(self, *_args, **_kwargs):
|
|
raise SecurityEventStoreError("unavailable")
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
|
|
monkeypatch.setattr(main.gitea_proxy, "delete_owned_comment", delete)
|
|
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"],
|
|
}
|
|
grant = await client.post(
|
|
"/api/v1/fresh-authorization",
|
|
json={
|
|
"access_token": "correct horse battery staple",
|
|
"action": "delete_comment",
|
|
"target": "stackchain/api#7:42",
|
|
},
|
|
headers=headers,
|
|
)
|
|
monkeypatch.setattr(main, "_security_event_store", lambda: UnavailableJournal())
|
|
response = await client.delete(
|
|
"/api/v1/repos/stackchain/api/issues/7/comments/42",
|
|
headers={**headers, "X-Step-Up-Grant": grant.json()["grant"]},
|
|
)
|
|
|
|
assert response.status_code == 503
|
|
assert calls == []
|
|
|
|
|
|
@pytest.mark.anyio
|
|
@pytest.mark.parametrize("finalize_fails", [False, True])
|
|
async def test_comment_deletion_journal_tracks_failed_and_confirmed_outcomes(
|
|
security_access, monkeypatch, finalize_fails
|
|
):
|
|
journal_calls = []
|
|
|
|
async def assigned(*_args):
|
|
return True
|
|
|
|
async def delete(*_args):
|
|
if not finalize_fails:
|
|
raise RuntimeError("upstream rejected deletion")
|
|
return {"id": 42, "deleted": True}
|
|
|
|
class Journal:
|
|
def reserve(self, kind, *, principal_id, target):
|
|
assert principal_id == 42
|
|
journal_calls.append(("reserve", kind, target))
|
|
return "operation-42"
|
|
|
|
def discard(self, operation_id):
|
|
journal_calls.append(("discard", operation_id))
|
|
|
|
def finalize(self, operation_id):
|
|
journal_calls.append(("finalize", operation_id))
|
|
raise SecurityEventStoreError("unavailable")
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
|
|
monkeypatch.setattr(main.gitea_proxy, "delete_owned_comment", delete)
|
|
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"],
|
|
}
|
|
grant = await client.post(
|
|
"/api/v1/fresh-authorization",
|
|
json={
|
|
"access_token": "correct horse battery staple",
|
|
"action": "delete_comment",
|
|
"target": "stackchain/api#7:42",
|
|
},
|
|
headers=headers,
|
|
)
|
|
monkeypatch.setattr(main, "_security_event_store", lambda: Journal())
|
|
response = await client.delete(
|
|
"/api/v1/repos/stackchain/api/issues/7/comments/42",
|
|
headers={**headers, "X-Step-Up-Grant": grant.json()["grant"]},
|
|
)
|
|
|
|
assert journal_calls[0] == (
|
|
"reserve", "comment_deleted", "stackchain/api#7:42"
|
|
)
|
|
if finalize_fails:
|
|
assert response.status_code == 200
|
|
assert response.json() == {"id": 42, "deleted": True}
|
|
assert journal_calls[1] == ("finalize", "operation-42")
|
|
else:
|
|
assert response.status_code == 503
|
|
assert journal_calls[1] == ("discard", "operation-42")
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_approved_pull_review_records_a_privacy_safe_completed_event(
|
|
security_access, monkeypatch
|
|
):
|
|
async def requested(*_args):
|
|
return True
|
|
|
|
async def submit(*_args):
|
|
return {"id": 91, "state": "APPROVED"}
|
|
|
|
monkeypatch.setattr(main, "is_requested_review", requested)
|
|
monkeypatch.setattr(main.gitea_proxy, "submit_pull_review", submit)
|
|
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"],
|
|
}
|
|
grant = await client.post(
|
|
"/api/v1/fresh-authorization",
|
|
json={
|
|
"access_token": "correct horse battery staple",
|
|
"action": "submit_pull_review",
|
|
"target": "stackchain/api#7@abc123:approve",
|
|
},
|
|
headers=headers,
|
|
)
|
|
response = await client.post(
|
|
"/api/v1/repos/stackchain/api/pulls/7/review",
|
|
json={
|
|
"expected_head_sha": "abc123",
|
|
"decision": "approve",
|
|
"body": "Ready to ship with private review notes.",
|
|
},
|
|
headers={**headers, "X-Step-Up-Grant": grant.json()["grant"]},
|
|
)
|
|
events = (await client.get("/api/v1/security-events")).json()["events"]
|
|
|
|
assert response.status_code == 201
|
|
assert events[0] == {
|
|
"id": events[0]["id"],
|
|
"kind": "pull_review_approved",
|
|
"method": None,
|
|
"device_label": None,
|
|
"target": "stackchain/api#7",
|
|
"created_at": events[0]["created_at"],
|
|
"status": "completed",
|
|
}
|
|
persisted = (security_access / "security.sqlite3").read_bytes()
|
|
assert b"abc123" not in persisted
|
|
assert b"private review notes" not in persisted
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_review_reservation_failure_prevents_the_gitea_mutation(
|
|
security_access, monkeypatch
|
|
):
|
|
submit_calls = []
|
|
|
|
async def requested(*_args):
|
|
return True
|
|
|
|
async def submit(*args):
|
|
submit_calls.append(args)
|
|
return {"id": 91, "state": "APPROVED"}
|
|
|
|
class UnavailableJournal:
|
|
def reserve(self, *_args, **_kwargs):
|
|
raise SecurityEventStoreError("unavailable")
|
|
|
|
monkeypatch.setattr(main, "is_requested_review", requested)
|
|
monkeypatch.setattr(main.gitea_proxy, "submit_pull_review", submit)
|
|
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"},
|
|
)
|
|
headers = {
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": client.cookies["stackchain_csrf"],
|
|
}
|
|
grant = await client.post(
|
|
"/api/v1/fresh-authorization",
|
|
json={
|
|
"access_token": "correct horse battery staple",
|
|
"action": "submit_pull_review",
|
|
"target": "stackchain/api#7@abc123:approve",
|
|
},
|
|
headers=headers,
|
|
)
|
|
monkeypatch.setattr(main, "_security_event_store", lambda: UnavailableJournal())
|
|
response = await client.post(
|
|
"/api/v1/repos/stackchain/api/pulls/7/review",
|
|
json={
|
|
"expected_head_sha": "abc123",
|
|
"decision": "approve",
|
|
"body": "Ready.",
|
|
},
|
|
headers={**headers, "X-Step-Up-Grant": grant.json()["grant"]},
|
|
)
|
|
|
|
assert response.status_code == 503
|
|
assert response.json()["detail"] == "Security activity is temporarily unavailable"
|
|
assert submit_calls == []
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_issue_close_reservation_failure_prevents_the_gitea_mutation(
|
|
security_access, monkeypatch
|
|
):
|
|
close_calls = []
|
|
|
|
async def yes(*_args):
|
|
return True
|
|
|
|
async def close(*args):
|
|
close_calls.append(args)
|
|
return {"number": 7, "state": "closed"}
|
|
|
|
class UnavailableJournal:
|
|
def reserve(self, *_args, **_kwargs):
|
|
raise SecurityEventStoreError("unavailable")
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", yes)
|
|
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 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"],
|
|
}
|
|
grant = await client.post(
|
|
"/api/v1/fresh-authorization",
|
|
json={
|
|
"access_token": "correct horse battery staple",
|
|
"action": "close_issue",
|
|
"target": "stackchain/api#7",
|
|
},
|
|
headers=headers,
|
|
)
|
|
monkeypatch.setattr(main, "_security_event_store", lambda: UnavailableJournal())
|
|
response = await client.patch(
|
|
"/api/v1/repos/stackchain/api/issues/7/close",
|
|
headers={**headers, "X-Step-Up-Grant": grant.json()["grant"]},
|
|
)
|
|
|
|
assert response.status_code == 503
|
|
assert close_calls == []
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_issue_close_finalization_failure_still_reports_the_closed_issue(
|
|
security_access, monkeypatch
|
|
):
|
|
async def yes(*_args):
|
|
return True
|
|
|
|
async def close(*_args):
|
|
return {"number": 7, "state": "closed"}
|
|
|
|
class FinalizationUnavailable:
|
|
def reserve(self, *_args, **_kwargs):
|
|
return "durable-operation"
|
|
|
|
def finalize(self, _operation_id):
|
|
raise SecurityEventStoreError("unavailable")
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", yes)
|
|
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 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"],
|
|
}
|
|
grant = await client.post(
|
|
"/api/v1/fresh-authorization",
|
|
json={
|
|
"access_token": "correct horse battery staple",
|
|
"action": "close_issue",
|
|
"target": "stackchain/api#7",
|
|
},
|
|
headers=headers,
|
|
)
|
|
monkeypatch.setattr(
|
|
main, "_security_event_store", lambda: FinalizationUnavailable()
|
|
)
|
|
response = await client.patch(
|
|
"/api/v1/repos/stackchain/api/issues/7/close",
|
|
headers={**headers, "X-Step-Up-Grant": grant.json()["grant"]},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {"number": 7, "state": "closed"}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_sign_out_is_journaled_before_the_session_is_removed(security_access):
|
|
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"},
|
|
)
|
|
signed_out = await client.delete(
|
|
"/api/v1/session",
|
|
headers={
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": client.cookies["stackchain_csrf"],
|
|
},
|
|
)
|
|
|
|
events = SecurityEventStore(
|
|
security_access / "security.sqlite3", clock=lambda: 0
|
|
).list(principal_id=42, limit=10).events
|
|
assert signed_out.status_code == 200
|
|
assert [(event.kind, event.device_label) for event in events[:2]] == [
|
|
("sign_out", None),
|
|
("sign_in", "Phone"),
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_sign_out_reservation_failure_preserves_the_active_session(
|
|
security_access, 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", "device_label": "Phone"},
|
|
)
|
|
|
|
class UnavailableJournal:
|
|
def reserve(self, *_args, **_kwargs):
|
|
raise SecurityEventStoreError("unavailable")
|
|
|
|
monkeypatch.setattr(main, "_security_event_store", lambda: UnavailableJournal())
|
|
response = await client.delete(
|
|
"/api/v1/session",
|
|
headers={
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": client.cookies["stackchain_csrf"],
|
|
},
|
|
)
|
|
still_authenticated = await client.get("/api/v1/sessions")
|
|
|
|
assert response.status_code == 503
|
|
assert still_authenticated.status_code == 200
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_sign_out_finalization_failure_reports_success_and_clears_cookies(
|
|
security_access, 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", "device_label": "Phone"},
|
|
)
|
|
|
|
class FinalizationUnavailable:
|
|
def reserve(self, *_args, **_kwargs):
|
|
return "durable-operation"
|
|
|
|
def finalize(self, _operation_id):
|
|
raise SecurityEventStoreError("unavailable")
|
|
|
|
monkeypatch.setattr(
|
|
main, "_security_event_store", lambda: FinalizationUnavailable()
|
|
)
|
|
response = await client.delete(
|
|
"/api/v1/session",
|
|
headers={
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": client.cookies["stackchain_csrf"],
|
|
},
|
|
)
|
|
rejected = await client.get("/api/v1/sessions")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["authenticated"] is False
|
|
assert "stackchain_session=" in response.headers["set-cookie"]
|
|
assert rejected.status_code == 401
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_sign_out_all_is_journaled_after_sessions_are_removed(security_access):
|
|
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"},
|
|
)
|
|
csrf_headers = {
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": client.cookies["stackchain_csrf"],
|
|
}
|
|
grant = await client.post(
|
|
"/api/v1/fresh-authorization",
|
|
json={
|
|
"access_token": "correct horse battery staple",
|
|
"action": "revoke_all_sessions",
|
|
"target": "all",
|
|
},
|
|
headers=csrf_headers,
|
|
)
|
|
response = await client.delete(
|
|
"/api/v1/sessions",
|
|
headers={**csrf_headers, "X-Step-Up-Grant": grant.json()["grant"]},
|
|
)
|
|
|
|
events = SecurityEventStore(
|
|
security_access / "security.sqlite3", clock=lambda: 0
|
|
).list(principal_id=42, limit=10).events
|
|
assert response.status_code == 200
|
|
assert events[0].kind == "all_sessions_revoked"
|
|
assert events[0].target == "all_devices"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_sign_out_all_reservation_failure_preserves_the_active_session(
|
|
security_access, 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", "device_label": "Phone"},
|
|
)
|
|
headers = {
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": client.cookies["stackchain_csrf"],
|
|
}
|
|
grant = await client.post(
|
|
"/api/v1/fresh-authorization",
|
|
json={
|
|
"access_token": "correct horse battery staple",
|
|
"action": "revoke_all_sessions",
|
|
"target": "all",
|
|
},
|
|
headers=headers,
|
|
)
|
|
|
|
class UnavailableJournal:
|
|
def reserve(self, *_args, **_kwargs):
|
|
raise SecurityEventStoreError("unavailable")
|
|
|
|
monkeypatch.setattr(main, "_security_event_store", lambda: UnavailableJournal())
|
|
response = await client.delete(
|
|
"/api/v1/sessions",
|
|
headers={**headers, "X-Step-Up-Grant": grant.json()["grant"]},
|
|
)
|
|
still_authenticated = await client.get("/api/v1/sessions")
|
|
|
|
assert response.status_code == 503
|
|
assert still_authenticated.status_code == 200
|