diff --git a/docs/ENTERPRISE_SECURITY.md b/docs/ENTERPRISE_SECURITY.md index 5d7a784a..515a3ce6 100644 --- a/docs/ENTERPRISE_SECURITY.md +++ b/docs/ENTERPRISE_SECURITY.md @@ -137,6 +137,27 @@ plus a database cross-check; HS256/none are rejected. > This also fixes a latent multi-replica bug where a per-process random secret > made tokens unverifiable across manager replicas/restarts. +### Scope-based authorization (roles are scope bundles) + +Authorization decisions are made on **scopes only** — never role names. A user's +role is a convenience bundle that the manager **expands into concrete +`resource:action` scopes at token issuance**, emitting them in the token's +`scope` claim. Middleware checks the `scope` claim; `global_role`/`team_roles` +remain on the token for audit and per-team membership checks. + +| Global role | Scope bundle (summary) | +|--------------|-------------------------------------------------------------------| +| SystemAdmin | every `*:write` / `*:admin` scope + `admin:super` + all `*:read` | +| OrgAdmin | `servers:write` `teams:write` `time:write` `dhcp:write` + reads | +| UserManager | `users:write` + reads | +| Viewer | `*:read` only | + +Endpoints declare the scope they need (e.g. `@requires_scope('servers:write')`); +the super-admin bypass for team/zone access checks the `admin:super` scope, not +the `SystemAdmin` role name. Bundle definitions live in one place +(`app/services/scopes.py`), so entitlements are auditable and adjustable +centrally. + --- ## 2. Tenant isolation diff --git a/manager/backend/app/blueprints/dhcp.py b/manager/backend/app/blueprints/dhcp.py index 7c65cd53..8d5e8d3e 100644 --- a/manager/backend/app/blueprints/dhcp.py +++ b/manager/backend/app/blueprints/dhcp.py @@ -5,7 +5,7 @@ from flask import Blueprint, request, jsonify, current_app from app.middleware.auth import token_required, get_current_user -from app.middleware.rbac import requires_role, check_team_access +from app.middleware.rbac import requires_scope, check_team_access from app.utils.decorators import validate_json, audit_log from datetime import datetime @@ -96,7 +96,7 @@ def list_dhcp_pools(): @dhcp_bp.route('/api/v1/dhcp/pools', methods=['POST']) @token_required -@requires_role('SystemAdmin', 'OrgAdmin') +@requires_scope('dhcp:write') @validate_json('name', 'network', 'rangeStart', 'rangeEnd') @audit_log('dhcp_pool_created') def create_dhcp_pool(): @@ -205,7 +205,7 @@ def get_dhcp_pool(pool_id): @dhcp_bp.route('/api/v1/dhcp/pools/', methods=['PUT']) @token_required -@requires_role('SystemAdmin', 'OrgAdmin') +@requires_scope('dhcp:write') @audit_log('dhcp_pool_updated') def update_dhcp_pool(pool_id): """Update DHCP pool configuration.""" @@ -255,7 +255,7 @@ def update_dhcp_pool(pool_id): @dhcp_bp.route('/api/v1/dhcp/pools/', methods=['DELETE']) @token_required -@requires_role('SystemAdmin') +@requires_scope('dhcp:admin') @audit_log('dhcp_pool_deleted') def delete_dhcp_pool(pool_id): """Delete DHCP pool and all associated leases/reservations.""" @@ -327,7 +327,7 @@ def list_pool_leases(pool_id): @dhcp_bp.route('/api/v1/dhcp/pools//leases/', methods=['DELETE']) @token_required -@requires_role('SystemAdmin', 'OrgAdmin') +@requires_scope('dhcp:write') @audit_log('dhcp_lease_released') def release_lease(pool_id, lease_id): """Manually release a DHCP lease.""" @@ -388,7 +388,7 @@ def list_reservations(): @dhcp_bp.route('/api/v1/dhcp/reservations', methods=['POST']) @token_required -@requires_role('SystemAdmin', 'OrgAdmin') +@requires_scope('dhcp:write') @validate_json('poolId', 'macAddress', 'ipAddress') @audit_log('dhcp_reservation_created') def create_reservation(): @@ -435,7 +435,7 @@ def create_reservation(): @dhcp_bp.route('/api/v1/dhcp/reservations/', methods=['DELETE']) @token_required -@requires_role('SystemAdmin', 'OrgAdmin') +@requires_scope('dhcp:write') @audit_log('dhcp_reservation_deleted') def delete_reservation(reservation_id): """Delete a DHCP reservation.""" diff --git a/manager/backend/app/blueprints/dns_servers.py b/manager/backend/app/blueprints/dns_servers.py index a928d862..77431070 100644 --- a/manager/backend/app/blueprints/dns_servers.py +++ b/manager/backend/app/blueprints/dns_servers.py @@ -8,7 +8,7 @@ from app.services.auth_service import AuthService from app.services.config_service import ConfigService from app.middleware.auth import token_required, server_token_required, get_current_server -from app.middleware.rbac import requires_role +from app.middleware.rbac import requires_scope from app.utils.decorators import validate_json, audit_log dns_servers_bp = Blueprint('dns_servers', __name__) @@ -16,7 +16,7 @@ @dns_servers_bp.route('/api/v1/dns-servers', methods=['GET']) @token_required -@requires_role('SystemAdmin', 'OrgAdmin') +@requires_scope('servers:write') def list_dns_servers(): """ List all DNS servers. @@ -55,7 +55,7 @@ def list_dns_servers(): @dns_servers_bp.route('/api/v1/dns-servers', methods=['POST']) @token_required -@requires_role('SystemAdmin', 'OrgAdmin') +@requires_scope('servers:write') @validate_json('name') @audit_log('dns_server_created') def create_dns_server(): @@ -89,7 +89,7 @@ def create_dns_server(): @dns_servers_bp.route('/api/v1/dns-servers/', methods=['GET']) @token_required -@requires_role('SystemAdmin', 'OrgAdmin') +@requires_scope('servers:write') def get_dns_server(server_id): """Get DNS server details.""" server = JoinKeyService.get_server_by_id(server_id) @@ -102,7 +102,7 @@ def get_dns_server(server_id): @dns_servers_bp.route('/api/v1/dns-servers/', methods=['DELETE']) @token_required -@requires_role('SystemAdmin') +@requires_scope('servers:admin') @audit_log('dns_server_deleted') def delete_dns_server(server_id): """Delete DNS server.""" @@ -123,7 +123,7 @@ def delete_dns_server(server_id): @dns_servers_bp.route('/api/v1/dns-servers//regenerate-key', methods=['POST']) @token_required -@requires_role('SystemAdmin') +@requires_scope('servers:admin') @audit_log('dns_server_key_regenerated') def regenerate_join_key(server_id): """ @@ -308,7 +308,7 @@ def refresh_server_token(server_id): @dns_servers_bp.route('/api/v1/dns-servers//metrics', methods=['GET']) @token_required -@requires_role('SystemAdmin', 'OrgAdmin') +@requires_scope('servers:write') def get_dns_server_metrics(server_id): """ Get historical metrics for DNS server. diff --git a/manager/backend/app/blueprints/ioc_feeds.py b/manager/backend/app/blueprints/ioc_feeds.py index c4f3c7cc..e3a4ff45 100644 --- a/manager/backend/app/blueprints/ioc_feeds.py +++ b/manager/backend/app/blueprints/ioc_feeds.py @@ -8,7 +8,7 @@ from typing import Optional from flask import Blueprint, request, jsonify, current_app from app.middleware.auth import token_required -from app.middleware.rbac import requires_system_admin, requires_role +from app.middleware.rbac import requires_scope from app.utils.decorators import validate_json, audit_log ioc_feeds_bp = Blueprint('ioc_feeds', __name__) @@ -131,7 +131,7 @@ def list_ioc_feeds(): @ioc_feeds_bp.route('/api/v1/ioc-feeds', methods=['POST']) @token_required -@requires_system_admin +@requires_scope('ioc:admin') @validate_json('name', 'url', 'feed_type') @_check_ioc_flag_and_license() @audit_log('ioc_feed_created') @@ -221,7 +221,7 @@ def get_ioc_feed(feed_id): @ioc_feeds_bp.route('/api/v1/ioc-feeds/', methods=['PUT']) @token_required -@requires_system_admin +@requires_scope('ioc:admin') @_check_ioc_flag_and_license() @audit_log('ioc_feed_updated') def update_ioc_feed(feed_id): @@ -271,7 +271,7 @@ def update_ioc_feed(feed_id): @ioc_feeds_bp.route('/api/v1/ioc-feeds/', methods=['DELETE']) @token_required -@requires_system_admin +@requires_scope('ioc:admin') @_check_ioc_flag_and_license() @audit_log('ioc_feed_deleted') def delete_ioc_feed(feed_id): @@ -292,7 +292,7 @@ def delete_ioc_feed(feed_id): @ioc_feeds_bp.route('/api/v1/ioc-feeds//sync', methods=['POST']) @token_required -@requires_system_admin +@requires_scope('ioc:admin') @_check_ioc_flag_and_license() @audit_log('ioc_feed_sync_triggered') def trigger_ioc_feed_sync(feed_id): diff --git a/manager/backend/app/blueprints/teams.py b/manager/backend/app/blueprints/teams.py index 94bb99f3..5a79341d 100644 --- a/manager/backend/app/blueprints/teams.py +++ b/manager/backend/app/blueprints/teams.py @@ -5,7 +5,7 @@ from flask import Blueprint, request, jsonify, current_app from app.middleware.auth import token_required, get_current_user -from app.middleware.rbac import requires_role, requires_team_role, can_access_team, filter_teams_by_access +from app.middleware.rbac import requires_scope, requires_team_role, can_access_team, filter_teams_by_access from app.utils.decorators import validate_json, audit_log from app.utils.validators import validate_team_role @@ -58,7 +58,7 @@ def list_teams(): @teams_bp.route('/api/v1/teams', methods=['POST']) @token_required -@requires_role('SystemAdmin', 'OrgAdmin') +@requires_scope('teams:write') @validate_json('name') @audit_log('team_created') def create_team(): @@ -170,7 +170,7 @@ def update_team(team_id): @teams_bp.route('/api/v1/teams/', methods=['DELETE']) @token_required -@requires_role('SystemAdmin', 'OrgAdmin') +@requires_scope('teams:write') @audit_log('team_deleted') def delete_team(team_id): """Delete team and all memberships.""" diff --git a/manager/backend/app/blueprints/time.py b/manager/backend/app/blueprints/time.py index af99b3a7..08922f5c 100644 --- a/manager/backend/app/blueprints/time.py +++ b/manager/backend/app/blueprints/time.py @@ -5,7 +5,7 @@ from flask import Blueprint, request, jsonify, current_app from app.middleware.auth import token_required, get_current_user -from app.middleware.rbac import requires_role +from app.middleware.rbac import requires_scope from app.utils.decorators import validate_json, audit_log from datetime import datetime, timedelta @@ -88,7 +88,7 @@ def list_time_servers(): @time_bp.route('/api/v1/time/servers', methods=['POST']) @token_required -@requires_role('SystemAdmin', 'OrgAdmin') +@requires_scope('time:write') @validate_json('name', 'serverUrl', 'protocol') @audit_log('time_server_created') def create_time_server(): @@ -203,7 +203,7 @@ def get_time_server(server_id): @time_bp.route('/api/v1/time/servers/', methods=['PUT']) @token_required -@requires_role('SystemAdmin', 'OrgAdmin') +@requires_scope('time:write') @audit_log('time_server_updated') def update_time_server(server_id): """Update time server configuration.""" @@ -236,7 +236,7 @@ def update_time_server(server_id): @time_bp.route('/api/v1/time/servers/', methods=['DELETE']) @token_required -@requires_role('SystemAdmin') +@requires_scope('time:admin') @audit_log('time_server_deleted') def delete_time_server(server_id): """Delete a time server.""" @@ -321,7 +321,7 @@ def get_time_status(): @time_bp.route('/api/v1/time/sync', methods=['POST']) @token_required -@requires_role('SystemAdmin', 'OrgAdmin') +@requires_scope('time:write') @audit_log('time_sync_triggered') def trigger_time_sync(): """ diff --git a/manager/backend/app/blueprints/users.py b/manager/backend/app/blueprints/users.py index fa4715f4..3b498f10 100644 --- a/manager/backend/app/blueprints/users.py +++ b/manager/backend/app/blueprints/users.py @@ -6,7 +6,7 @@ from flask import Blueprint, request, jsonify, current_app from app.services.auth_service import AuthService from app.middleware.auth import token_required -from app.middleware.rbac import requires_role +from app.middleware.rbac import requires_scope from app.utils.decorators import validate_json, audit_log from app.utils.validators import validate_email, validate_username, validate_global_role @@ -15,7 +15,7 @@ @users_bp.route('/api/v1/users', methods=['GET']) @token_required -@requires_role('SystemAdmin', 'UserManager') +@requires_scope('users:write') def list_users(): """ List all users. @@ -67,7 +67,7 @@ def list_users(): @users_bp.route('/api/v1/users', methods=['POST']) @token_required -@requires_role('SystemAdmin', 'UserManager') +@requires_scope('users:write') @validate_json('username', 'email', 'password', 'global_role') @audit_log('user_created') def create_user(): @@ -139,7 +139,7 @@ def create_user(): @users_bp.route('/api/v1/users/', methods=['GET']) @token_required -@requires_role('SystemAdmin', 'UserManager') +@requires_scope('users:write') def get_user(user_id): """Get user by ID.""" db = current_app.db @@ -178,7 +178,7 @@ def get_user(user_id): @users_bp.route('/api/v1/users/', methods=['PUT']) @token_required -@requires_role('SystemAdmin', 'UserManager') +@requires_scope('users:write') @audit_log('user_updated') def update_user(user_id): """ @@ -228,7 +228,7 @@ def update_user(user_id): @users_bp.route('/api/v1/users/', methods=['DELETE']) @token_required -@requires_role('SystemAdmin') +@requires_scope('users:admin') @audit_log('user_deleted') def delete_user(user_id): """ diff --git a/manager/backend/app/middleware/auth.py b/manager/backend/app/middleware/auth.py index bfb9d067..c4bff597 100644 --- a/manager/backend/app/middleware/auth.py +++ b/manager/backend/app/middleware/auth.py @@ -41,6 +41,7 @@ def decorated(*args, **kwargs): 'user_id': payload.get('user_id'), 'server_id': payload.get('server_id'), 'username': payload.get('username'), + 'scope': payload.get('scope', ''), 'global_role': payload.get('global_role'), 'team_roles': payload.get('team_roles', {}), 'token_type': payload.get('type') @@ -132,6 +133,7 @@ def decorated(*args, **kwargs): g.current_user = { 'user_id': payload.get('user_id'), 'username': payload.get('username'), + 'scope': payload.get('scope', ''), 'global_role': payload.get('global_role'), 'team_roles': payload.get('team_roles', {}), 'token_type': payload.get('type') diff --git a/manager/backend/app/middleware/rbac.py b/manager/backend/app/middleware/rbac.py index eed104b3..370822ba 100644 --- a/manager/backend/app/middleware/rbac.py +++ b/manager/backend/app/middleware/rbac.py @@ -6,6 +6,50 @@ from functools import wraps from flask import jsonify, current_app from app.middleware.auth import get_current_user +from app.services.scopes import SUPERADMIN_SCOPE + + +def _user_scopes(user) -> set: + """Return the set of scopes carried by the token (authz source of truth).""" + if not user: + return set() + return set((user.get('scope') or '').split()) + + +def _is_superadmin(user) -> bool: + """True if the token holds the super-admin scope (global bypass privilege).""" + return SUPERADMIN_SCOPE in _user_scopes(user) + + +def requires_scope(*required_scopes): + """ + Decorator to enforce that the caller holds at least one of the given scopes. + + Authorization is scope-based — never role names. Roles are expanded into + concrete scopes at token issuance (see app.services.scopes). + + Example: + @requires_scope('servers:write') + def create_server(): + ... + """ + def decorator(f): + @wraps(f) + def decorated(*args, **kwargs): + user = get_current_user() + if not user: + return jsonify({'error': 'Authentication required'}), 401 + + if not _user_scopes(user).intersection(required_scopes): + return jsonify({ + 'error': 'Insufficient permissions', + 'required_scopes': list(required_scopes), + }), 403 + + return f(*args, **kwargs) + + return decorated + return decorator def requires_role(*allowed_roles): @@ -61,7 +105,7 @@ def decorated(*args, **kwargs): return jsonify({'error': 'Authentication required'}), 401 # System admins bypass team restrictions - if user.get('global_role') == 'SystemAdmin': + if _is_superadmin(user): return f(*args, **kwargs) # Get team_id from kwargs or route parameters @@ -101,7 +145,7 @@ def can_access_team(team_id: int) -> bool: return False # System admins can access all teams - if user.get('global_role') == 'SystemAdmin': + if _is_superadmin(user): return True # Check team membership @@ -120,7 +164,8 @@ def can_manage_users() -> bool: if not user: return False - return user.get('global_role') in ['SystemAdmin', 'UserManager'] + # SystemAdmin (users:write+users:admin) and UserManager (users:write). + return bool(_user_scopes(user).intersection({'users:write', 'users:admin'})) def can_manage_teams() -> bool: @@ -134,7 +179,8 @@ def can_manage_teams() -> bool: if not user: return False - return user.get('global_role') in ['SystemAdmin', 'OrgAdmin'] + # SystemAdmin and OrgAdmin both hold teams:write. + return 'teams:write' in _user_scopes(user) def filter_teams_by_access(): @@ -149,7 +195,7 @@ def filter_teams_by_access(): return [] # System admins can access all teams - if user.get('global_role') == 'SystemAdmin': + if _is_superadmin(user): db = current_app.db teams = db(db.team).select(db.team.id) return [t.id for t in teams] @@ -161,17 +207,18 @@ def filter_teams_by_access(): def requires_system_admin(f): """ - Decorator to require SystemAdmin role. - Shorthand for @requires_role('SystemAdmin'). + Decorator to require super-admin privilege (scope-based). + Only the SystemAdmin bundle carries SUPERADMIN_SCOPE. """ - return requires_role('SystemAdmin')(f) + return requires_scope(SUPERADMIN_SCOPE)(f) def requires_admin(f): """ - Decorator to require admin role (SystemAdmin or OrgAdmin). + Decorator to require org-admin privilege (scope-based). + SystemAdmin and OrgAdmin both hold teams:write. """ - return requires_role('SystemAdmin', 'OrgAdmin')(f) + return requires_scope('teams:write')(f) def check_team_access(team_id: int) -> bool: @@ -189,7 +236,7 @@ def check_team_access(team_id: int) -> bool: return False # System admins can access all teams - if user.get('global_role') == 'SystemAdmin': + if _is_superadmin(user): return True # Check if user is a member of the team @@ -212,7 +259,7 @@ def check_zone_access(zone_id: int) -> bool: return False # System admins can access all zones - if user.get('global_role') == 'SystemAdmin': + if _is_superadmin(user): return True db = current_app.db @@ -247,7 +294,7 @@ def filter_zones_by_access(): return [z.id for z in zones] # System admins can access all zones - if user.get('global_role') == 'SystemAdmin': + if _is_superadmin(user): zones = db(db.dns_zone).select(db.dns_zone.id) return [z.id for z in zones] diff --git a/manager/backend/app/services/auth_service.py b/manager/backend/app/services/auth_service.py index b680f737..34d0cb08 100644 --- a/manager/backend/app/services/auth_service.py +++ b/manager/backend/app/services/auth_service.py @@ -43,6 +43,11 @@ def create_access_token(user_id: int, username: str, global_role: str, # TODO: extract tenant from user.org if schema adds org/tenant column tenant = current_app.config.get('TENANT_ID', 'default') + # Expand the role into its concrete scope bundle at issuance. Authz + # decisions are made on `scope`; global_role/team_roles are retained + # for audit and per-team membership checks only. + from app.services.scopes import scope_string + payload = { 'sub': str(user_id), 'iss': current_app.config['JWT_ISSUER'], @@ -50,6 +55,7 @@ def create_access_token(user_id: int, username: str, global_role: str, 'tenant': tenant, 'user_id': user_id, 'username': username, + 'scope': scope_string(global_role, team_roles), 'global_role': global_role, 'team_roles': team_roles or {}, 'type': 'access', diff --git a/manager/backend/app/services/scopes.py b/manager/backend/app/services/scopes.py new file mode 100644 index 00000000..b02075d3 --- /dev/null +++ b/manager/backend/app/services/scopes.py @@ -0,0 +1,67 @@ +"""Scope-bundle definitions and role→scope expansion. + +Authorization is **scope-based** (security standard): roles are convenience +bundles that are expanded into concrete ``resource:action`` scopes at token +issuance. Middleware checks scopes only — never role names. The ``roles``/ +``global_role`` claims remain on tokens for audit/display only. + +The bundles below reproduce the pre-refactor access matrix exactly: + + SystemAdmin → every write/admin scope + admin:super + all reads + OrgAdmin → servers/teams/time/dhcp :write + all reads + UserManager → users:write + all reads + Viewer → all reads +""" + +from __future__ import annotations + +from typing import Dict, List, Optional + +# Sentinel scope held only by SystemAdmin. Global bypasses that previously +# branched on the 'SystemAdmin' role name (team-membership, zone access) now +# check for this scope instead. +SUPERADMIN_SCOPE = "admin:super" + +# Read scopes granted to every authenticated principal via their role bundle. +# (No endpoint gates on these today; included for forward use + audit clarity.) +_READ_SCOPES: List[str] = [ + "users:read", "servers:read", "teams:read", "time:read", + "dhcp:read", "ioc:read", "zones:read", "tokens:read", + "analytics:read", "whois:read", +] + +# Global role → concrete scope bundle. +ROLE_SCOPES: Dict[str, List[str]] = { + "SystemAdmin": [ + SUPERADMIN_SCOPE, + "users:write", "users:admin", + "servers:write", "servers:admin", + "teams:write", "teams:admin", + "time:write", "time:admin", + "dhcp:write", "dhcp:admin", + "ioc:admin", + ] + _READ_SCOPES, + "OrgAdmin": [ + "servers:write", "teams:write", "time:write", "dhcp:write", + ] + _READ_SCOPES, + "UserManager": [ + "users:write", + ] + _READ_SCOPES, + "Viewer": list(_READ_SCOPES), +} + + +def expand_scopes( + global_role: Optional[str], team_roles: Optional[Dict] = None +) -> List[str]: + """Expand a global role into its concrete scope bundle (sorted, de-duped). + + Team roles are enforced per-resource via team-membership data (not flat + token scopes), so they intentionally contribute no global scopes here. + """ + return sorted(set(ROLE_SCOPES.get(global_role or "", []))) + + +def scope_string(global_role: Optional[str], team_roles: Optional[Dict] = None) -> str: + """Space-delimited scope claim value for a token (RFC 8693 style).""" + return " ".join(expand_scopes(global_role, team_roles)) diff --git a/manager/backend/tests/conftest.py b/manager/backend/tests/conftest.py index e25895eb..fdb7cc5a 100644 --- a/manager/backend/tests/conftest.py +++ b/manager/backend/tests/conftest.py @@ -136,6 +136,8 @@ def _make_token(user_id: int = 1, username: str = "testuser", else: exp = now + timedelta(hours=1) + from app.services.scopes import scope_string + payload = { 'sub': str(user_id), 'iss': issuer, @@ -143,6 +145,7 @@ def _make_token(user_id: int = 1, username: str = "testuser", 'tenant': tenant, 'user_id': user_id, 'username': username, + 'scope': scope_string(global_role, team_roles), 'global_role': global_role, 'team_roles': team_roles or {}, 'type': token_type, diff --git a/manager/backend/tests/test_scope_authz.py b/manager/backend/tests/test_scope_authz.py new file mode 100644 index 00000000..907e8327 --- /dev/null +++ b/manager/backend/tests/test_scope_authz.py @@ -0,0 +1,152 @@ +"""Tests for scope-bundle authorization. + +Verifies that (1) roles expand into the correct concrete scope bundles, +(2) issued access tokens carry a `scope` claim, (3) the `requires_scope` +decorator gates on scopes only, and (4) the pre-refactor access matrix is +preserved (SystemAdmin ⊇ OrgAdmin/UserManager privileges; Viewer read-only). +""" + +from __future__ import annotations + +import jwt +import pytest + +from app.middleware import rbac +from app.services.scopes import ( + ROLE_SCOPES, + SUPERADMIN_SCOPE, + expand_scopes, + scope_string, +) + + +# --------------------------------------------------------------------------- +# Role → scope-bundle expansion +# --------------------------------------------------------------------------- +def test_only_systemadmin_holds_superadmin_scope(): + assert SUPERADMIN_SCOPE in expand_scopes("SystemAdmin") + for role in ("OrgAdmin", "UserManager", "Viewer"): + assert SUPERADMIN_SCOPE not in expand_scopes(role) + + +def test_systemadmin_is_superset_of_all_roles(): + sysadmin = set(expand_scopes("SystemAdmin")) + for role in ("OrgAdmin", "UserManager", "Viewer"): + assert set(expand_scopes(role)).issubset(sysadmin) + + +@pytest.mark.parametrize( + "role,present,absent", + [ + ("OrgAdmin", {"servers:write", "teams:write", "time:write", "dhcp:write"}, + {"users:write", "servers:admin", "ioc:admin", SUPERADMIN_SCOPE}), + ("UserManager", {"users:write"}, + {"users:admin", "servers:write", "dhcp:write", "ioc:admin"}), + ("Viewer", {"users:read", "servers:read"}, + {"users:write", "servers:write", "teams:write", "ioc:admin"}), + ], +) +def test_role_bundles_preserve_access_matrix(role, present, absent): + scopes = set(expand_scopes(role)) + assert present.issubset(scopes) + assert scopes.isdisjoint(absent) + + +def test_unknown_role_gets_no_scopes(): + assert expand_scopes(None) == [] + assert expand_scopes("Nonexistent") == [] + + +def test_scope_string_is_space_delimited_and_sorted(): + s = scope_string("OrgAdmin") + assert s == " ".join(sorted(ROLE_SCOPES["OrgAdmin"])) + + +# --------------------------------------------------------------------------- +# Token issuance carries the scope claim +# --------------------------------------------------------------------------- +def test_issued_access_token_carries_scope_claim(app, jwt_keypair): + from app.services.auth_service import AuthService + + with app.app_context(): + token = AuthService.create_access_token( + user_id=7, username="admin", global_role="SystemAdmin" + ) + payload = jwt.decode( + token, jwt_keypair["public"], algorithms=["ES256"], + audience="squawk", issuer="squawk-manager", + ) + assert "scope" in payload + scopes = set(payload["scope"].split()) + assert scopes == set(expand_scopes("SystemAdmin")) + # global_role retained for audit only. + assert payload["global_role"] == "SystemAdmin" + + +# --------------------------------------------------------------------------- +# requires_scope decorator behavior +# --------------------------------------------------------------------------- +def _decorate(scopes): + @rbac.requires_scope(*scopes) + def view(): + return "ok", 200 + return view + + +def test_requires_scope_401_when_unauthenticated(app, monkeypatch): + monkeypatch.setattr(rbac, "get_current_user", lambda: None) + with app.test_request_context(): + _body, status = _decorate(["servers:write"])() + assert status == 401 + + +def test_requires_scope_403_when_scope_missing(app, monkeypatch): + monkeypatch.setattr(rbac, "get_current_user", lambda: {"scope": "servers:read"}) + with app.test_request_context(): + _body, status = _decorate(["servers:write"])() + assert status == 403 + + +def test_requires_scope_200_when_scope_present(app, monkeypatch): + monkeypatch.setattr( + rbac, "get_current_user", + lambda: {"scope": "servers:read servers:write"}, + ) + with app.test_request_context(): + body, status = _decorate(["servers:write"])() + assert (body, status) == ("ok", 200) + + +def test_requires_scope_any_of_semantics(app, monkeypatch): + """Endpoint grants access if the user holds ANY one of the listed scopes.""" + monkeypatch.setattr(rbac, "get_current_user", lambda: {"scope": "dhcp:admin"}) + with app.test_request_context(): + body, status = _decorate(["dhcp:write", "dhcp:admin"])() + assert status == 200 + + +# --------------------------------------------------------------------------- +# Scope-based helpers (no role-name branching) +# --------------------------------------------------------------------------- +def test_is_superadmin_only_for_superadmin_scope(): + assert rbac._is_superadmin({"scope": scope_string("SystemAdmin")}) is True + assert rbac._is_superadmin({"scope": scope_string("OrgAdmin")}) is False + assert rbac._is_superadmin(None) is False + + +def test_can_manage_users_matches_old_matrix(app, monkeypatch): + for role, expected in [("SystemAdmin", True), ("UserManager", True), + ("OrgAdmin", False), ("Viewer", False)]: + monkeypatch.setattr( + rbac, "get_current_user", lambda r=role: {"scope": scope_string(r)} + ) + assert rbac.can_manage_users() is expected + + +def test_can_manage_teams_matches_old_matrix(app, monkeypatch): + for role, expected in [("SystemAdmin", True), ("OrgAdmin", True), + ("UserManager", False), ("Viewer", False)]: + monkeypatch.setattr( + rbac, "get_current_user", lambda r=role: {"scope": scope_string(r)} + ) + assert rbac.can_manage_teams() is expected