189 lines
8.8 KiB
Python
189 lines
8.8 KiB
Python
# #region Test.McpRbacVisibility [C:4] [TYPE Module] [SEMANTICS test,mcp,rbac,visibility,sc-004,sc-009,tools-list]
|
|
# @BRIEF 050 SC-004: tools/list is RBAC-exact for admin/analyst/viewer fixture principals — each
|
|
# assertion pins the EXACT visible catalog subset, not a superset. 050 SC-009: a mid-flow
|
|
# role change flips the next tools/list and denies cached-catalog calls by name, with the
|
|
# same identity-only token (no new consent).
|
|
# @RELATION VERIFIES -> [McpServer.RbacServer]
|
|
# @TEST_INVARIANT rbac_exact_catalog -> VERIFIED_BY: [admin_full_catalog, analyst_exact_subset, viewer_exact_subset]
|
|
# @TEST_INVARIANT live_role_change -> VERIFIED_BY: [role_change_flips_visibility_and_denies_cached_call]
|
|
# @TEST_EDGE revoked_permission -> tool hidden from list AND denied when called by name from a stale catalog.
|
|
# @TEST_EDGE granted_permission -> tool appears in the next list without re-authentication.
|
|
|
|
import os
|
|
import secrets
|
|
|
|
os.environ.setdefault("AUTH_SECRET_KEY", "test-secret-key-for-mcp-rbac")
|
|
os.environ.setdefault("DATABASE_URL", "sqlite:////tmp/ss_tools_mcp_rbac_visibility_test.db")
|
|
|
|
from typing import Any # noqa: E402
|
|
|
|
import pytest # noqa: E402
|
|
|
|
from src.core.auth.security import get_password_hash # noqa: E402
|
|
from src.core.database import SessionLocal # noqa: E402
|
|
from src.mcp_server import server as mcp_server # noqa: E402
|
|
from src.mcp_server.server import _access_token_context, _MCP_CATALOG, _MCP_CATALOG_BY_NAME # noqa: E402
|
|
from src.models.auth import McpToolInvocationRecord, Permission, Role, User # noqa: E402
|
|
|
|
# Exact expected subsets derived from the frozen catalog declaration (McpServer.Catalog):
|
|
# tools with permission=None are visible to any authenticated human principal.
|
|
_PERMISSION_NONE_TOOLS = sorted(
|
|
definition.name for definition in _MCP_CATALOG if definition.permission is None
|
|
)
|
|
|
|
_ANALYST_GRANTS = (("scenario", "RUN"), ("dashboard:testing", "WRITE"), ("tasks", "READ"))
|
|
|
|
_ANALYST_EXTRA_TOOLS = sorted([
|
|
"get_task_status", # tasks:READ
|
|
"list_checkpoints", # scenario:RUN
|
|
"decide_checkpoint", # scenario:RUN
|
|
"start_scenario_run", # scenario:RUN (RUN or RUN_PROD)
|
|
"capture_baseline_candidate", # dashboard:testing:WRITE
|
|
"create_verification_run", # dashboard:testing:WRITE
|
|
"register_draft_pack", # dashboard:testing:WRITE
|
|
])
|
|
|
|
def _persist_principal(prefix: str, *, is_admin: bool, grants: tuple[tuple[str, str], ...]) -> str:
|
|
suffix = secrets.token_hex(4)
|
|
username = f"{prefix}-{suffix}"
|
|
role = Role(
|
|
name=f"{prefix.title()}-{suffix}",
|
|
is_admin=is_admin,
|
|
permissions=[Permission(resource=resource, action=action) for resource, action in grants],
|
|
)
|
|
user = User(username=username, password_hash=get_password_hash("pw"), is_active=True, roles=[role])
|
|
with SessionLocal() as db:
|
|
db.add_all([user])
|
|
db.commit()
|
|
return username
|
|
|
|
|
|
def _cleanup_principal(username: str) -> None:
|
|
with SessionLocal() as db:
|
|
db.query(McpToolInvocationRecord).filter(McpToolInvocationRecord.subject == username).delete()
|
|
user = db.query(User).filter(User.username == username).first()
|
|
if user is not None:
|
|
role_names = [role.name for role in user.roles]
|
|
db.delete(user)
|
|
db.flush()
|
|
for role_name in role_names:
|
|
role_row = db.query(Role).filter(Role.name == role_name).first()
|
|
if role_row is not None:
|
|
db.delete(role_row)
|
|
db.commit()
|
|
|
|
|
|
def _token_for(username: str):
|
|
return mcp_server.AccessToken(
|
|
token=f"rbac-token-{username}",
|
|
client_id="rbac-client",
|
|
scopes=["mcp"],
|
|
subject=username,
|
|
claims={"principal_type": "user"},
|
|
)
|
|
|
|
|
|
async def _visible_names(server: Any) -> list[str]:
|
|
return sorted(tool.name for tool in await server.list_tools())
|
|
|
|
|
|
# #region Test.McpRbacVisibility.ExactCatalog [C:4] [TYPE Function] [SEMANTICS test,mcp,rbac,catalog,sc-004]
|
|
# @ingroup Test.McpRbacVisibility
|
|
# @BRIEF SC-004: admin sees the full explicit catalog; analyst and viewer see EXACT role subsets.
|
|
@pytest.mark.asyncio
|
|
async def test_tools_list_is_rbac_exact_for_admin_analyst_viewer() -> None:
|
|
admin = _persist_principal("rbac-admin", is_admin=True, grants=())
|
|
analyst = _persist_principal("rbac-analyst", is_admin=False, grants=_ANALYST_GRANTS)
|
|
viewer = _persist_principal("rbac-viewer", is_admin=False, grants=())
|
|
try:
|
|
server = mcp_server._build_probe_server()
|
|
|
|
token = _access_token_context.set(_token_for(admin))
|
|
try:
|
|
assert await _visible_names(server) == sorted(d.name for d in _MCP_CATALOG)
|
|
finally:
|
|
_access_token_context.reset(token)
|
|
|
|
token = _access_token_context.set(_token_for(viewer))
|
|
try:
|
|
# permission=None tools only — no permission-bearing tool leaks to a zero-grant human.
|
|
assert await _visible_names(server) == _PERMISSION_NONE_TOOLS
|
|
finally:
|
|
_access_token_context.reset(token)
|
|
|
|
token = _access_token_context.set(_token_for(analyst))
|
|
try:
|
|
assert await _visible_names(server) == sorted(_PERMISSION_NONE_TOOLS + _ANALYST_EXTRA_TOOLS)
|
|
finally:
|
|
_access_token_context.reset(token)
|
|
finally:
|
|
for username in (admin, analyst, viewer):
|
|
_cleanup_principal(username)
|
|
# #endregion Test.McpRbacVisibility.ExactCatalog
|
|
|
|
|
|
# #region Test.McpRbacVisibility.RoleChange [C:4] [TYPE Function] [SEMANTICS test,mcp,rbac,role-change,sc-009]
|
|
# @ingroup Test.McpRbacVisibility
|
|
# @BRIEF SC-009: revoking a grant mid-flow hides the tool in the next tools/list and denies a
|
|
# cached-catalog call by name; granting a new one exposes it — all on the SAME token.
|
|
@pytest.mark.asyncio
|
|
async def test_role_change_flips_visibility_and_denies_cached_call() -> None:
|
|
username = _persist_principal("rbac-flow", is_admin=False, grants=_ANALYST_GRANTS)
|
|
try:
|
|
server = mcp_server._build_probe_server()
|
|
token = _access_token_context.set(_token_for(username))
|
|
try:
|
|
assert "decide_checkpoint" in await _visible_names(server)
|
|
|
|
# Revoke scenario:RUN mid-flow (same token, no re-authentication / new consent).
|
|
with SessionLocal() as db:
|
|
user = db.query(User).filter(User.username == username).first()
|
|
role = user.roles[0]
|
|
role.permissions = [
|
|
Permission(resource=r, action=a)
|
|
for (r, a) in _ANALYST_GRANTS
|
|
if (r, a) != ("scenario", "RUN")
|
|
]
|
|
db.commit()
|
|
|
|
visible_after_revoke = await _visible_names(server)
|
|
assert "decide_checkpoint" not in visible_after_revoke
|
|
assert "list_checkpoints" not in visible_after_revoke
|
|
assert "start_scenario_run" not in visible_after_revoke
|
|
# tasks/dashboard grants are untouched — visibility stays for them.
|
|
assert "get_task_status" in visible_after_revoke
|
|
|
|
# A stale client catalog calling by name is denied, not silently executed.
|
|
with pytest.raises(PermissionError, match="permission_denied"):
|
|
await server.call_tool("decide_checkpoint", {"run_id": "run-x", "disposition": "confirmed"})
|
|
|
|
# Grant scenario:RUN_PROD mid-flow -> approvals surface appears without a new token.
|
|
with SessionLocal() as db:
|
|
user = db.query(User).filter(User.username == username).first()
|
|
role = user.roles[0]
|
|
role.permissions = list(role.permissions) + [Permission(resource="scenario", action="RUN_PROD")]
|
|
db.commit()
|
|
|
|
visible_after_grant = await _visible_names(server)
|
|
assert "list_pending_approvals" in visible_after_grant
|
|
assert "decide_approval" in visible_after_grant
|
|
assert "start_scenario_run" in visible_after_grant # RUN or RUN_PROD
|
|
finally:
|
|
_access_token_context.reset(token)
|
|
finally:
|
|
_cleanup_principal(username)
|
|
# #endregion Test.McpRbacVisibility.RoleChange
|
|
|
|
|
|
# #region Test.McpRbacVisibility.CatalogCoverage [C:3] [TYPE Function] [SEMANTICS test,mcp,rbac,catalog,coverage]
|
|
# @ingroup Test.McpRbacVisibility
|
|
# @BRIEF The exact-subset fixtures must stay derived from the live catalog: every registered tool
|
|
# has a catalog definition, so no unregistered shadow tool can escape the visibility pins.
|
|
def test_every_catalog_entry_matches_registered_surface() -> None:
|
|
assert len(_MCP_CATALOG_BY_NAME) == len(_MCP_CATALOG)
|
|
assert _MCP_CATALOG_BY_NAME["decide_checkpoint"].permission == ("scenario", "RUN")
|
|
assert _MCP_CATALOG_BY_NAME["superset_execute_sql"].service_allowed is False
|
|
# #endregion Test.McpRbacVisibility.CatalogCoverage
|
|
|
|
# #endregion Test.McpRbacVisibility
|