feat(access): derive authenticated baseline roles

This commit is contained in:
2026-07-21 20:48:21 +02:00
parent 21f77ffc50
commit 4075ac4d73
4 changed files with 218 additions and 7 deletions

View File

@@ -18,7 +18,13 @@ module directly. Access-side admin service helpers remain
here for users, groups, roles, system accounts, sessions, API keys, tenant here for users, groups, roles, system accounts, sessions, API keys, tenant
access enforcement, admin/audit lookup capabilities, tenant owner access enforcement, admin/audit lookup capabilities, tenant owner
provisioning, and governance-template materialization into access-owned groups provisioning, and governance-template materialization into access-owned groups
and roles. Governance-template metadata CRUD lives in `govoplan-admin`. The and roles. The module also enforces narrowly declared managed
`RoleTemplate.default_authenticated` baselines. Their explicit permissions are
derived from the active manifest set for every authenticated tenant member,
without writing assignments during an authorization read. The optional role
row is a non-assignable administration projection, not the source of the
automatic grant. Domain permissions and resource policy remain separate checks.
Governance-template metadata CRUD lives in `govoplan-admin`. The
transitional administration WebUI route shell and transitional administration WebUI route shell and
access-owned panels live under `webui/src` as `@govoplan/access-webui`. Generic access-owned panels live under `webui/src` as `@govoplan/access-webui`. Generic
system administration panels are contributed by `@govoplan/admin-webui` through system administration panels are contributed by `@govoplan/admin-webui` through

View File

@@ -56,7 +56,7 @@ def ensure_default_roles(session: Session, tenant: Tenant | None = None) -> dict
query = query.filter(Role.tenant_id == tenant.id) if tenant is not None else query.filter(Role.tenant_id.is_(None)) query = query.filter(Role.tenant_id == tenant.id) if tenant is not None else query.filter(Role.tenant_id.is_(None))
role = query.one_or_none() role = query.one_or_none()
is_builtin = _template_is_builtin(template_managed=template.managed, protected=template.protected, tenant_role=tenant is not None) is_builtin = _template_is_builtin(template_managed=template.managed, protected=template.protected, tenant_role=tenant is not None)
is_assignable = True is_assignable = not template.default_authenticated
if role is None: if role is None:
role = Role( role = Role(
tenant_id=tenant.id if tenant is not None else None, tenant_id=tenant.id if tenant is not None else None,
@@ -200,7 +200,24 @@ def set_user_groups(session: Session, *, user: User, group_ids: Iterable[str]) -
def set_user_roles(session: Session, *, user: User, role_ids: Iterable[str]) -> None: def set_user_roles(session: Session, *, user: User, role_ids: Iterable[str]) -> None:
ids = sorted(set(role_ids)) default_slugs = {
template.slug
for template in role_templates_for_level("tenant")
if template.default_authenticated
}
default_roles = (
session.query(Role)
.filter(
Role.tenant_id == user.tenant_id,
Role.slug.in_(default_slugs),
)
.all()
if default_slugs
else []
)
default_role_ids = {role.id for role in default_roles}
requested_ids = set(role_ids) - default_role_ids
ids = sorted(requested_ids)
roles = ( roles = (
session.query(Role) session.query(Role)
.filter(Role.tenant_id == user.tenant_id, Role.id.in_(ids), Role.is_assignable.is_(True)) .filter(Role.tenant_id == user.tenant_id, Role.id.in_(ids), Role.is_assignable.is_(True))

View File

@@ -20,7 +20,7 @@ from govoplan_access.backend.db.models import (
UserRoleAssignment, UserRoleAssignment,
) )
from govoplan_access.backend.semantic import collect_function_authorization_context, collect_function_roles from govoplan_access.backend.semantic import collect_function_authorization_context, collect_function_roles
from govoplan_access.backend.permissions.catalog import expand_scopes from govoplan_access.backend.permissions.catalog import expand_scopes, role_templates_for_level
from govoplan_core.security.time import ensure_aware_utc, utc_now from govoplan_core.security.time import ensure_aware_utc, utc_now
SESSION_RANDOM_BYTES = 32 SESSION_RANDOM_BYTES = 32
@@ -183,6 +183,8 @@ def collect_user_roles(session: Session, user: User) -> list[Role]:
roles_by_id[role.id] = role roles_by_id[role.id] = role
for role in collect_function_roles(session, user): for role in collect_function_roles(session, user):
roles_by_id[role.id] = role roles_by_id[role.id] = role
for role in _materialized_default_authenticated_roles(session, user):
roles_by_id[role.id] = role
return list(roles_by_id.values()) return list(roles_by_id.values())
@@ -242,14 +244,24 @@ def collect_user_authorization_context(
roles_by_id[role.id] = role roles_by_id[role.id] = role
for role in extra_roles: for role in extra_roles:
roles_by_id[role.id] = role roles_by_id[role.id] = role
for role in _materialized_default_authenticated_roles(session, user):
roles_by_id[role.id] = role
tenant_roles = list(roles_by_id.values()) tenant_roles = list(roles_by_id.values())
system_roles = collect_system_roles(session, account) if include_system and account is not None else [] system_roles = collect_system_roles(session, account) if include_system and account is not None else []
default_slugs = _default_authenticated_slugs()
scopes = { scopes = {
scope scope
for role in [*tenant_roles, *system_roles] for role in tenant_roles
if role.slug not in default_slugs
for scope in (role.permissions or []) for scope in (role.permissions or [])
} }
scopes.update(
scope
for role in system_roles
for scope in (role.permissions or [])
)
scopes.update(_default_authenticated_scopes())
return UserAuthorizationContext( return UserAuthorizationContext(
tenant_roles=tenant_roles, tenant_roles=tenant_roles,
system_roles=system_roles, system_roles=system_roles,
@@ -261,8 +273,10 @@ def collect_user_authorization_context(
def collect_user_scopes(session: Session, user: User, *, include_system: bool = True) -> list[str]: def collect_user_scopes(session: Session, user: User, *, include_system: bool = True) -> list[str]:
scopes: set[str] = set() scopes = _default_authenticated_scopes()
default_slugs = _default_authenticated_slugs()
for role in collect_user_roles(session, user): for role in collect_user_roles(session, user):
if role.slug not in default_slugs:
scopes.update(role.permissions or []) scopes.update(role.permissions or [])
if include_system and user.account: if include_system and user.account:
for role in collect_system_roles(session, user.account): for role in collect_system_roles(session, user.account):
@@ -270,6 +284,43 @@ def collect_user_scopes(session: Session, user: User, *, include_system: bool =
return expand_scopes(scopes) return expand_scopes(scopes)
def _default_authenticated_slugs() -> set[str]:
return {
template.slug
for template in role_templates_for_level("tenant")
if template.default_authenticated
}
def _default_authenticated_scopes() -> set[str]:
return {
scope
for template in role_templates_for_level("tenant")
if template.default_authenticated
for scope in template.permissions
}
def _materialized_default_authenticated_roles(
session: Session,
user: User,
) -> list[Role]:
"""Return the optional database projection without mutating auth reads."""
default_slugs = _default_authenticated_slugs()
if not default_slugs:
return []
return (
session.query(Role)
.filter(
Role.tenant_id == user.tenant_id,
Role.slug.in_(default_slugs),
)
.order_by(Role.name.asc(), Role.id.asc())
.all()
)
def collect_tenant_memberships(session: Session, account: Account) -> list[tuple[User, Tenant]]: def collect_tenant_memberships(session: Session, account: Account) -> list[tuple[User, Tenant]]:
return ( return (
session.query(User, Tenant) session.query(User, Tenant)

View File

@@ -0,0 +1,137 @@
from __future__ import annotations
import unittest
from unittest.mock import patch
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from govoplan_access.backend.admin.service import ensure_default_roles, set_user_roles
from govoplan_access.backend.db.models import Account, Role, User, UserRoleAssignment
from govoplan_access.backend.security.sessions import collect_user_authorization_context
from govoplan_core.core.modules import RoleTemplate
from govoplan_core.db.base import Base
from govoplan_core.tenancy.scope import Tenant, create_scope_tables, scope_registry
DOCS_READER = RoleTemplate(
slug="docs_reader",
name="Documentation reader",
description="Authenticated documentation baseline.",
permissions=("docs:documentation:read",),
default_authenticated=True,
)
class DefaultAuthenticatedRoleTests(unittest.TestCase):
def setUp(self) -> None:
self.engine = create_engine("sqlite:///:memory:")
create_scope_tables(self.engine)
Base.metadata.create_all(bind=self.engine)
self.Session = sessionmaker(bind=self.engine)
self.session = self.Session()
self.tenant = Tenant(id="tenant-1", slug="tenant-1", name="Tenant 1")
self.session.add(self.tenant)
self.session.flush()
def tearDown(self) -> None:
self.session.close()
Base.metadata.drop_all(bind=self.engine)
scope_registry.metadata.drop_all(bind=self.engine)
self.engine.dispose()
def _add_user(self, suffix: str) -> User:
account = Account(
id=f"account-{suffix}",
email=f"{suffix}@example.test",
normalized_email=f"{suffix}@example.test",
)
user = User(
id=f"user-{suffix}",
tenant_id=self.tenant.id,
account_id=account.id,
email=account.email,
)
self.session.add_all([account, user])
self.session.flush()
return user
def test_materialized_default_role_is_implicit_and_cannot_be_assigned_or_removed(self) -> None:
first = self._add_user("first")
self._add_user("second")
with (
patch(
"govoplan_access.backend.admin.service.role_templates_for_level",
return_value=(DOCS_READER,),
),
patch(
"govoplan_access.backend.security.sessions.role_templates_for_level",
return_value=(DOCS_READER,),
),
):
roles = ensure_default_roles(self.session, self.tenant)
default_role = roles["docs_reader"]
self.assertFalse(default_role.is_assignable)
self.assertEqual(self.session.query(UserRoleAssignment).count(), 0)
# The database row is only an administration projection. Even if
# stale or tampered, it must never broaden the manifest baseline.
default_role.permissions = ["mail:profile:write"]
self.session.flush()
set_user_roles(self.session, user=first, role_ids=[])
context = collect_user_authorization_context(
self.session,
first,
account=first.account,
include_system=False,
)
self.assertEqual(self.session.query(UserRoleAssignment).count(), 0)
self.assertIn("docs:documentation:read", context.scopes)
self.assertNotIn("mail:profile:write", context.scopes)
self.assertEqual([role.slug for role in context.tenant_roles], ["docs_reader"])
def test_first_authorized_request_grants_default_without_database_mutation(self) -> None:
user = self._add_user("new")
with patch(
"govoplan_access.backend.security.sessions.role_templates_for_level",
return_value=(DOCS_READER,),
):
context = collect_user_authorization_context(
self.session,
user,
account=user.account,
include_system=False,
)
self.assertIn("docs:documentation:read", context.scopes)
self.assertEqual(context.tenant_roles, [])
self.assertEqual(self.session.query(Role).count(), 0)
self.assertEqual(self.session.query(UserRoleAssignment).count(), 0)
self.assertFalse(self.session.new)
self.assertFalse(self.session.dirty)
user_id = user.id
self.session.commit()
self.session.close()
self.session = self.Session()
persisted_user = self.session.get(User, user_id)
with patch(
"govoplan_access.backend.security.sessions.role_templates_for_level",
return_value=(DOCS_READER,),
):
reopened_context = collect_user_authorization_context(
self.session,
persisted_user,
account=persisted_user.account,
include_system=False,
)
self.assertIn("docs:documentation:read", reopened_context.scopes)
self.assertEqual(self.session.query(Role).count(), 0)
self.assertEqual(self.session.query(UserRoleAssignment).count(), 0)
if __name__ == "__main__":
unittest.main()