fix(security): F10 (Astra P1) — Token-Scopes sind Obergrenze, kein Ersatz fuer User-Rechte
Vorher: require_permission machte bei passendem Token-Scope ein early-return — die User-Rechte wurden NIE geprueft. Ein Token mit mail:write erlaubte mail:write selbst dann, wenn der Benutzer die Berechtigung nie hatte oder sie entzogen bekam (Astra-Repro isoliert bestaetigt). Fix: Nach bestandenem Scope-Check in den normalen User-Rechte-Check fallen. Effektives Recht = User-Rechte UND Token-Scope. Rechteentzug wirkt sofort auf bestehende Tokens. System-Admin-Bypass unveraendert. Tests: tests/test_s1_security_guards.py 7/7 (neue Suite): Scope-ohne-User-Recht 403, beide-present pass, Scope-fehlt-User-hat 403 insufficient_scope, Deny-Revocation wirkt, Session-Pfad unveraendert, *:*-Scope umgeht nicht, Admin-Bypass bleibt. ruff clean.
This commit is contained in:
+5
-1
@@ -311,6 +311,10 @@ def require_permission(permission: str):
|
|||||||
current_user: dict[str, Any] = Depends(get_current_user),
|
current_user: dict[str, Any] = Depends(get_current_user),
|
||||||
) -> dict[str, Any]:
|
) -> dict[str, Any]:
|
||||||
# API token scope enforcement (Problem 2 fix)
|
# API token scope enforcement (Problem 2 fix)
|
||||||
|
# F10 (Astra): token scopes are an UPPER BOUND, not a replacement —
|
||||||
|
# the user's own permissions must ALSO grant the permission. A token
|
||||||
|
# can never grant more than its owner has; revoking the user's
|
||||||
|
# permission takes effect on existing tokens.
|
||||||
token_scopes = current_user.get("_token_scopes")
|
token_scopes = current_user.get("_token_scopes")
|
||||||
if token_scopes is not None:
|
if token_scopes is not None:
|
||||||
from app.core.permissions import _permission_matches_any
|
from app.core.permissions import _permission_matches_any
|
||||||
@@ -322,7 +326,7 @@ def require_permission(permission: str):
|
|||||||
"code": "insufficient_scope",
|
"code": "insufficient_scope",
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
return current_user
|
# fall through: the normal user-permission check applies too
|
||||||
|
|
||||||
if current_user.get("is_system_admin"):
|
if current_user.get("is_system_admin"):
|
||||||
return current_user
|
return current_user
|
||||||
|
|||||||
@@ -0,0 +1,113 @@
|
|||||||
|
"""S1 security guard tests (Astra audit findings — Phase S Welle 1).
|
||||||
|
|
||||||
|
F10: token scopes must be an UPPER BOUND, not a replacement for user
|
||||||
|
permissions. A token can never grant more than its owner has.
|
||||||
|
|
||||||
|
Covers require_permission directly (unit level, no HTTP) plus the
|
||||||
|
acceptance criteria from docs/audits/astra-audit-2026-09-17.md.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import uuid
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from fastapi import HTTPException
|
||||||
|
|
||||||
|
from app.deps import require_permission
|
||||||
|
|
||||||
|
|
||||||
|
def _user_ctx(
|
||||||
|
permissions: list[str] | None = None,
|
||||||
|
denied: list[str] | None = None,
|
||||||
|
token_scopes: list[str] | None = None,
|
||||||
|
is_system_admin: bool = False,
|
||||||
|
) -> dict:
|
||||||
|
"""Build a current_user dict in the session/bearer context shape."""
|
||||||
|
ctx: dict = {
|
||||||
|
"user_id": str(uuid.uuid4()),
|
||||||
|
"tenant_id": str(uuid.uuid4()),
|
||||||
|
"permissions": permissions or [],
|
||||||
|
"denied_permissions": denied or [],
|
||||||
|
"field_permissions": {},
|
||||||
|
"is_system_admin": is_system_admin,
|
||||||
|
"role": "user",
|
||||||
|
}
|
||||||
|
if token_scopes is not None:
|
||||||
|
ctx["_token_scopes"] = token_scopes
|
||||||
|
return ctx
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
class TestF10TokenScopesAreUpperBound:
|
||||||
|
"""F10 (Astra P1): token scopes restrict, they never grant."""
|
||||||
|
|
||||||
|
async def test_token_scope_does_not_replace_user_permission(self):
|
||||||
|
"""Token with mail:write scope, user WITHOUT mail:write → 403.
|
||||||
|
|
||||||
|
Astra repro: this exact combination was allowed before the fix.
|
||||||
|
"""
|
||||||
|
checker = require_permission("mail:write")
|
||||||
|
ctx = _user_ctx(permissions=[], token_scopes=["mail:write"])
|
||||||
|
with pytest.raises(HTTPException) as exc_info:
|
||||||
|
await checker(current_user=ctx)
|
||||||
|
assert exc_info.value.status_code == 403
|
||||||
|
assert exc_info.value.detail.get("code") != "insufficient_scope"
|
||||||
|
|
||||||
|
async def test_token_scope_with_user_permission_passes(self):
|
||||||
|
"""Token scope AND user permission both present → allowed."""
|
||||||
|
checker = require_permission("mail:write")
|
||||||
|
ctx = _user_ctx(permissions=["mail:write"], token_scopes=["mail:write"])
|
||||||
|
result = await checker(current_user=ctx)
|
||||||
|
assert result is ctx
|
||||||
|
|
||||||
|
async def test_missing_scope_rejected_even_with_user_permission(self):
|
||||||
|
"""User HAS the permission but the token lacks the scope → 403
|
||||||
|
insufficient_scope (the token may do less than its owner)."""
|
||||||
|
checker = require_permission("mail:write")
|
||||||
|
ctx = _user_ctx(permissions=["mail:write"], token_scopes=["mail:read"])
|
||||||
|
with pytest.raises(HTTPException) as exc_info:
|
||||||
|
await checker(current_user=ctx)
|
||||||
|
assert exc_info.value.status_code == 403
|
||||||
|
assert exc_info.value.detail.get("code") == "insufficient_scope"
|
||||||
|
|
||||||
|
async def test_permission_revocation_applies_to_existing_tokens(self):
|
||||||
|
"""User permission revoked (moved to deny) → token rejected.
|
||||||
|
|
||||||
|
Astra acceptance: later permission revocation must take effect on
|
||||||
|
existing tokens.
|
||||||
|
"""
|
||||||
|
checker = require_permission("mail:write")
|
||||||
|
ctx = _user_ctx(
|
||||||
|
permissions=["mail:write"],
|
||||||
|
denied=["mail:write"],
|
||||||
|
token_scopes=["mail:write"],
|
||||||
|
)
|
||||||
|
with pytest.raises(HTTPException) as exc_info:
|
||||||
|
await checker(current_user=ctx)
|
||||||
|
assert exc_info.value.status_code == 403
|
||||||
|
|
||||||
|
async def test_session_path_unchanged(self):
|
||||||
|
"""Without token scopes the normal permission check applies."""
|
||||||
|
checker = require_permission("mail:write")
|
||||||
|
ok_ctx = _user_ctx(permissions=["mail:write"])
|
||||||
|
assert await checker(current_user=ok_ctx) is ok_ctx
|
||||||
|
|
||||||
|
bad_ctx = _user_ctx(permissions=[])
|
||||||
|
with pytest.raises(HTTPException) as exc_info:
|
||||||
|
await checker(current_user=bad_ctx)
|
||||||
|
assert exc_info.value.status_code == 403
|
||||||
|
|
||||||
|
async def test_wildcard_scope_still_bound_to_user_permissions(self):
|
||||||
|
"""A *:* token scope cannot bypass missing user permissions."""
|
||||||
|
checker = require_permission("mail:write")
|
||||||
|
ctx = _user_ctx(permissions=[], token_scopes=['*:*'])
|
||||||
|
with pytest.raises(HTTPException) as exc_info:
|
||||||
|
await checker(current_user=ctx)
|
||||||
|
assert exc_info.value.status_code == 403
|
||||||
|
|
||||||
|
async def test_system_admin_bypass_unchanged(self):
|
||||||
|
"""System admins keep their bypass (both session and token path)."""
|
||||||
|
checker = require_permission("mail:write")
|
||||||
|
ctx = _user_ctx(token_scopes=["mail:write"], is_system_admin=True)
|
||||||
|
assert await checker(current_user=ctx) is ctx
|
||||||
Reference in New Issue
Block a user