284 lines
7.5 KiB
Python
284 lines
7.5 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Iterable, Mapping
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_access.backend.db.base import utcnow
|
|
from govoplan_access.backend.db.models import (
|
|
Account,
|
|
ServiceAccount,
|
|
Tenant,
|
|
User,
|
|
new_uuid,
|
|
)
|
|
from govoplan_core.auth import ApiPrincipal
|
|
|
|
|
|
class ServiceAccountError(ValueError):
|
|
pass
|
|
|
|
|
|
class ServiceAccountNotFoundError(ServiceAccountError):
|
|
pass
|
|
|
|
|
|
class ServiceAccountConflictError(ServiceAccountError):
|
|
pass
|
|
|
|
|
|
def list_service_accounts(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
) -> list[ServiceAccount]:
|
|
return list(
|
|
session.scalars(
|
|
select(ServiceAccount)
|
|
.where(ServiceAccount.tenant_id == tenant_id)
|
|
.order_by(
|
|
ServiceAccount.normalized_name,
|
|
ServiceAccount.id,
|
|
)
|
|
)
|
|
)
|
|
|
|
|
|
def get_service_account(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
service_account_id: str,
|
|
lock: bool = False,
|
|
) -> ServiceAccount:
|
|
query = select(ServiceAccount).where(
|
|
ServiceAccount.id == service_account_id,
|
|
ServiceAccount.tenant_id == tenant_id,
|
|
)
|
|
if lock:
|
|
query = query.with_for_update()
|
|
item = session.scalar(query)
|
|
if item is None:
|
|
raise ServiceAccountNotFoundError(
|
|
"Service account was not found"
|
|
)
|
|
return item
|
|
|
|
|
|
def create_service_account(
|
|
session: Session,
|
|
*,
|
|
tenant: Tenant,
|
|
principal: ApiPrincipal,
|
|
name: str,
|
|
description: str | None,
|
|
scope_ceiling: Iterable[str],
|
|
) -> ServiceAccount:
|
|
clean_name = _service_account_name(name)
|
|
scopes = _service_account_scopes(
|
|
principal,
|
|
scope_ceiling,
|
|
)
|
|
service_account_id = new_uuid()
|
|
internal_email = (
|
|
f"service-account-{service_account_id}@govoplan.invalid"
|
|
)
|
|
account = Account(
|
|
id=new_uuid(),
|
|
email=internal_email,
|
|
normalized_email=internal_email,
|
|
display_name=clean_name,
|
|
is_active=True,
|
|
auth_provider="service_account",
|
|
password_hash=None,
|
|
password_reset_required=False,
|
|
)
|
|
membership = User(
|
|
id=new_uuid(),
|
|
tenant_id=tenant.id,
|
|
account=account,
|
|
email=internal_email,
|
|
display_name=clean_name,
|
|
is_active=True,
|
|
is_tenant_admin=False,
|
|
auth_provider="service_account",
|
|
password_hash=None,
|
|
settings={"managed_service_account": service_account_id},
|
|
)
|
|
item = ServiceAccount(
|
|
id=service_account_id,
|
|
tenant_id=tenant.id,
|
|
account_id=account.id,
|
|
membership_id=membership.id,
|
|
name=clean_name,
|
|
normalized_name=_normalized_name(clean_name),
|
|
description=_optional_text(description),
|
|
scope_ceiling=list(scopes),
|
|
is_active=True,
|
|
revision=1,
|
|
created_by_account_id=principal.account_id,
|
|
updated_by_account_id=principal.account_id,
|
|
settings={},
|
|
)
|
|
session.add_all((account, membership, item))
|
|
try:
|
|
session.flush()
|
|
except IntegrityError as exc:
|
|
raise ServiceAccountConflictError(
|
|
"A service account with this name already exists"
|
|
) from exc
|
|
return item
|
|
|
|
|
|
def update_service_account(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
service_account_id: str,
|
|
principal: ApiPrincipal,
|
|
expected_revision: int,
|
|
changes: Mapping[str, object],
|
|
) -> ServiceAccount:
|
|
item = get_service_account(
|
|
session,
|
|
tenant_id=tenant_id,
|
|
service_account_id=service_account_id,
|
|
lock=True,
|
|
)
|
|
if item.revision != expected_revision:
|
|
raise ServiceAccountConflictError(
|
|
"Service account changed on the server; reload before saving"
|
|
)
|
|
account = session.get(Account, item.account_id)
|
|
membership = session.get(User, item.membership_id)
|
|
if account is None or membership is None:
|
|
raise ServiceAccountConflictError(
|
|
"Service account backing identity is missing"
|
|
)
|
|
if "name" in changes:
|
|
clean_name = _service_account_name(str(changes["name"]))
|
|
item.name = clean_name
|
|
item.normalized_name = _normalized_name(clean_name)
|
|
account.display_name = clean_name
|
|
membership.display_name = clean_name
|
|
if "description" in changes:
|
|
value = changes["description"]
|
|
item.description = _optional_text(
|
|
str(value) if value is not None else None
|
|
)
|
|
if "scope_ceiling" in changes:
|
|
raw_scopes = changes["scope_ceiling"]
|
|
if not isinstance(raw_scopes, Iterable) or isinstance(
|
|
raw_scopes,
|
|
(str, bytes),
|
|
):
|
|
raise ServiceAccountError(
|
|
"Service account scope ceiling is invalid"
|
|
)
|
|
item.scope_ceiling = list(
|
|
_service_account_scopes(
|
|
principal,
|
|
(str(scope) for scope in raw_scopes),
|
|
)
|
|
)
|
|
if "is_active" in changes:
|
|
active = bool(changes["is_active"])
|
|
item.is_active = active
|
|
account.is_active = active
|
|
membership.is_active = active
|
|
item.retired_at = None if active else utcnow()
|
|
item.revision += 1
|
|
item.updated_by_account_id = principal.account_id
|
|
try:
|
|
session.flush()
|
|
except IntegrityError as exc:
|
|
raise ServiceAccountConflictError(
|
|
"A service account with this name already exists"
|
|
) from exc
|
|
return item
|
|
|
|
|
|
def retire_service_account(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
service_account_id: str,
|
|
principal: ApiPrincipal,
|
|
expected_revision: int,
|
|
) -> ServiceAccount:
|
|
return update_service_account(
|
|
session,
|
|
tenant_id=tenant_id,
|
|
service_account_id=service_account_id,
|
|
principal=principal,
|
|
expected_revision=expected_revision,
|
|
changes={"is_active": False},
|
|
)
|
|
|
|
|
|
def _service_account_name(value: str) -> str:
|
|
clean = " ".join(value.split())
|
|
if not 1 <= len(clean) <= 255:
|
|
raise ServiceAccountError(
|
|
"Service account name must contain between 1 and 255 characters"
|
|
)
|
|
return clean
|
|
|
|
|
|
def _normalized_name(value: str) -> str:
|
|
return value.casefold()
|
|
|
|
|
|
def _optional_text(value: str | None) -> str | None:
|
|
if value is None:
|
|
return None
|
|
clean = value.strip()
|
|
if len(clean) > 4000:
|
|
raise ServiceAccountError(
|
|
"Service account description is too long"
|
|
)
|
|
return clean or None
|
|
|
|
|
|
def _service_account_scopes(
|
|
principal: ApiPrincipal,
|
|
values: Iterable[str],
|
|
) -> tuple[str, ...]:
|
|
scopes = tuple(
|
|
sorted(
|
|
{
|
|
str(value).strip()
|
|
for value in values
|
|
if str(value).strip()
|
|
}
|
|
)
|
|
)
|
|
if len(scopes) > 200:
|
|
raise ServiceAccountError(
|
|
"Service accounts support at most 200 scope grants"
|
|
)
|
|
denied = tuple(
|
|
scope for scope in scopes
|
|
if not principal.has(scope)
|
|
)
|
|
if denied:
|
|
raise PermissionError(
|
|
"Cannot grant service-account scopes outside the current "
|
|
f"administrator authority: {', '.join(denied)}"
|
|
)
|
|
return scopes
|
|
|
|
|
|
__all__ = [
|
|
"ServiceAccountConflictError",
|
|
"ServiceAccountError",
|
|
"ServiceAccountNotFoundError",
|
|
"create_service_account",
|
|
"get_service_account",
|
|
"list_service_accounts",
|
|
"retire_service_account",
|
|
"update_service_account",
|
|
]
|