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


@pytest.mark.asyncio
class TestF03SessionRevocation:
    """F03 (Astra P1): revocation must work in BOTH session stores.

    Deactivation, deletion and password changes kill Redis runtime sessions
    AND expire PostgreSQL fallback sessions — a Redis outage must not let a
    revoked session come back to life.
    """

    async def test_revoke_helper_expires_db_sessions(self, monkeypatch):
        """revoke_user_sessions_all_stores expires PG sessions (expires_at=now)
        instead of deleting them — audit trail stays, access dies."""
        import json
        from unittest.mock import AsyncMock, MagicMock

        import app.core.auth as auth_mod

        # Fake Redis with one live session for our user (async iteration —
        # the real client's scan_iter is an async iterator)
        user_id = str(uuid.uuid4())
        redis = MagicMock()
        redis.scan_iter = MagicMock(return_value=iter(["session:abc"]))
        redis.get = AsyncMock(return_value=json.dumps({"user_id": user_id}))
        redis.delete = AsyncMock()
        monkeypatch.setattr(auth_mod, "get_redis", lambda: redis)
        # Patch the inner iteration: real code does `async for key in scan_iter`

        class _AsyncIter:
            def __init__(self, items):
                self._items = iter(items)

            def __aiter__(self):
                return self

            async def __anext__(self):
                try:
                    return next(self._items)
                except StopIteration:
                    raise StopAsyncIteration from None

        redis.scan_iter = MagicMock(return_value=_AsyncIter(["session:abc"]))

        # Fake DB factory recording the update
        db = MagicMock()
        db.execute = AsyncMock(return_value=MagicMock(rowcount=2))
        db.commit = AsyncMock()
        factory = MagicMock()
        factory.return_value.__aenter__ = AsyncMock(return_value=db)
        factory.return_value.__aexit__ = AsyncMock(return_value=False)
        import app.core.db as db_mod

        monkeypatch.setattr(db_mod, "get_session_factory", lambda: factory)

        await auth_mod.revoke_user_sessions_all_stores(user_id)

        # Redis session deleted
        redis.delete.assert_awaited_once()
        # DB update executed (expire) + committed
        db.execute.assert_awaited_once()
        db.commit.assert_awaited_once()

    async def test_db_fallback_rejects_expired_session(self, monkeypatch):
        """get_session_data (DB fallback) must reject sessions whose
        expires_at is in the past — the revoke helper relies on this."""
        from datetime import UTC, datetime, timedelta
        from unittest.mock import AsyncMock, MagicMock

        import app.core.auth as auth_mod
        from app.models.session import Session as SessionModel

        # Force DB fallback path: circuit says "cannot proceed"
        circuit = MagicMock()
        circuit.can_proceed = AsyncMock(return_value=False)
        monkeypatch.setattr(
            "app.core.resilience.get_circuit", lambda name: circuit
        )

        expired_session = MagicMock(spec=SessionModel)
        expired_session.expires_at = datetime.now(UTC) - timedelta(hours=1)
        db = MagicMock()
        db.execute = AsyncMock(return_value=MagicMock(scalar_one_or_none=lambda: expired_session))
        factory = MagicMock()
        factory.return_value.__aenter__ = AsyncMock(return_value=db)
        factory.return_value.__aexit__ = AsyncMock(return_value=False)
        import app.core.db as db_mod

        monkeypatch.setattr(db_mod, "get_auth_session_factory", lambda: factory)

        result = await auth_mod.get_session_data(MagicMock(), "some-session-id")
        assert result is None

    async def test_membership_check_fails_closed(self):
        """F03: a session whose tenant membership is MISSING must be
        rejected (403), not waved through.

        Pins the decision rule used by get_current_user (deps.py):
        reject when `membership_status is None or != 'active'`.
        """
        def _rejects(status: str | None) -> bool:
            return status is None or status != "active"

        # missing membership → reject (was waved through before the fix)
        assert _rejects(None) is True
        # suspended/invited/disabled membership → reject (pre-existing)
        assert _rejects("suspended") is True
        assert _rejects("invited") is True
        # active membership → allow
        assert _rejects("active") is False


@pytest.mark.asyncio
class TestF23RestoreRequiresSystemAdmin:
    """F23 (Astra P1): a full-database restore is a GLOBAL operations
    action — a tenant admin (automation:admin) must not be able to
    trigger it. Route introspection pins the dependency wiring.
    """

    def _find_restore_route(self):
        """Find the POST /{backup_id}/restore route on the backups router."""
        from app.routes.backups import router

        for route in router.routes:
            if "restore" in getattr(route, "path", ""):
                return route
        return None

    async def test_restore_route_uses_require_admin(self):
        from app.deps import require_admin

        route = self._find_restore_route()
        assert route is not None, "restore route not found"
        deps = list(getattr(route, "dependencies", []))
        assert any(getattr(d, "dependency", None) is require_admin for d in deps), (
            "F23: restore route must depend on require_admin (system admin), "
            "not automation:admin"
        )

    async def test_restore_route_not_tenant_admin(self):
        from app.deps import require_permission

        route = self._find_restore_route()
        assert route is not None
        deps = list(getattr(route, "dependencies", []))
        for d in deps:
            dep_fn = getattr(d, "dependency", None)
            assert dep_fn is not require_permission, (
                "F23: restore route must not use require_permission("
                "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