467 lines
19 KiB
Python
467 lines
19 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.api.v1.schemas import (
|
|
GroupInfo,
|
|
LoginRequest,
|
|
LoginResponse,
|
|
MeResponse,
|
|
PrincipalContextInfo,
|
|
ProfileUpdateRequest,
|
|
RoleInfo,
|
|
SwitchTenantRequest,
|
|
TenantInfo,
|
|
TenantMembershipInfo,
|
|
UserInfo,
|
|
UserUiPreferences,
|
|
)
|
|
from govoplan_core.core.access import AuthMethod, PrincipalRef
|
|
from govoplan_core.core.identity import CAPABILITY_IDENTITY_DIRECTORY, IdentityDirectory
|
|
from govoplan_core.core.registry import PlatformRegistry
|
|
from govoplan_access.backend.auth.dependencies import ApiPrincipal, get_api_principal
|
|
from govoplan_core.admin.settings import get_system_settings
|
|
from govoplan_core.audit.logging import audit_from_principal
|
|
from govoplan_core.core.maintenance import MAINTENANCE_ACCESS_SCOPE, maintenance_response_detail, saved_maintenance_mode
|
|
from govoplan_access.backend.db.models import Account, Tenant, User
|
|
from govoplan_core.db.session import get_session
|
|
from govoplan_core.i18n import (
|
|
i18n_settings,
|
|
normalize_enabled_language_codes,
|
|
normalize_language_code,
|
|
preferred_language_code,
|
|
system_enabled_language_codes,
|
|
system_i18n_payload,
|
|
tenant_enabled_language_codes,
|
|
user_enabled_language_codes,
|
|
)
|
|
from govoplan_access.backend.permissions.catalog import normalize_email, scopes_grant
|
|
from govoplan_core.security.time import utc_now
|
|
from govoplan_core.settings import settings
|
|
from govoplan_access.backend.semantic import collect_function_assignment_ids, collect_function_delegation_ids, identity_id_for_account
|
|
from govoplan_access.backend.auth.tenant_context import AccessTenantContextSwitcher
|
|
from govoplan_access.backend.security.passwords import verify_password
|
|
from govoplan_access.backend.security.sessions import (
|
|
collect_system_roles,
|
|
collect_tenant_memberships,
|
|
collect_user_groups,
|
|
collect_user_roles,
|
|
collect_user_scopes,
|
|
create_auth_session,
|
|
)
|
|
|
|
router = APIRouter(prefix="/auth", tags=["auth"])
|
|
|
|
|
|
def _cookie_samesite() -> str:
|
|
value = settings.auth_cookie_samesite.lower().strip()
|
|
if value not in {"lax", "strict", "none"}:
|
|
return "lax"
|
|
if value == "none" and not settings.auth_cookie_secure:
|
|
return "lax"
|
|
return value
|
|
|
|
|
|
def _set_auth_cookies(response: Response, created) -> None:
|
|
max_age = max(0, int((created.model.expires_at - utc_now()).total_seconds()))
|
|
common = {
|
|
"secure": settings.auth_cookie_secure,
|
|
"samesite": _cookie_samesite(),
|
|
"max_age": max_age,
|
|
"path": "/",
|
|
}
|
|
if settings.auth_cookie_domain:
|
|
common["domain"] = settings.auth_cookie_domain
|
|
response.set_cookie(settings.auth_session_cookie_name, created.token, httponly=True, **common)
|
|
response.set_cookie(settings.auth_csrf_cookie_name, created.csrf_token, httponly=False, **common)
|
|
|
|
|
|
def _clear_auth_cookies(response: Response) -> None:
|
|
kwargs = {"path": "/"}
|
|
if settings.auth_cookie_domain:
|
|
kwargs["domain"] = settings.auth_cookie_domain
|
|
response.delete_cookie(settings.auth_session_cookie_name, **kwargs)
|
|
response.delete_cookie(settings.auth_csrf_cookie_name, **kwargs)
|
|
|
|
|
|
def _tenant_info(tenant: Tenant, *, enabled_language_codes: list[str] | None = None) -> TenantInfo:
|
|
return TenantInfo(
|
|
id=tenant.id,
|
|
slug=tenant.slug,
|
|
name=tenant.name,
|
|
is_active=tenant.is_active,
|
|
default_locale=tenant.default_locale,
|
|
enabled_language_codes=enabled_language_codes or [],
|
|
)
|
|
|
|
|
|
def _user_ui_preferences(settings_payload: object) -> UserUiPreferences:
|
|
raw = settings_payload.get("ui") if isinstance(settings_payload, dict) else None
|
|
try:
|
|
return UserUiPreferences.model_validate(raw if isinstance(raw, dict) else {})
|
|
except ValueError:
|
|
return UserUiPreferences()
|
|
|
|
|
|
def _user_info(
|
|
user: User,
|
|
account: Account,
|
|
*,
|
|
preferred_language: str | None = None,
|
|
enabled_language_codes: list[str] | None = None,
|
|
) -> UserInfo:
|
|
return UserInfo(
|
|
id=user.id,
|
|
account_id=account.id,
|
|
email=account.email,
|
|
display_name=account.display_name or user.display_name,
|
|
tenant_display_name=user.display_name,
|
|
is_tenant_admin=user.is_tenant_admin,
|
|
password_reset_required=account.password_reset_required,
|
|
preferred_language=preferred_language,
|
|
enabled_language_codes=enabled_language_codes or [],
|
|
ui_preferences=_user_ui_preferences(user.settings),
|
|
)
|
|
|
|
|
|
def _roles_info(roles, *, level: str = "tenant") -> list[RoleInfo]:
|
|
return [
|
|
RoleInfo(
|
|
id=role.id,
|
|
slug=role.slug,
|
|
name=role.name,
|
|
permissions=role.permissions or [],
|
|
level=level,
|
|
)
|
|
for role in roles
|
|
]
|
|
|
|
|
|
def _groups_info(groups) -> list[GroupInfo]:
|
|
return [GroupInfo(id=group.id, slug=group.slug, name=group.name) for group in groups]
|
|
|
|
|
|
def _tenant_memberships(session: Session, account: Account) -> list[TenantMembershipInfo]:
|
|
memberships: list[TenantMembershipInfo] = []
|
|
for user, tenant in collect_tenant_memberships(session, account):
|
|
roles = collect_user_roles(session, user)
|
|
memberships.append(
|
|
TenantMembershipInfo(
|
|
id=tenant.id,
|
|
slug=tenant.slug,
|
|
name=tenant.name,
|
|
is_active=tenant.is_active and user.is_active,
|
|
default_locale=tenant.default_locale,
|
|
roles=[role.slug for role in roles],
|
|
)
|
|
)
|
|
return memberships
|
|
|
|
|
|
def _language_context(session: Session, *, tenant: Tenant, user: User) -> dict[str, object]:
|
|
system_settings = get_system_settings(session)
|
|
system_payload = system_i18n_payload(system_settings)
|
|
system_enabled = system_enabled_language_codes(system_settings.settings, default_locale=system_settings.default_locale)
|
|
tenant_enabled = tenant_enabled_language_codes(tenant.settings, system_enabled, default_locale=tenant.default_locale)
|
|
user_enabled = user_enabled_language_codes(user.settings, tenant_enabled)
|
|
preferred = preferred_language_code(
|
|
user.settings,
|
|
user_enabled,
|
|
default_locale=tenant.default_locale or system_payload.get("default_language"),
|
|
)
|
|
return {
|
|
"available_languages": system_payload["available_languages"],
|
|
"tenant_enabled_language_codes": tenant_enabled,
|
|
"user_enabled_language_codes": user_enabled,
|
|
"preferred_language": preferred,
|
|
}
|
|
|
|
|
|
def _identity_directory_from_request(request: Request) -> IdentityDirectory | None:
|
|
registry = getattr(request.app.state, "govoplan_registry", None)
|
|
if not isinstance(registry, PlatformRegistry) or not registry.has_capability(CAPABILITY_IDENTITY_DIRECTORY):
|
|
return None
|
|
capability = registry.require_capability(CAPABILITY_IDENTITY_DIRECTORY)
|
|
if not isinstance(capability, IdentityDirectory):
|
|
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Invalid capability: {CAPABILITY_IDENTITY_DIRECTORY}")
|
|
return capability
|
|
|
|
|
|
def _resolve_login_user(session: Session, payload: LoginRequest) -> tuple[Account, User, Tenant]:
|
|
account = (
|
|
session.query(Account)
|
|
.filter(Account.normalized_email == normalize_email(payload.email), Account.is_active.is_(True))
|
|
.one_or_none()
|
|
)
|
|
if account is None or not verify_password(payload.password, account.password_hash):
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid login")
|
|
|
|
query = (
|
|
session.query(User, Tenant)
|
|
.join(Tenant, Tenant.id == User.tenant_id)
|
|
.filter(
|
|
User.account_id == account.id,
|
|
User.is_active.is_(True),
|
|
Tenant.is_active.is_(True),
|
|
)
|
|
)
|
|
if payload.tenant_slug:
|
|
query = query.filter(Tenant.slug == payload.tenant_slug)
|
|
row = query.order_by(Tenant.name.asc()).first()
|
|
if row is None:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="No active tenant membership")
|
|
return account, row[0], row[1]
|
|
|
|
|
|
def _me_response(
|
|
session: Session,
|
|
*,
|
|
account: Account,
|
|
user: User,
|
|
tenant: Tenant,
|
|
effective_scopes: list[str] | None = None,
|
|
auth_method: AuthMethod = "session",
|
|
api_key_id: str | None = None,
|
|
session_id: str | None = None,
|
|
service_account_id: str | None = None,
|
|
include_system: bool = True,
|
|
include_all_memberships: bool = True,
|
|
identity_directory: IdentityDirectory | None = None,
|
|
identity_id: str | None = None,
|
|
) -> MeResponse:
|
|
tenant_roles = collect_user_roles(session, user)
|
|
system_roles = collect_system_roles(session, account) if include_system else []
|
|
groups = collect_user_groups(session, user)
|
|
scopes = effective_scopes if effective_scopes is not None else collect_user_scopes(session, user, include_system=include_system)
|
|
languages = _language_context(session, tenant=tenant, user=user)
|
|
tenant_enabled = list(languages["tenant_enabled_language_codes"])
|
|
user_enabled = list(languages["user_enabled_language_codes"])
|
|
preferred_language = str(languages["preferred_language"])
|
|
active_tenant = _tenant_info(tenant, enabled_language_codes=tenant_enabled)
|
|
memberships = _tenant_memberships(session, account) if include_all_memberships else [
|
|
TenantMembershipInfo(
|
|
id=tenant.id,
|
|
slug=tenant.slug,
|
|
name=tenant.name,
|
|
is_active=tenant.is_active and user.is_active,
|
|
default_locale=tenant.default_locale,
|
|
roles=[role.slug for role in tenant_roles],
|
|
)
|
|
]
|
|
return MeResponse(
|
|
user=_user_info(user, account, preferred_language=preferred_language, enabled_language_codes=user_enabled),
|
|
tenant=active_tenant,
|
|
active_tenant=active_tenant,
|
|
tenants=memberships,
|
|
scopes=scopes,
|
|
roles=_roles_info(tenant_roles) + _roles_info(system_roles, level="system"),
|
|
groups=_groups_info(groups),
|
|
principal=PrincipalContextInfo.model_validate(
|
|
PrincipalRef(
|
|
account_id=account.id,
|
|
membership_id=user.id,
|
|
tenant_id=tenant.id,
|
|
identity_id=identity_id or identity_id_for_account(session, account.id, identity_directory=identity_directory),
|
|
scopes=frozenset(scopes),
|
|
group_ids=frozenset(group.id for group in groups),
|
|
role_ids=frozenset(role.id for role in tenant_roles + system_roles),
|
|
function_assignment_ids=frozenset(collect_function_assignment_ids(session, user)),
|
|
delegation_ids=frozenset(collect_function_delegation_ids(session, user)),
|
|
auth_method=auth_method,
|
|
api_key_id=api_key_id,
|
|
session_id=session_id,
|
|
service_account_id=service_account_id,
|
|
email=account.email,
|
|
display_name=account.display_name or user.display_name,
|
|
).to_dict()
|
|
),
|
|
available_languages=languages["available_languages"],
|
|
enabled_language_codes=tenant_enabled,
|
|
default_language=preferred_language,
|
|
)
|
|
|
|
|
|
@router.post("/login", response_model=LoginResponse)
|
|
def login(payload: LoginRequest, request: Request, response: Response, session: Session = Depends(get_session)):
|
|
account, user, tenant = _resolve_login_user(session, payload)
|
|
identity_directory = _identity_directory_from_request(request)
|
|
me_payload = _me_response(session, account=account, user=user, tenant=tenant, identity_directory=identity_directory)
|
|
maintenance_mode = saved_maintenance_mode(session)
|
|
if maintenance_mode.enabled and not scopes_grant(me_payload.scopes, MAINTENANCE_ACCESS_SCOPE):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail=maintenance_response_detail(maintenance_mode),
|
|
)
|
|
user_agent = request.headers.get("user-agent")
|
|
ip_address = request.client.host if request.client else None
|
|
created = create_auth_session(
|
|
session,
|
|
user=user,
|
|
hours=settings.auth_session_hours,
|
|
user_agent=user_agent,
|
|
ip_address=ip_address,
|
|
)
|
|
session.commit()
|
|
_set_auth_cookies(response, created)
|
|
return LoginResponse(
|
|
access_token=created.token,
|
|
expires_at=created.model.expires_at,
|
|
**_me_response(
|
|
session,
|
|
account=account,
|
|
user=user,
|
|
tenant=tenant,
|
|
effective_scopes=me_payload.scopes,
|
|
auth_method="session",
|
|
session_id=created.model.id,
|
|
identity_directory=identity_directory,
|
|
).model_dump(),
|
|
)
|
|
|
|
|
|
@router.get("/me", response_model=MeResponse)
|
|
def me(principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session)):
|
|
tenant = session.get(Tenant, principal.tenant_id)
|
|
if tenant is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Active tenant not found")
|
|
return _me_response(
|
|
session,
|
|
account=principal.account,
|
|
user=principal.user,
|
|
tenant=tenant,
|
|
effective_scopes=principal.scopes,
|
|
auth_method=principal.auth_method,
|
|
api_key_id=principal.api_key_id,
|
|
session_id=principal.session_id,
|
|
service_account_id=principal.principal.service_account_id,
|
|
include_system=principal.auth_session is not None,
|
|
include_all_memberships=principal.auth_session is not None,
|
|
identity_id=principal.principal.identity_id,
|
|
)
|
|
|
|
|
|
@router.patch("/profile", response_model=MeResponse)
|
|
def update_profile(
|
|
payload: ProfileUpdateRequest,
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
session: Session = Depends(get_session),
|
|
):
|
|
if principal.auth_session is None:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="API keys cannot edit an interactive user profile")
|
|
|
|
if "display_name" in payload.model_fields_set:
|
|
principal.account.display_name = payload.display_name.strip() if payload.display_name else None
|
|
session.add(principal.account)
|
|
if "tenant_display_name" in payload.model_fields_set:
|
|
principal.user.display_name = payload.tenant_display_name.strip() if payload.tenant_display_name else None
|
|
session.add(principal.user)
|
|
next_settings = dict(principal.user.settings or {})
|
|
settings_changed = False
|
|
if {"preferred_language", "enabled_language_codes"}.intersection(payload.model_fields_set):
|
|
tenant = session.get(Tenant, principal.tenant_id)
|
|
if tenant is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Active tenant not found")
|
|
system_settings = get_system_settings(session)
|
|
system_enabled = system_enabled_language_codes(system_settings.settings, default_locale=system_settings.default_locale)
|
|
tenant_enabled = tenant_enabled_language_codes(tenant.settings, system_enabled, default_locale=tenant.default_locale)
|
|
next_i18n = i18n_settings(next_settings)
|
|
if "preferred_language" in payload.model_fields_set:
|
|
preferred = normalize_language_code(payload.preferred_language)
|
|
if preferred:
|
|
normalized_preferred = normalize_enabled_language_codes(
|
|
[preferred],
|
|
[{"code": code} for code in tenant_enabled],
|
|
fallback_codes=tenant_enabled,
|
|
)[0]
|
|
next_i18n["preferred_language"] = normalized_preferred
|
|
else:
|
|
next_i18n.pop("preferred_language", None)
|
|
if "enabled_language_codes" in payload.model_fields_set:
|
|
next_i18n["enabled_language_codes"] = normalize_enabled_language_codes(
|
|
payload.enabled_language_codes,
|
|
[{"code": code} for code in tenant_enabled],
|
|
fallback_codes=tenant_enabled,
|
|
)
|
|
next_settings["i18n"] = next_i18n
|
|
settings_changed = True
|
|
if "ui_preferences" in payload.model_fields_set:
|
|
if payload.ui_preferences is None:
|
|
next_settings["ui"] = UserUiPreferences().model_dump()
|
|
else:
|
|
next_ui = _user_ui_preferences(next_settings).model_dump()
|
|
next_ui.update(payload.ui_preferences.model_dump(exclude_unset=True))
|
|
next_settings["ui"] = UserUiPreferences.model_validate(next_ui).model_dump()
|
|
settings_changed = True
|
|
if settings_changed:
|
|
principal.user.settings = next_settings
|
|
session.add(principal.user)
|
|
|
|
audit_from_principal(
|
|
session,
|
|
principal,
|
|
action="profile.updated",
|
|
object_type="account",
|
|
object_id=principal.account.id,
|
|
details={"fields": sorted(payload.model_fields_set)},
|
|
)
|
|
tenant = session.get(Tenant, principal.tenant_id)
|
|
if tenant is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Active tenant not found")
|
|
session.commit()
|
|
return _me_response(
|
|
session,
|
|
account=principal.account,
|
|
user=principal.user,
|
|
tenant=tenant,
|
|
auth_method=principal.auth_method,
|
|
session_id=principal.session_id,
|
|
include_system=True,
|
|
include_all_memberships=True,
|
|
identity_id=principal.principal.identity_id,
|
|
)
|
|
|
|
|
|
@router.post("/switch-tenant", response_model=MeResponse)
|
|
def switch_tenant(
|
|
payload: SwitchTenantRequest,
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
session: Session = Depends(get_session),
|
|
):
|
|
if principal.auth_session is None:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="API keys cannot switch tenant context")
|
|
try:
|
|
switched = AccessTenantContextSwitcher().switch_tenant_context(session, principal=principal, tenant_id=payload.tenant_id)
|
|
except LookupError as exc:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc)) from exc
|
|
tenant = session.get(Tenant, switched.tenant_id)
|
|
membership = session.get(User, switched.membership_id)
|
|
if tenant is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Tenant not found")
|
|
if membership is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Tenant membership not found")
|
|
session.commit()
|
|
return _me_response(
|
|
session,
|
|
account=principal.account,
|
|
user=membership,
|
|
tenant=tenant,
|
|
auth_method="session",
|
|
session_id=principal.session_id,
|
|
identity_id=principal.principal.identity_id,
|
|
)
|
|
|
|
|
|
@router.post("/logout")
|
|
def logout(
|
|
response: Response,
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
session: Session = Depends(get_session),
|
|
):
|
|
if principal.auth_session is not None:
|
|
principal.auth_session.revoked_at = utc_now()
|
|
session.add(principal.auth_session)
|
|
session.commit()
|
|
_clear_auth_cookies(response)
|
|
return {"ok": True}
|