GovOPlaN/govoplan-access
2
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(access): add governed session management

Browse Source
This commit is contained in:
zemion
2026-08-19 22:50:15 +02:00
parent 38fc22c06b
commit 2be1dbc132
16 changed files with 1302 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/govoplan_access/backend/api/v1/admin_schemas.py
+27
View File
@@ -37,6 +37,33 @@ class AdminOverviewResponse(BaseModel):
capabilities: list[str] = Field(default_factory=list)
class AdminSessionItem(BaseModel):
id: str
tenant_id: str
current: bool
status: Literal["active", "expired", "revoked"]
created_at: datetime
last_seen_at: datetime | None = None
expires_at: datetime
revoked_at: datetime | None = None
client: str | None = None
class AdminSessionListResponse(BaseModel):
sessions: list[AdminSessionItem] = Field(default_factory=list)
class AdminSessionRevocationRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
current_password: str = Field(min_length=1, max_length=1024)
class AdminSessionRevocationResponse(BaseModel):
session: AdminSessionItem
revoked: bool
class TenantAdminItem(BaseModel):
id: str
slug: str = Field(min_length=1, max_length=100)
src/govoplan_access/backend/api/v1/auth.py
+140 -2
View File
@@ -1,8 +1,9 @@
from __future__ import annotations
from dataclasses import dataclass
from dataclasses import asdict, dataclass
from datetime import datetime
from functools import lru_cache
from typing import Literal
from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
from pydantic import BaseModel, Field
@@ -38,7 +39,7 @@ from govoplan_core.core.idm import (
IdmDirectory,
OrganizationFunctionAssignmentRef,
)
from govoplan_access.backend.auth.dependencies import ApiPrincipal, get_api_principal
from govoplan_access.backend.auth.dependencies import ApiPrincipal, get_api_principal, require_scope
from govoplan_core.admin.settings import get_system_settings
from govoplan_core.audit.logging import audit_event
from govoplan_core.core.maintenance import MAINTENANCE_ACCESS_SCOPE, maintenance_response_detail, saved_maintenance_mode
@@ -78,6 +79,13 @@ from govoplan_access.backend.security.sessions import (
create_auth_session,
verify_auth_session_csrf,
)
from govoplan_access.backend.session_management import (
SessionSummary,
list_account_sessions,
revoke_account_session,
revoke_other_account_sessions,
session_summary,
)
router = APIRouter(prefix="/auth", tags=["auth"])
@@ -96,6 +104,44 @@ class ActingContextListResponse(BaseModel):
active_assignment_id: str | None = None
class AccountSessionInfo(BaseModel):
id: str
tenant_id: str
current: bool
status: Literal["active", "expired", "revoked"]
created_at: datetime
last_seen_at: datetime | None = None
expires_at: datetime
revoked_at: datetime | None = None
client: str | None = None
class AccountSessionListResponse(BaseModel):
sessions: list[AccountSessionInfo] = Field(default_factory=list)
class AccountSessionRevocationResponse(BaseModel):
session: AccountSessionInfo
revoked: bool
class OtherSessionRevocationResponse(BaseModel):
revoked_count: int
def _account_session_info(item: SessionSummary) -> AccountSessionInfo:
return AccountSessionInfo(**asdict(item))
def _interactive_session(principal: ApiPrincipal) -> AuthSession:
if principal.auth_session is None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Session management requires an interactive browser session.",
)
return principal.auth_session
def _acting_assignments(
request: Request,
*,
@@ -1002,6 +1048,98 @@ def switch_tenant(
)
@router.get("/sessions", response_model=AccountSessionListResponse)
def list_own_sessions(
principal: ApiPrincipal = Depends(require_scope("access:session:manage_own")),
session: Session = Depends(get_session),
) -> AccountSessionListResponse:
current = _interactive_session(principal)
items = list_account_sessions(
session,
account_id=principal.account_id,
current_session_id=current.id,
)
return AccountSessionListResponse(
sessions=[_account_session_info(item) for item in items]
)
@router.post(
"/sessions/revoke-others",
response_model=OtherSessionRevocationResponse,
)
def revoke_own_other_sessions(
principal: ApiPrincipal = Depends(require_scope("access:session:manage_own")),
session: Session = Depends(get_session),
) -> OtherSessionRevocationResponse:
current = _interactive_session(principal)
revoked_ids = revoke_other_account_sessions(
session,
account_id=principal.account_id,
current_session_id=current.id,
)
if revoked_ids:
audit_event(
session,
tenant_id=principal.tenant_id,
user_id=principal.user.id,
action="access.sessions.other_sessions_revoked",
object_type="access_account",
object_id=principal.account_id,
details={"revoked_count": len(revoked_ids)},
)
session.commit()
principal_summary_cache.clear()
return OtherSessionRevocationResponse(revoked_count=len(revoked_ids))
@router.post(
"/sessions/{session_id}/revoke",
response_model=AccountSessionRevocationResponse,
)
def revoke_own_session(
session_id: str,
principal: ApiPrincipal = Depends(require_scope("access:session:manage_own")),
session: Session = Depends(get_session),
) -> AccountSessionRevocationResponse:
current = _interactive_session(principal)
try:
item, changed = revoke_account_session(
session,
account_id=principal.account_id,
session_id=session_id,
protected_session_id=current.id,
)
except ValueError as exc:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=str(exc),
) from exc
if item is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Session not found.",
)
if changed:
audit_event(
session,
tenant_id=principal.tenant_id,
user_id=principal.user.id,
action="access.session.revoked",
object_type="access_auth_session",
object_id=item.id,
details={"actor_kind": "self"},
)
session.commit()
principal_summary_cache.clear()
return AccountSessionRevocationResponse(
session=_account_session_info(
session_summary(item, current_session_id=current.id)
),
revoked=changed,
)
@router.get("/acting-contexts", response_model=ActingContextListResponse)
def list_acting_contexts(
request: Request,
src/govoplan_access/backend/api/v1/routes.py
+131
View File
@@ -1,6 +1,7 @@
from __future__ import annotations
from collections.abc import Iterable
from dataclasses import asdict
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, Query, status
@@ -62,6 +63,10 @@ from govoplan_access.backend.api.v1.admin_common import (
_user_item,
)
from govoplan_access.backend.api.v1.admin_schemas import (
AdminSessionItem,
AdminSessionListResponse,
AdminSessionRevocationRequest,
AdminSessionRevocationResponse,
AdminApiKeyCreateRequest,
AdminApiKeyCreateResponse,
ApiKeyAdminItem,
@@ -137,6 +142,14 @@ from govoplan_access.backend.api.v1.admin_schemas import (
UserUpdateRequest,
)
from govoplan_access.backend.security.api_keys import create_api_key
from govoplan_access.backend.security.passwords import verify_password
from govoplan_access.backend.session_management import (
SessionSummary,
list_account_sessions,
revoke_account_session,
session_summary,
)
from govoplan_access.backend.auth.principal_cache import principal_summary_cache
from govoplan_access.backend.auth.dependencies import ApiPrincipal, get_api_principal, has_scope, require_any_scope, require_scope
from govoplan_core.audit.logging import audit_event, audit_from_principal
from govoplan_access.backend.configuration_provider import ACCESS_CONFIGURATION_CAPABILITY, SqlAccessConfigurationProvider
@@ -2380,6 +2393,124 @@ def get_user_access_explanation(
)
def _admin_session_item(item: SessionSummary) -> AdminSessionItem:
return AdminSessionItem(**asdict(item))
def _require_session_admin_reauthorization(
principal: ApiPrincipal,
current_password: str,
) -> None:
if principal.auth_session is None:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Administrative session revocation requires an interactive session.",
)
if not verify_password(current_password, principal.account.password_hash):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Current password re-authorization failed.",
)
@router.get(
"/users/{user_id}/sessions",
response_model=AdminSessionListResponse,
)
def list_user_sessions(
user_id: str,
tenant_id: str | None = Query(default=None),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(
require_any_scope("admin:users:read", "access:membership:read")
),
) -> AdminSessionListResponse:
tenant = _resolve_tenant(session, principal, tenant_id)
user = (
session.query(User)
.filter(User.id == user_id, User.tenant_id == tenant.id)
.one_or_none()
)
if user is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found",
)
items = list_account_sessions(
session,
account_id=user.account_id,
tenant_id=tenant.id,
current_session_id=principal.session_id,
)
return AdminSessionListResponse(
sessions=[_admin_session_item(item) for item in items]
)
@router.post(
"/users/{user_id}/sessions/{session_id}/revoke",
response_model=AdminSessionRevocationResponse,
)
def revoke_user_session(
user_id: str,
session_id: str,
payload: AdminSessionRevocationRequest,
tenant_id: str | None = Query(default=None),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(
require_any_scope("admin:users:update", "access:membership:update")
),
) -> AdminSessionRevocationResponse:
_require_session_admin_reauthorization(principal, payload.current_password)
tenant = _resolve_tenant(session, principal, tenant_id)
user = (
session.query(User)
.filter(User.id == user_id, User.tenant_id == tenant.id)
.one_or_none()
)
if user is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found",
)
try:
item, changed = revoke_account_session(
session,
account_id=user.account_id,
session_id=session_id,
tenant_id=tenant.id,
protected_session_id=principal.session_id,
)
except ValueError as exc:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=str(exc),
) from exc
if item is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Session not found",
)
if changed:
audit_from_principal(
session,
principal,
action="access.session.revoked_by_administrator",
scope="tenant",
object_type="access_auth_session",
object_id=item.id,
details={"target_membership_id": user.id},
)
session.commit()
principal_summary_cache.clear()
return AdminSessionRevocationResponse(
session=_admin_session_item(
session_summary(item, current_session_id=principal.session_id)
),
revoked=changed,
)
@router.get("/access/resource-explanation", response_model=ResourceAccessExplanationResponse)
def get_resource_access_explanation(
user_id: str = Query(...),
src/govoplan_access/backend/manifest.py
+73
View File
@@ -84,6 +84,7 @@ ACCESS_PERMISSIONS: tuple[PermissionDefinition, ...] = (
_permission("access:membership:read", "View memberships", "List tenant memberships and effective access.", "Tenant access", "tenant"),
_permission("access:membership:create", "Create memberships", "Create tenant-local account memberships.", "Tenant access", "tenant"),
_permission("access:membership:update", "Update memberships", "Update or suspend tenant memberships.", "Tenant access", "tenant"),
_permission("access:session:manage_own", "Manage own sessions", "Inspect and revoke the current account's browser sessions without exposing credentials.", "Tenant access", "tenant"),
_permission("access:group:read", "View groups", "List tenant groups and members.", "Tenant access", "tenant"),
_permission("access:group:write", "Manage groups", "Create and update tenant groups.", "Tenant access", "tenant"),
_permission("access:group:manage_members", "Manage group members", "Add and remove memberships from groups.", "Tenant access", "tenant"),
@@ -176,6 +177,16 @@ ACCESS_ROLE_TEMPLATES: tuple[RoleTemplate, ...] = (
managed=False,
protected=False,
),
RoleTemplate(
slug="account_security",
name="Account security",
description="Authenticated baseline for inspecting and revoking the current account's browser sessions.",
permissions=("access:session:manage_own",),
level="tenant",
managed=True,
protected=True,
default_authenticated=True,
),
RoleTemplate(
slug="owner",
name="Tenant owner",
@@ -628,6 +639,67 @@ ACCESS_DOCUMENTATION: tuple[DocumentationTopic, ...] = (
"verification": "The administration table shows the expected active credential count, last-use timestamp, revision, and audit events without exposing secret material.",
},
),
DocumentationTopic(
id="access.workflow.manage-sessions",
title="Review and revoke account sessions",
summary="Inspect active browser sessions and revoke one or every other session without exposing credentials or network identifiers.",
body=(
"Settings > Sessions and devices marks the current browser session and shows only bounded client metadata plus creation, last-seen, and expiry times. "
"Users can revoke another session or all other active sessions; the command session is protected and normal logout remains the way to end it. Revocation is idempotent and takes effect on the next authenticated request. "
"Tenant administrators can inspect only sessions belonging to a membership in their governed tenant. Administrative revocation requires central membership-update permission and current-password re-authorization from an interactive session. "
"Audit evidence records stable actors, targets, and counts without tokens, hashes, cookies, IP addresses, or client strings."
),
layer="always",
documentation_types=("admin", "user"),
audience=("user", "tenant_admin", "access_admin", "operator"),
conditions=(
DocumentationCondition(
required_modules=("access",),
any_scopes=("access:session:manage_own",),
),
),
links=(
DocumentationLink(label="Sessions and devices", href="/settings?section=sessions", kind="runtime"),
DocumentationLink(label="Own sessions API", href="/api/v1/auth/sessions", kind="api"),
DocumentationLink(label="Session management reference", href="docs/SESSION_MANAGEMENT.md", kind="repository"),
),
translations={
"de": {
"title": "Kontositzungen prüfen und widerrufen",
"summary": "Aktive Browsersitzungen prüfen und einzelne oder alle anderen Sitzungen widerrufen, ohne Zugangsdaten oder Netzwerkkennungen offenzulegen.",
"body": (
"Einstellungen > Sitzungen und Geräte kennzeichnet die aktuelle Browsersitzung und zeigt nur begrenzte Clientmetadaten sowie Erstellungs-, Aktivitäts- und Ablaufzeitpunkte. "
"Benutzende können eine andere oder alle anderen aktiven Sitzungen widerrufen; die ausführende Sitzung bleibt geschützt und wird regulär abgemeldet. Der Widerruf ist idempotent und gilt beim nächsten authentifizierten Aufruf. "
"Mandantenadministrierende sehen nur Sitzungen einer Mitgliedschaft im verwalteten Mandanten. Der administrative Widerruf erfordert die zentrale Berechtigung zur Mitgliedschaftsänderung und eine erneute Passwortbestätigung in einer interaktiven Sitzung. "
"Auditnachweise speichern stabile Akteure, Ziele und Anzahlen, aber keine Token, Hashes, Cookies, IP-Adressen oder Clienttexte."
),
}
},
metadata={
"kind": "workflow",
"help_contexts": [
"access.settings.sessions",
"access.sessions.action.revoke",
"access.sessions.action.revoke-others",
"access.admin.user-sessions",
],
"api_paths": [
"/api/v1/auth/sessions",
"/api/v1/auth/sessions/{session_id}/revoke",
"/api/v1/auth/sessions/revoke-others",
"/api/v1/admin/users/{user_id}/sessions",
"/api/v1/admin/users/{user_id}/sessions/{session_id}/revoke",
],
"sensitive_fields_never_returned": [
"token",
"token_hash",
"csrf_token_hash",
"cookie",
"ip_address",
],
},
order=33,
),
DocumentationTopic(
id="access.reference.external-function-role-mappings",
title="Organization function facts and access roles",
@@ -992,6 +1064,7 @@ manifest = ModuleManifest(
ViewSurface(id="access.admin.group-credentials", module_id="access", kind="section", label="Group credentials", order=30),
ViewSurface(id="access.admin.user-credentials", module_id="access", kind="section", label="User credentials", order=30),
ViewSurface(id="access.settings.credentials", module_id="access", kind="section", label="Personal credentials", order=30),
ViewSurface(id="access.settings.sessions", module_id="access", kind="section", label="Sessions and devices", order=20),
),
),
capability_factories={
src/govoplan_access/backend/session_management.py
+155
View File
@@ -0,0 +1,155 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
from sqlalchemy.orm import Session
from govoplan_access.backend.db.models import AuthSession
from govoplan_core.security.time import ensure_aware_utc, utc_now
MAX_SESSION_LIST_ITEMS = 100
MAX_CLIENT_LABEL_LENGTH = 160
@dataclass(frozen=True, slots=True)
class SessionSummary:
id: str
tenant_id: str
current: bool
status: str
created_at: datetime
last_seen_at: datetime | None
expires_at: datetime
revoked_at: datetime | None
client: str | None
def session_summary(
item: AuthSession,
*,
current_session_id: str | None,
now: datetime | None = None,
) -> SessionSummary:
effective_at = now or utc_now()
expires_at = ensure_aware_utc(item.expires_at)
revoked_at = ensure_aware_utc(item.revoked_at)
if revoked_at is not None:
status = "revoked"
elif expires_at is None or expires_at <= effective_at:
status = "expired"
else:
status = "active"
return SessionSummary(
id=item.id,
tenant_id=item.tenant_id,
current=item.id == current_session_id,
status=status,
created_at=item.created_at,
last_seen_at=item.last_seen_at,
expires_at=item.expires_at,
revoked_at=item.revoked_at,
client=_bounded_client(item.user_agent),
)
def list_account_sessions(
session: Session,
*,
account_id: str,
current_session_id: str | None,
tenant_id: str | None = None,
include_inactive: bool = False,
limit: int = MAX_SESSION_LIST_ITEMS,
now: datetime | None = None,
) -> tuple[SessionSummary, ...]:
effective_at = now or utc_now()
query = session.query(AuthSession).filter(AuthSession.account_id == account_id)
if tenant_id is not None:
query = query.filter(AuthSession.tenant_id == tenant_id)
rows = query.order_by(AuthSession.created_at.desc(), AuthSession.id.asc()).all()
summaries = tuple(
session_summary(
item,
current_session_id=current_session_id,
now=effective_at,
)
for item in rows
)
if not include_inactive:
summaries = tuple(item for item in summaries if item.status == "active")
return summaries[: max(1, min(limit, MAX_SESSION_LIST_ITEMS))]
def revoke_account_session(
session: Session,
*,
account_id: str,
session_id: str,
tenant_id: str | None = None,
protected_session_id: str | None = None,
now: datetime | None = None,
) -> tuple[AuthSession | None, bool]:
query = session.query(AuthSession).filter(
AuthSession.id == session_id,
AuthSession.account_id == account_id,
)
if tenant_id is not None:
query = query.filter(AuthSession.tenant_id == tenant_id)
item = query.one_or_none()
if item is None:
return None, False
if protected_session_id is not None and item.id == protected_session_id:
raise ValueError("The current session cannot be revoked through session management.")
if item.revoked_at is not None:
return item, False
item.revoked_at = now or utc_now()
session.add(item)
return item, True
def revoke_other_account_sessions(
session: Session,
*,
account_id: str,
current_session_id: str,
now: datetime | None = None,
) -> tuple[str, ...]:
effective_at = now or utc_now()
rows = (
session.query(AuthSession)
.filter(
AuthSession.account_id == account_id,
AuthSession.id != current_session_id,
AuthSession.revoked_at.is_(None),
)
.all()
)
revoked: list[str] = []
for item in rows:
expires_at = ensure_aware_utc(item.expires_at)
if expires_at is None or expires_at <= effective_at:
continue
item.revoked_at = effective_at
session.add(item)
revoked.append(item.id)
return tuple(sorted(revoked))
def _bounded_client(value: str | None) -> str | None:
normalized = " ".join(str(value or "").split())
if not normalized:
return None
return normalized[:MAX_CLIENT_LABEL_LENGTH]
__all__ = [
"MAX_CLIENT_LABEL_LENGTH",
"MAX_SESSION_LIST_ITEMS",
"SessionSummary",
"list_account_sessions",
"revoke_account_session",
"revoke_other_account_sessions",
"session_summary",
]