Use core auth and scope contracts

This commit is contained in:
2026-07-10 17:33:43 +02:00
parent 04681f1d75
commit c71de86a1f
6 changed files with 85 additions and 51 deletions

View File

@@ -8,6 +8,8 @@ from sqlalchemy.orm import Session
from govoplan_core.core.access import (
CAPABILITY_ACCESS_PERMISSION_EVALUATOR,
CAPABILITY_ACCESS_PRINCIPAL_RESOLVER,
CAPABILITY_AUTH_PERMISSION_EVALUATOR,
CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
PermissionEvaluator,
PrincipalRef,
PrincipalResolver,
@@ -16,7 +18,7 @@ from govoplan_core.core.modules import AccessDecision
from govoplan_core.core.registry import PlatformRegistry
from govoplan_core.core.maintenance import MAINTENANCE_ACCESS_SCOPE, maintenance_response_detail, saved_maintenance_mode
from govoplan_core.db.session import get_database, get_session
from govoplan_access.backend.db.models import Account, ApiKey, AuthSession, User
from govoplan_access.backend.db.models import Account, ApiKey, AuthSession, Tenant, User
from govoplan_access.backend.semantic import collect_function_assignment_ids, collect_function_delegation_ids, identity_id_for_account
from govoplan_access.backend.security.api_keys import authenticate_api_key
from govoplan_access.backend.security.sessions import (
@@ -26,7 +28,6 @@ from govoplan_access.backend.security.sessions import (
collect_user_scopes,
verify_auth_session_csrf,
)
from govoplan_tenancy.backend.db.models import Tenant
from govoplan_core.security.module_permissions import scopes_grant_compatible
from govoplan_core.security.permissions import intersect_api_key_scopes
from govoplan_core.settings import settings
@@ -278,21 +279,35 @@ def _registry_from_request(request: Request) -> PlatformRegistry | None:
def _principal_resolver_from_request(request: Request) -> PrincipalResolver | None:
registry = _registry_from_request(request)
if registry is None or not registry.has_capability(CAPABILITY_ACCESS_PRINCIPAL_RESOLVER):
if registry is None:
return None
capability = registry.require_capability(CAPABILITY_ACCESS_PRINCIPAL_RESOLVER)
capability_name = (
CAPABILITY_AUTH_PRINCIPAL_RESOLVER
if registry.has_capability(CAPABILITY_AUTH_PRINCIPAL_RESOLVER)
else CAPABILITY_ACCESS_PRINCIPAL_RESOLVER
)
if not registry.has_capability(capability_name):
return None
capability = registry.require_capability(capability_name)
if not isinstance(capability, PrincipalResolver):
raise HTTPException(status_code=500, detail=f"Invalid capability: {CAPABILITY_ACCESS_PRINCIPAL_RESOLVER}")
raise HTTPException(status_code=500, detail=f"Invalid capability: {capability_name}")
return capability
def _permission_evaluator_from_request(request: Request) -> PermissionEvaluator | None:
registry = _registry_from_request(request)
if registry is None or not registry.has_capability(CAPABILITY_ACCESS_PERMISSION_EVALUATOR):
if registry is None:
return None
capability = registry.require_capability(CAPABILITY_ACCESS_PERMISSION_EVALUATOR)
capability_name = (
CAPABILITY_AUTH_PERMISSION_EVALUATOR
if registry.has_capability(CAPABILITY_AUTH_PERMISSION_EVALUATOR)
else CAPABILITY_ACCESS_PERMISSION_EVALUATOR
)
if not registry.has_capability(capability_name):
return None
capability = registry.require_capability(capability_name)
if not isinstance(capability, PermissionEvaluator):
raise HTTPException(status_code=500, detail=f"Invalid capability: {CAPABILITY_ACCESS_PERMISSION_EVALUATOR}")
raise HTTPException(status_code=500, detail=f"Invalid capability: {capability_name}")
return capability