from __future__ import annotations from collections.abc import Iterable, Mapping from govoplan_access.backend.permissions.evaluator import scope_grants as _scope_grants from govoplan_core.core.modules import PermissionDefinition, PermissionLevel, RoleTemplate from govoplan_core.security.module_permissions import compatible_required_scopes from govoplan_core.security.permissions import ALL_PERMISSIONS as CORE_LEGACY_PERMISSION_DEFINITIONS from govoplan_core.security.permissions import PermissionDefinition as CoreLegacyPermissionDefinition from govoplan_core.security.scope_aliases import LEGACY_SCOPE_ALIASES def _legacy_permission(permission: CoreLegacyPermissionDefinition) -> PermissionDefinition: parts = permission.scope.split(":", 2) if len(parts) == 2: module_id, action = parts resource = module_id else: module_id, resource, action = parts return PermissionDefinition( scope=permission.scope, label=permission.label, description=permission.description, category=permission.category, level=permission.level, # type: ignore[arg-type] module_id=module_id, resource=resource, action=action, deprecated=True, ) LEGACY_PERMISSION_DEFINITIONS: tuple[PermissionDefinition, ...] = tuple( _legacy_permission(permission) for permission in CORE_LEGACY_PERMISSION_DEFINITIONS ) def normalize_email(value: str) -> str: return value.strip().casefold() def permission_catalog(*, include_legacy: bool = True) -> tuple[PermissionDefinition, ...]: catalog: dict[str, PermissionDefinition] = {} for permission in _active_permission_definitions(): catalog[permission.scope] = permission if include_legacy: for permission in LEGACY_PERMISSION_DEFINITIONS: catalog.setdefault(permission.scope, permission) return tuple(catalog.values()) def permission_map(*, include_legacy: bool = True) -> dict[str, PermissionDefinition]: return {permission.scope: permission for permission in permission_catalog(include_legacy=include_legacy)} def role_templates() -> tuple[RoleTemplate, ...]: registry = _registry() if registry is not None and hasattr(registry, "role_templates"): return tuple(registry.role_templates()) from govoplan_access.backend.manifest import ACCESS_ROLE_TEMPLATES return ACCESS_ROLE_TEMPLATES def role_templates_for_level(level: PermissionLevel) -> tuple[RoleTemplate, ...]: return tuple(template for template in role_templates() if template.level == level) def scope_grants(granted: str, required: str, *, catalog: Mapping[str, PermissionDefinition] | None = None) -> bool: catalog = catalog if catalog is not None else permission_map(include_legacy=True) if _scope_grants(granted, required, catalog=catalog): return True for alias in LEGACY_SCOPE_ALIASES.get(granted, frozenset()): if scope_grants(alias, required, catalog=catalog): return True return any( _scope_grants(granted, alias, catalog=catalog) for alias in compatible_required_scopes(required) if alias != required ) def scopes_grant(scopes: Iterable[str], required: str, *, catalog: Mapping[str, PermissionDefinition] | None = None) -> bool: catalog = catalog if catalog is not None else permission_map(include_legacy=True) return any(scope_grants(scope, required, catalog=catalog) for scope in scopes) def expand_scopes(scopes: Iterable[str], *, include_unknown: bool = True) -> list[str]: catalog = permission_map(include_legacy=True) raw = {str(scope) for scope in scopes if scope} expanded: set[str] = set() for scope in raw: if scope in {"*", "tenant:*", "system:*"} or scope.endswith(":*"): expanded.add(scope) for alias in LEGACY_SCOPE_ALIASES.get(scope, frozenset()): expanded.add(alias) matched = {candidate for candidate in catalog if scope_grants(scope, candidate, catalog=catalog)} expanded.update(matched) for alias in compatible_required_scopes(scope): if alias != scope: expanded.add(alias) # Preserve explicitly granted scopes in the presentation set. A # canonical module scope can still match its legacy compatibility # alias when the providing module is not loaded; treating that match # as proof that the original scope is known would otherwise replace # the canonical grant with only the deprecated alias. if include_unknown or scope in catalog: expanded.add(scope) return sorted(expanded) def effective_permission_scopes(scopes: Iterable[str], *, level: PermissionLevel | None = None) -> set[str]: candidates = _effective_permission_candidates(level=level) granted = list(scopes) catalog = permission_map(include_legacy=True) return {scope for scope in candidates if scopes_grant(granted, scope, catalog=catalog)} def effective_permission_count(scopes: Iterable[str], *, level: PermissionLevel | None = None) -> int: return len(effective_permission_scopes(scopes, level=level)) def _effective_permission_candidates(*, level: PermissionLevel | None = None) -> set[str]: active_catalog = permission_map(include_legacy=False) full_catalog = permission_map(include_legacy=True) active_scopes = { scope for scope, definition in active_catalog.items() if level is None or definition.level == level } candidates = set(active_scopes) for scope, definition in full_catalog.items(): if scope in active_catalog: continue if level is not None and definition.level != level: continue if any( scope_grants(active_scope, scope, catalog=full_catalog) or scope_grants(scope, active_scope, catalog=full_catalog) for active_scope in active_scopes ): continue candidates.add(scope) return candidates def validate_permissions(scopes: Iterable[str], *, level: PermissionLevel) -> list[str]: normalized = {str(scope) for scope in scopes if scope} wildcard = "system:*" if level == "system" else "tenant:*" if "*" in normalized or wildcard in normalized: return [wildcard] catalog = permission_map(include_legacy=True) expanded: set[str] = set() invalid: list[str] = [] for scope in sorted(normalized): aliases = LEGACY_SCOPE_ALIASES.get(scope, frozenset()) if aliases: for alias in aliases: definition = catalog.get(alias) if definition is not None and definition.level == level: expanded.add(alias) continue if scope.endswith(":*"): matching = {candidate for candidate, definition in catalog.items() if definition.level == level and scope_grants(scope, candidate, catalog=catalog)} if matching: expanded.add(scope) continue definition = catalog.get(scope) if definition is not None and definition.level == level: expanded.add(scope) continue invalid.append(scope) if invalid: raise ValueError(f"Unsupported {level} permissions: {', '.join(invalid)}") return sorted(expanded) def validate_tenant_permissions(scopes: Iterable[str]) -> list[str]: return validate_permissions(scopes, level="tenant") def validate_system_permissions(scopes: Iterable[str]) -> list[str]: return validate_permissions(scopes, level="system") def delegateable_scopes(scopes: Iterable[str], *, level: PermissionLevel) -> set[str]: expanded = set(expand_scopes(scopes, include_unknown=False)) wildcard = "system:*" if level == "system" else "tenant:*" if "*" in expanded or wildcard in expanded: return {scope for scope, definition in permission_map(include_legacy=True).items() if definition.level == level} return effective_permission_scopes(expanded, level=level) def delegateable_tenant_scopes(scopes: Iterable[str]) -> set[str]: return delegateable_scopes(scopes, level="tenant") def delegateable_system_scopes(scopes: Iterable[str]) -> set[str]: return delegateable_scopes(scopes, level="system") def intersect_api_key_scopes(user_scopes: Iterable[str], key_scopes: Iterable[str]) -> list[str]: user = list(user_scopes) key = list(key_scopes) catalog = permission_map(include_legacy=True) tenant_scopes = {scope for scope, definition in catalog.items() if definition.level == "tenant"} allowed = {scope for scope in tenant_scopes if scopes_grant(user, scope, catalog=catalog) and scopes_grant(key, scope, catalog=catalog)} user_raw = set(expand_scopes(user)) key_raw = set(expand_scopes(key)) allowed.update( scope for scope in user_raw.intersection(key_raw) if _is_concrete_tenant_credential_scope(scope, catalog) ) return sorted(allowed) def _is_concrete_tenant_credential_scope( scope: str, catalog: Mapping[str, PermissionDefinition], ) -> bool: # Wildcards are expanded against the tenant catalogue above. Returning the # wildcard itself could grant system permissions sharing the module prefix, # or permissions outside the currently known tenant catalogue. if scope == "*" or scope.endswith(":*"): return False # System permissions can use module-native names (e.g. access:tenant:create), # so excluding only the historical system: prefix is not sufficient. return all( not alias.startswith("system:") and (alias not in catalog or catalog[alias].level == "tenant") for alias in compatible_required_scopes(scope) ) def _active_permission_definitions() -> tuple[PermissionDefinition, ...]: registry = _registry() if registry is not None and hasattr(registry, "permissions"): return tuple(registry.permissions()) from govoplan_access.backend.manifest import ACCESS_PERMISSIONS return ACCESS_PERMISSIONS def _registry() -> object | None: from govoplan_access.backend.runtime import get_registry return get_registry()