leocrm/tests/test_s1_security_guards.py

"""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