Create access module package
This commit is contained in:
2
src/govoplan_access/backend/permissions/__init__.py
Normal file
2
src/govoplan_access/backend/permissions/__init__.py
Normal file
@@ -0,0 +1,2 @@
|
||||
"""Permission definitions, evaluation, and registry helpers."""
|
||||
|
||||
6
src/govoplan_access/backend/permissions/definitions.py
Normal file
6
src/govoplan_access/backend/permissions/definitions.py
Normal file
@@ -0,0 +1,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from govoplan_core.core.modules import PermissionDefinition, PermissionLevel, RoleTemplate
|
||||
|
||||
__all__ = ["PermissionDefinition", "PermissionLevel", "RoleTemplate"]
|
||||
|
||||
52
src/govoplan_access/backend/permissions/evaluator.py
Normal file
52
src/govoplan_access/backend/permissions/evaluator.py
Normal file
@@ -0,0 +1,52 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Iterable, Mapping
|
||||
|
||||
from govoplan_core.core.modules import PermissionDefinition
|
||||
|
||||
|
||||
def scope_grants(granted: str, required: str, *, catalog: Mapping[str, PermissionDefinition] | None = None) -> bool:
|
||||
if granted == required or granted == "*":
|
||||
return True
|
||||
if granted == "tenant:*":
|
||||
if catalog is None:
|
||||
return not required.startswith("system:")
|
||||
definition = catalog.get(required)
|
||||
return definition is not None and definition.level == "tenant"
|
||||
if granted == "system:*":
|
||||
if catalog is None:
|
||||
return required.startswith("system:")
|
||||
definition = catalog.get(required)
|
||||
return definition is not None and definition.level == "system"
|
||||
if granted.endswith(":*"):
|
||||
return required.startswith(granted[:-1])
|
||||
return False
|
||||
|
||||
|
||||
def scopes_grant(
|
||||
scopes: Iterable[str],
|
||||
required: str,
|
||||
*,
|
||||
catalog: Mapping[str, PermissionDefinition] | None = None,
|
||||
) -> bool:
|
||||
return any(scope_grants(scope, required, catalog=catalog) for scope in scopes)
|
||||
|
||||
|
||||
def expand_scopes(
|
||||
scopes: Iterable[str],
|
||||
*,
|
||||
catalog: Mapping[str, PermissionDefinition],
|
||||
include_unknown: bool = False,
|
||||
) -> list[str]:
|
||||
expanded: set[str] = set()
|
||||
for scope in {str(scope) for scope in scopes if scope}:
|
||||
matched = {candidate for candidate in catalog if scope_grants(scope, candidate, catalog=catalog)}
|
||||
if matched:
|
||||
expanded.update(matched)
|
||||
if scope.endswith(":*") or scope in {"*", "tenant:*", "system:*"}:
|
||||
expanded.add(scope)
|
||||
continue
|
||||
if include_unknown:
|
||||
expanded.add(scope)
|
||||
return sorted(expanded)
|
||||
|
||||
37
src/govoplan_access/backend/permissions/registry.py
Normal file
37
src/govoplan_access/backend/permissions/registry.py
Normal file
@@ -0,0 +1,37 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Iterable
|
||||
|
||||
from govoplan_access.backend.permissions.evaluator import expand_scopes, scopes_grant
|
||||
from govoplan_core.core.modules import PermissionDefinition
|
||||
|
||||
|
||||
class PermissionRegistry:
|
||||
def __init__(self, permissions: Iterable[PermissionDefinition] = ()) -> None:
|
||||
self._catalog: dict[str, PermissionDefinition] = {}
|
||||
for permission in permissions:
|
||||
self.register(permission)
|
||||
|
||||
@property
|
||||
def catalog(self) -> dict[str, PermissionDefinition]:
|
||||
return dict(self._catalog)
|
||||
|
||||
def register(self, permission: PermissionDefinition) -> None:
|
||||
if permission.scope in self._catalog:
|
||||
raise ValueError(f"Duplicate permission scope: {permission.scope}")
|
||||
self._catalog[permission.scope] = permission
|
||||
|
||||
def has(self, scope: str) -> bool:
|
||||
return scope in self._catalog
|
||||
|
||||
def grants(self, scopes: Iterable[str], required: str) -> bool:
|
||||
return scopes_grant(scopes, required, catalog=self._catalog)
|
||||
|
||||
def expand(self, scopes: Iterable[str], *, include_unknown: bool = False) -> list[str]:
|
||||
return expand_scopes(scopes, catalog=self._catalog, include_unknown=include_unknown)
|
||||
|
||||
def validate_known(self, scopes: Iterable[str]) -> list[str]:
|
||||
unknown = sorted(scope for scope in scopes if scope not in self._catalog and not scope.endswith(":*") and scope not in {"*", "tenant:*", "system:*"})
|
||||
if unknown:
|
||||
raise ValueError("Unknown permission scope(s): " + ", ".join(unknown))
|
||||
return sorted(set(scopes))
|
||||
Reference in New Issue
Block a user