ITWorx_MCP_Hub-Public/tests/test_wp230_authz.py

"""WP-230A regression tests: the management-API permission matrix itself.

Pure unit tests against services/control_api/authz.py, with no FastAPI app
involved -- see tests/test_wp230_rbac.py for the real-HTTP, real-routing
counterpart that proves each route actually invokes the action these tests
assume.
"""

from __future__ import annotations

import logging
from collections.abc import Iterator
from datetime import UTC, datetime

import pytest

from services.control_api.auth.models import AuthenticatedPrincipal, ManagementRole, PrincipalKind
from services.control_api.authz import (
    ENDPOINT_PERMISSIONS,
    InsufficientRole,
    check_role,
)

_NOW = datetime.now(UTC)
_AUTHZ_LOGGER = logging.getLogger("itworx_mcp_hub.management_authz")


@pytest.fixture(autouse=True)
def _ensure_authz_logger_enabled() -> Iterator[None]:
    """tests/test_migrations.py imports alembic's env.py, which calls
    `logging.config.fileConfig()`; that function's default
    `disable_existing_loggers=True` silently disables every logger not
    explicitly listed in alembic.ini, including this one (it already exists
    by the time fileConfig runs, if test_migrations.py happens to execute
    first in the suite). Reset defensively rather than depending on
    suite-wide test file ordering to avoid tripping this."""
    previous = _AUTHZ_LOGGER.disabled
    _AUTHZ_LOGGER.disabled = False
    yield
    _AUTHZ_LOGGER.disabled = previous


def _principal(*roles: ManagementRole) -> AuthenticatedPrincipal:
    return AuthenticatedPrincipal(
        subject="test-subject",
        issuer="https://fake.oidc.test",
        audience=("itworx-mcp-hub",),
        kind=PrincipalKind.MANAGEMENT_USER,
        expires_at=_NOW,
        issued_at=_NOW,
        roles=frozenset(roles),
    )


def test_matrix_has_no_action_granting_nothing() -> None:
    for action, allowed in ENDPOINT_PERMISSIONS.items():
        assert allowed, f"{action} grants no role at all -- unreachable by design or a typo"


def test_matrix_never_grants_an_action_to_nobody_but_also_never_to_everyone_implicitly() -> None:
    # Every action's allowed set must be an explicit, real subset of the
    # four defined roles -- catches an accidental typo'd role reference.
    for action, allowed in ENDPOINT_PERMISSIONS.items():
        assert allowed <= set(ManagementRole), action


@pytest.mark.parametrize(
    "action",
    ["dashboard.read", "resource.read", "policy.simulate", "catalog.preview", "client.read"],
)
@pytest.mark.parametrize("role", list(ManagementRole))
def test_baseline_reads_available_to_every_role(action: str, role: ManagementRole) -> None:
    check_role(_principal(role), action)  # must not raise


@pytest.mark.parametrize(
    "role", [ManagementRole.AUDITOR, ManagementRole.OPERATOR, ManagementRole.ADMIN]
)
def test_audit_read_allowed_for_auditor_and_above(role: ManagementRole) -> None:
    check_role(_principal(role), "audit.read")


def test_audit_read_denied_for_viewer_alone() -> None:
    with pytest.raises(InsufficientRole):
        check_role(_principal(ManagementRole.VIEWER), "audit.read")


@pytest.mark.parametrize("role", [ManagementRole.OPERATOR, ManagementRole.ADMIN])
@pytest.mark.parametrize(
    "action",
    [
        "tool.test",
        "tool.invoke",
        "operations.check",
        "client.bootstrap",
        "client.verify",
        "client.disable",
        "client.revoke",
    ],
)
def test_client_lifecycle_actions_allowed_for_operator_and_admin(
    role: ManagementRole, action: str
) -> None:
    check_role(_principal(role), action)


@pytest.mark.parametrize("role", [ManagementRole.VIEWER, ManagementRole.AUDITOR])
@pytest.mark.parametrize(
    "action",
    [
        "tool.test",
        "tool.invoke",
        "operations.check",
        "client.bootstrap",
        "client.verify",
        "client.disable",
        "client.revoke",
    ],
)
def test_client_lifecycle_actions_denied_below_operator(role: ManagementRole, action: str) -> None:
    with pytest.raises(InsufficientRole):
        check_role(_principal(role), action)


@pytest.mark.parametrize("action", ["client.create", "catalog.apply"])
def test_admin_only_actions_allowed_for_admin(action: str) -> None:
    check_role(_principal(ManagementRole.ADMIN), action)


@pytest.mark.parametrize(
    "role", [ManagementRole.VIEWER, ManagementRole.AUDITOR, ManagementRole.OPERATOR]
)
@pytest.mark.parametrize("action", ["client.create", "catalog.apply"])
def test_admin_only_actions_denied_below_admin(role: ManagementRole, action: str) -> None:
    with pytest.raises(InsufficientRole):
        check_role(_principal(role), action)


def test_multiple_roles_grant_via_any_matching_role() -> None:
    # A principal holding both viewer and admin roles can do admin things --
    # roles are additive, the union of what each individual role grants.
    check_role(_principal(ManagementRole.VIEWER, ManagementRole.ADMIN), "catalog.apply")


def test_principal_with_no_roles_is_denied_every_action() -> None:
    principal = _principal()
    for action in ENDPOINT_PERMISSIONS:
        with pytest.raises(InsufficientRole):
            check_role(principal, action)


def test_unknown_action_raises_value_error_not_silently_denied_or_allowed() -> None:
    with pytest.raises(ValueError, match="unknown management action"):
        check_role(_principal(ManagementRole.ADMIN), "not-a-real-action")


def test_denial_is_logged_as_a_warning(caplog: pytest.LogCaptureFixture) -> None:
    with caplog.at_level(logging.WARNING, logger="itworx_mcp_hub.management_authz"):
        with pytest.raises(InsufficientRole):
            check_role(_principal(ManagementRole.VIEWER), "catalog.apply")
    assert any(record.levelno == logging.WARNING for record in caplog.records)
    assert any("management_authz_denied" in record.getMessage() for record in caplog.records)


def test_privileged_allow_is_logged_as_info(caplog: pytest.LogCaptureFixture) -> None:
    with caplog.at_level(logging.INFO, logger="itworx_mcp_hub.management_authz"):
        check_role(_principal(ManagementRole.ADMIN), "catalog.apply")
    assert any(
        record.levelno == logging.INFO and "management_authz_allowed" in record.getMessage()
        for record in caplog.records
    )


def test_baseline_read_allow_is_not_logged_at_info(caplog: pytest.LogCaptureFixture) -> None:
    # Dashboard polling by every role, every few seconds, is not a
    # "privileged action" -- logging every allow here would be pure noise.
    with caplog.at_level(logging.INFO, logger="itworx_mcp_hub.management_authz"):
        check_role(_principal(ManagementRole.VIEWER), "dashboard.read")
    assert not any("management_authz_allowed" in record.getMessage() for record in caplog.records)