fix(security): F11 (Astra P1) — Freigaben an Entscheider, Ablauf und Atomizitaet binden
Vorher: resolve_approval_request pruefte nur Mandant + pending — NICHT
Ablaufdatum, NICHT den vorgesehenen Genehmiger, und ueberschrieb
approver_id mit dem tatsaechlichen Entscheider (Zuordnung verloren).
Astra-Repro: Eine abgelaufene Anfrage konnte von einem anderen
Entscheider genehmigt werden; konkurrierende Entscheidungen waren
moeglich.
Fix:
- Migration 0146: neue Spalte resolved_by (Zuordnung vs. Entscheider
getrennt — approver_id bleibt die ZUORDNUNG)
- resolve_approval_request komplett ueberarbeitet:
* Ablauf-Check: expires_at vorbei -> Status expired + 410
* Genehmiger-Check: approver_id match ODER approver_group-Mitgliedschaft;
unassigned = jeder mit approvals:approve; System-Admin als
dokumentierter Ops-Override; falscher Entscheider -> 403
* Atomarer Statusuebergang: UPDATE ... WHERE status=pending —
konkurrierende Entscheidung -> 409
* approver_id wird NIE ueberschrieben; resolved_by dokumentiert den
Entscheider
- ApprovalDecisionError mit HTTP-Status-Codes; approve/reject-Routen
fangen sie sauber ab (404/409/410/403 statt Flat-404)
- ApprovalResponse + Mapper um resolved_by ergaenzt
Abnahme (Astra): Falscher Entscheider, abgelaufene Anfrage und doppelte
Entscheidung werden abgewiesen — erfuellt (6 Tests).
Hinweis: workflows.py approve/reject-Aufrufer waren bereits kaputt
(F12, S2-Welle: approval[id] auf ORM-Objekt) und werden dort gefixt.
Tests: test_s1_security_guards.py 18/18 (6 neue F11-Tests). ruff clean.
Damit ist S1 — ALLE 11 Sicherheits-Findings der Astra-Welle 1 gefixt.
This commit is contained in:
@@ -263,3 +263,144 @@ class TestF23RestoreRequiresSystemAdmin:
|
||||
"automation:admin) — a tenant admin must not trigger a "
|
||||
"full-database restore"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestF11ApprovalBinding:
|
||||
"""F11 (Astra P1): approvals bound to decider, expiry and atomicity.
|
||||
|
||||
Acceptance (Astra):
|
||||
- wrong decider is rejected
|
||||
- expired request is rejected and marked expired
|
||||
- a concurrent/duplicate decision does not win twice
|
||||
- assignment (approver_id) is preserved; resolved_by records the decider
|
||||
"""
|
||||
|
||||
async def _seed_request(self, db_session, **overrides):
|
||||
"""Create a pending ApprovalRequest row for testing."""
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
from app.core.approval import create_approval_request
|
||||
|
||||
defaults = dict(
|
||||
entity_type="contact",
|
||||
entity_id=uuid.uuid4(),
|
||||
action="tool:send_mail",
|
||||
requested_by=uuid.uuid4(),
|
||||
requested_by_type="agent",
|
||||
)
|
||||
defaults.update(overrides)
|
||||
req = await create_approval_request(db_session, uuid.uuid4(), **defaults)
|
||||
await db_session.flush()
|
||||
return req
|
||||
|
||||
async def test_wrong_approver_rejected(self, db_session):
|
||||
"""A request assigned to user A cannot be decided by user B."""
|
||||
from app.core.approval import ApprovalDecisionError, resolve_approval_request
|
||||
|
||||
assigned = uuid.uuid4()
|
||||
req = await self._seed_request(db_session, approver_id=assigned)
|
||||
other = uuid.uuid4()
|
||||
|
||||
with pytest.raises(ApprovalDecisionError) as exc_info:
|
||||
await resolve_approval_request(
|
||||
db_session, req.tenant_id, req.id,
|
||||
decision="approved", approver_id=other,
|
||||
)
|
||||
assert exc_info.value.code == "wrong_approver"
|
||||
assert exc_info.value.http_status == 403
|
||||
# request stays pending
|
||||
await db_session.refresh(req)
|
||||
assert req.status == "pending"
|
||||
|
||||
async def test_assigned_approver_can_decide(self, db_session):
|
||||
"""The assigned approver may decide; assignment survives."""
|
||||
from app.core.approval import resolve_approval_request
|
||||
|
||||
assigned = uuid.uuid4()
|
||||
req = await self._seed_request(db_session, approver_id=assigned)
|
||||
|
||||
result = await resolve_approval_request(
|
||||
db_session, req.tenant_id, req.id,
|
||||
decision="approved", approver_id=assigned,
|
||||
)
|
||||
assert result is not None and result.status == "approved"
|
||||
# F11: assignment preserved, decider recorded separately
|
||||
assert result.approver_id == assigned
|
||||
assert result.resolved_by == assigned
|
||||
|
||||
async def test_expired_request_rejected_and_marked(self, db_session):
|
||||
"""An expired request cannot be decided — even by its assignee."""
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
from app.core.approval import ApprovalDecisionError, resolve_approval_request
|
||||
|
||||
assigned = uuid.uuid4()
|
||||
req = await self._seed_request(
|
||||
db_session,
|
||||
approver_id=assigned,
|
||||
expires_at=datetime.now(UTC) - timedelta(hours=1),
|
||||
)
|
||||
|
||||
with pytest.raises(ApprovalDecisionError) as exc_info:
|
||||
await resolve_approval_request(
|
||||
db_session, req.tenant_id, req.id,
|
||||
decision="approved", approver_id=assigned,
|
||||
)
|
||||
assert exc_info.value.code == "expired"
|
||||
assert exc_info.value.http_status == 410
|
||||
await db_session.refresh(req)
|
||||
assert req.status == "expired"
|
||||
|
||||
async def test_already_decided_rejected(self, db_session):
|
||||
"""A second decision (concurrent or duplicate) is rejected with 409."""
|
||||
from app.core.approval import ApprovalDecisionError, resolve_approval_request
|
||||
|
||||
assigned = uuid.uuid4()
|
||||
req = await self._seed_request(db_session, approver_id=assigned)
|
||||
|
||||
first = await resolve_approval_request(
|
||||
db_session, req.tenant_id, req.id,
|
||||
decision="approved", approver_id=assigned,
|
||||
)
|
||||
assert first is not None
|
||||
|
||||
with pytest.raises(ApprovalDecisionError) as exc_info:
|
||||
await resolve_approval_request(
|
||||
db_session, req.tenant_id, req.id,
|
||||
decision="rejected", approver_id=assigned,
|
||||
)
|
||||
assert exc_info.value.code == "not_pending"
|
||||
assert exc_info.value.http_status == 409
|
||||
|
||||
async def test_unassigned_request_any_decider(self, db_session):
|
||||
"""Unassigned requests (no approver_id, no group) may be decided
|
||||
by anyone — the approvals:approve permission is enforced by the
|
||||
route dependency, not the resolver."""
|
||||
from app.core.approval import resolve_approval_request
|
||||
|
||||
req = await self._seed_request(db_session) # no assignment
|
||||
decider = uuid.uuid4()
|
||||
result = await resolve_approval_request(
|
||||
db_session, req.tenant_id, req.id,
|
||||
decision="approved", approver_id=decider,
|
||||
)
|
||||
assert result is not None
|
||||
assert result.resolved_by == decider
|
||||
assert result.approver_id is None # assignment untouched
|
||||
|
||||
async def test_system_admin_override_documented(self, db_session):
|
||||
"""System admins may decide assigned requests — documented
|
||||
operations override (Astra: Ausnahme dokumentiert)."""
|
||||
from app.core.approval import resolve_approval_request
|
||||
|
||||
assigned = uuid.uuid4()
|
||||
req = await self._seed_request(db_session, approver_id=assigned)
|
||||
sysadmin = uuid.uuid4()
|
||||
result = await resolve_approval_request(
|
||||
db_session, req.tenant_id, req.id,
|
||||
decision="approved", approver_id=sysadmin, is_system_admin=True,
|
||||
)
|
||||
assert result is not None
|
||||
assert result.approver_id == assigned # assignment preserved
|
||||
assert result.resolved_by == sysadmin # decider recorded
|
||||
|
||||
Reference in New Issue
Block a user