622 lines
19 KiB
Python
622 lines
19 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
|
|
from pydantic import BaseModel, Field
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.auth import ApiPrincipal, get_api_principal
|
|
from govoplan_core.core.ownership import (
|
|
OwnershipAuthorizationError,
|
|
OwnershipIdempotencyConflict,
|
|
OwnershipResourceRef,
|
|
OwnershipSubjectRef,
|
|
OwnershipTransfer,
|
|
OwnershipTransferError,
|
|
OwnershipTransferExpired,
|
|
OwnershipTransferPolicy,
|
|
accept_ownership_transfer,
|
|
approve_administrative_recovery,
|
|
approve_ownership_request,
|
|
cancel_ownership_transfer,
|
|
decline_ownership_transfer,
|
|
execute_administrative_recovery,
|
|
request_ownership,
|
|
start_administrative_recovery,
|
|
start_owner_initiated_transfer,
|
|
)
|
|
from govoplan_core.core.registry import PlatformRegistry
|
|
from govoplan_core.db.session import get_session
|
|
|
|
|
|
RECENT_AUTHENTICATION_WINDOW = timedelta(minutes=15)
|
|
|
|
|
|
class OwnershipSubjectRequest(BaseModel):
|
|
type: str = Field(min_length=1, max_length=40, pattern=r"^[a-z][a-z0-9_-]*$")
|
|
id: str = Field(min_length=1, max_length=255)
|
|
|
|
|
|
class OwnershipResourceRequest(BaseModel):
|
|
module_id: str = Field(
|
|
min_length=1,
|
|
max_length=100,
|
|
pattern=r"^[a-z][a-z0-9_-]*$",
|
|
)
|
|
resource_type: str = Field(
|
|
min_length=1,
|
|
max_length=100,
|
|
pattern=r"^[a-z][a-z0-9_-]*$",
|
|
)
|
|
resource_id: str = Field(min_length=1, max_length=255)
|
|
|
|
|
|
class OwnershipTransferStartRequest(BaseModel):
|
|
resource: OwnershipResourceRequest
|
|
target_owner: OwnershipSubjectRequest
|
|
idempotency_key: str = Field(min_length=1, max_length=200)
|
|
reason: str | None = Field(default=None, max_length=4000)
|
|
expiry_days: int | None = Field(default=None, ge=1, le=30)
|
|
|
|
|
|
class OwnershipRequestStartRequest(BaseModel):
|
|
resource: OwnershipResourceRequest
|
|
target_owner: OwnershipSubjectRequest | None = None
|
|
idempotency_key: str = Field(min_length=1, max_length=200)
|
|
reason: str | None = Field(default=None, max_length=4000)
|
|
expiry_days: int | None = Field(default=None, ge=1, le=30)
|
|
|
|
|
|
class OwnershipRecoveryStartRequest(BaseModel):
|
|
resource: OwnershipResourceRequest
|
|
target_owner: OwnershipSubjectRequest
|
|
idempotency_key: str = Field(min_length=1, max_length=200)
|
|
reason: str = Field(min_length=1, max_length=4000)
|
|
|
|
|
|
class OwnershipTransferResponse(BaseModel):
|
|
id: str
|
|
tenant_id: str
|
|
resource_module: str
|
|
resource_type: str
|
|
resource_id: str
|
|
kind: str
|
|
status: str
|
|
current_owner: OwnershipSubjectRequest
|
|
target_owner: OwnershipSubjectRequest
|
|
initiated_by: OwnershipSubjectRequest
|
|
owner_approved_by: OwnershipSubjectRequest | None = None
|
|
target_accepted_by: OwnershipSubjectRequest | None = None
|
|
reason: str | None = None
|
|
assurance_profile: str | None = None
|
|
required_approvals: int
|
|
approvals: list[dict[str, Any]] = Field(default_factory=list)
|
|
decisions: list[dict[str, Any]] = Field(default_factory=list)
|
|
expires_at: datetime
|
|
execute_after: datetime | None = None
|
|
completed_at: datetime | None = None
|
|
declined_at: datetime | None = None
|
|
cancelled_at: datetime | None = None
|
|
expired_at: datetime | None = None
|
|
revision: int
|
|
metadata: dict[str, Any] = Field(default_factory=dict)
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
|
|
class OwnershipTransferListResponse(BaseModel):
|
|
transfers: list[OwnershipTransferResponse] = Field(default_factory=list)
|
|
|
|
|
|
router = APIRouter(prefix="/ownership/transfers", tags=["ownership"])
|
|
|
|
|
|
@router.get("", response_model=OwnershipTransferListResponse)
|
|
def list_ownership_transfers(
|
|
request: Request,
|
|
resource_type: str | None = Query(default=None, max_length=100),
|
|
resource_id: str | None = Query(default=None, max_length=255),
|
|
transfer_status: str | None = Query(default=None, alias="status", max_length=50),
|
|
limit: int = Query(default=50, ge=1, le=200),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
session: Session = Depends(get_session),
|
|
) -> OwnershipTransferListResponse:
|
|
actor = _actor(principal)
|
|
query = session.query(OwnershipTransfer).filter(
|
|
OwnershipTransfer.tenant_id == principal.tenant_id
|
|
)
|
|
if resource_type:
|
|
query = query.filter(
|
|
OwnershipTransfer.resource_type == resource_type.strip().lower()
|
|
)
|
|
if resource_id:
|
|
query = query.filter(OwnershipTransfer.resource_id == resource_id)
|
|
if transfer_status:
|
|
query = query.filter(OwnershipTransfer.status == transfer_status)
|
|
rows = (
|
|
query.order_by(
|
|
OwnershipTransfer.created_at.desc(),
|
|
OwnershipTransfer.id.desc(),
|
|
)
|
|
.limit(min(limit * 5, 1000))
|
|
.all()
|
|
)
|
|
visible = [
|
|
row
|
|
for row in rows
|
|
if _can_view_transfer(request, session, row, actor)
|
|
]
|
|
return OwnershipTransferListResponse(
|
|
transfers=[_response(row) for row in visible[:limit]]
|
|
)
|
|
|
|
|
|
@router.get("/{transfer_id}", response_model=OwnershipTransferResponse)
|
|
def get_ownership_transfer(
|
|
transfer_id: str,
|
|
request: Request,
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
session: Session = Depends(get_session),
|
|
) -> OwnershipTransferResponse:
|
|
transfer = _visible_transfer(
|
|
request,
|
|
session,
|
|
principal,
|
|
transfer_id,
|
|
)
|
|
return _response(transfer)
|
|
|
|
|
|
@router.post(
|
|
"",
|
|
response_model=OwnershipTransferResponse,
|
|
status_code=status.HTTP_201_CREATED,
|
|
)
|
|
def create_owner_initiated_transfer(
|
|
payload: OwnershipTransferStartRequest,
|
|
request: Request,
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
session: Session = Depends(get_session),
|
|
) -> OwnershipTransferResponse:
|
|
return _write(
|
|
session,
|
|
lambda: start_owner_initiated_transfer(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
resource=_resource(payload.resource),
|
|
provider=_provider(request, payload.resource),
|
|
actor=_actor(principal),
|
|
target_owner=_subject(payload.target_owner),
|
|
idempotency_key=payload.idempotency_key,
|
|
reason=payload.reason,
|
|
expiry_days=payload.expiry_days,
|
|
),
|
|
)
|
|
|
|
|
|
@router.post(
|
|
"/requests",
|
|
response_model=OwnershipTransferResponse,
|
|
status_code=status.HTTP_201_CREATED,
|
|
)
|
|
def create_ownership_request(
|
|
payload: OwnershipRequestStartRequest,
|
|
request: Request,
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
session: Session = Depends(get_session),
|
|
) -> OwnershipTransferResponse:
|
|
return _write(
|
|
session,
|
|
lambda: request_ownership(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
resource=_resource(payload.resource),
|
|
provider=_provider(request, payload.resource),
|
|
actor=_actor(principal),
|
|
target_owner=(
|
|
_subject(payload.target_owner) if payload.target_owner else None
|
|
),
|
|
idempotency_key=payload.idempotency_key,
|
|
reason=payload.reason,
|
|
expiry_days=payload.expiry_days,
|
|
),
|
|
)
|
|
|
|
|
|
@router.post(
|
|
"/recoveries",
|
|
response_model=OwnershipTransferResponse,
|
|
status_code=status.HTTP_201_CREATED,
|
|
)
|
|
def create_administrative_recovery(
|
|
payload: OwnershipRecoveryStartRequest,
|
|
request: Request,
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
session: Session = Depends(get_session),
|
|
) -> OwnershipTransferResponse:
|
|
_require_recovery_authority(principal)
|
|
return _write(
|
|
session,
|
|
lambda: start_administrative_recovery(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
resource=_resource(payload.resource),
|
|
provider=_provider(request, payload.resource),
|
|
actor=_actor(principal),
|
|
target_owner=_subject(payload.target_owner),
|
|
idempotency_key=payload.idempotency_key,
|
|
reason=payload.reason,
|
|
policy=OwnershipTransferPolicy(),
|
|
),
|
|
)
|
|
|
|
|
|
@router.post("/{transfer_id}/owner-approval", response_model=OwnershipTransferResponse)
|
|
def approve_requested_transfer(
|
|
transfer_id: str,
|
|
request: Request,
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
session: Session = Depends(get_session),
|
|
) -> OwnershipTransferResponse:
|
|
return _mutate_existing(
|
|
request,
|
|
session,
|
|
principal,
|
|
transfer_id,
|
|
approve_ownership_request,
|
|
)
|
|
|
|
|
|
@router.post("/{transfer_id}/acceptance", response_model=OwnershipTransferResponse)
|
|
def accept_transfer(
|
|
transfer_id: str,
|
|
request: Request,
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
session: Session = Depends(get_session),
|
|
) -> OwnershipTransferResponse:
|
|
return _mutate_existing(
|
|
request,
|
|
session,
|
|
principal,
|
|
transfer_id,
|
|
accept_ownership_transfer,
|
|
)
|
|
|
|
|
|
@router.post("/{transfer_id}/decline", response_model=OwnershipTransferResponse)
|
|
def decline_transfer(
|
|
transfer_id: str,
|
|
request: Request,
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
session: Session = Depends(get_session),
|
|
) -> OwnershipTransferResponse:
|
|
return _mutate_existing(
|
|
request,
|
|
session,
|
|
principal,
|
|
transfer_id,
|
|
decline_ownership_transfer,
|
|
)
|
|
|
|
|
|
@router.post("/{transfer_id}/cancel", response_model=OwnershipTransferResponse)
|
|
def cancel_transfer(
|
|
transfer_id: str,
|
|
request: Request,
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
session: Session = Depends(get_session),
|
|
) -> OwnershipTransferResponse:
|
|
return _mutate_existing(
|
|
request,
|
|
session,
|
|
principal,
|
|
transfer_id,
|
|
cancel_ownership_transfer,
|
|
)
|
|
|
|
|
|
@router.post(
|
|
"/{transfer_id}/recovery-approval",
|
|
response_model=OwnershipTransferResponse,
|
|
)
|
|
def approve_recovery(
|
|
transfer_id: str,
|
|
request: Request,
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
session: Session = Depends(get_session),
|
|
) -> OwnershipTransferResponse:
|
|
_require_recovery_authority(principal)
|
|
return _mutate_existing(
|
|
request,
|
|
session,
|
|
principal,
|
|
transfer_id,
|
|
approve_administrative_recovery,
|
|
)
|
|
|
|
|
|
@router.post(
|
|
"/{transfer_id}/recovery-execution",
|
|
response_model=OwnershipTransferResponse,
|
|
)
|
|
def execute_recovery(
|
|
transfer_id: str,
|
|
request: Request,
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
session: Session = Depends(get_session),
|
|
) -> OwnershipTransferResponse:
|
|
_require_recovery_authority(principal)
|
|
return _mutate_existing(
|
|
request,
|
|
session,
|
|
principal,
|
|
transfer_id,
|
|
execute_administrative_recovery,
|
|
)
|
|
|
|
|
|
def _mutate_existing(
|
|
request: Request,
|
|
session: Session,
|
|
principal: ApiPrincipal,
|
|
transfer_id: str,
|
|
operation,
|
|
) -> OwnershipTransferResponse:
|
|
transfer = _tenant_transfer(
|
|
session,
|
|
principal,
|
|
transfer_id,
|
|
for_update=True,
|
|
)
|
|
provider = _provider_for_transfer(request, transfer)
|
|
return _write(
|
|
session,
|
|
lambda: operation(
|
|
session,
|
|
transfer=transfer,
|
|
provider=provider,
|
|
actor=_actor(principal),
|
|
),
|
|
)
|
|
|
|
|
|
def _write(session: Session, operation) -> OwnershipTransferResponse:
|
|
try:
|
|
transfer = operation()
|
|
session.commit()
|
|
session.refresh(transfer)
|
|
return _response(transfer)
|
|
except OwnershipAuthorizationError as exc:
|
|
session.rollback()
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=str(exc),
|
|
) from exc
|
|
except OwnershipIdempotencyConflict as exc:
|
|
session.rollback()
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail=str(exc),
|
|
) from exc
|
|
except OwnershipTransferExpired as exc:
|
|
session.commit()
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail=str(exc),
|
|
) from exc
|
|
except OwnershipTransferError as exc:
|
|
session.rollback()
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail=str(exc),
|
|
) from exc
|
|
|
|
|
|
def _tenant_transfer(
|
|
session: Session,
|
|
principal: ApiPrincipal,
|
|
transfer_id: str,
|
|
*,
|
|
for_update: bool = False,
|
|
) -> OwnershipTransfer:
|
|
query = session.query(OwnershipTransfer).filter(
|
|
OwnershipTransfer.id == transfer_id,
|
|
OwnershipTransfer.tenant_id == principal.tenant_id,
|
|
)
|
|
if for_update:
|
|
query = query.with_for_update()
|
|
transfer = query.one_or_none()
|
|
if transfer is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Ownership transfer not found",
|
|
)
|
|
return transfer
|
|
|
|
|
|
def _visible_transfer(
|
|
request: Request,
|
|
session: Session,
|
|
principal: ApiPrincipal,
|
|
transfer_id: str,
|
|
) -> OwnershipTransfer:
|
|
transfer = _tenant_transfer(session, principal, transfer_id)
|
|
if not _can_view_transfer(request, session, transfer, _actor(principal)):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Ownership transfer not found",
|
|
)
|
|
return transfer
|
|
|
|
|
|
def _can_view_transfer(
|
|
request: Request,
|
|
session: Session,
|
|
transfer: OwnershipTransfer,
|
|
actor: OwnershipSubjectRef,
|
|
) -> bool:
|
|
provider = _provider_for_transfer(request, transfer)
|
|
decision = provider.authorize_ownership_action(
|
|
session,
|
|
tenant_id=transfer.tenant_id,
|
|
resource_id=transfer.resource_id,
|
|
action="view_transfer",
|
|
actor=actor,
|
|
current_owner=OwnershipSubjectRef(
|
|
type=transfer.current_owner_type,
|
|
id=transfer.current_owner_id,
|
|
),
|
|
target_owner=OwnershipSubjectRef(
|
|
type=transfer.target_owner_type,
|
|
id=transfer.target_owner_id,
|
|
),
|
|
)
|
|
return decision.allowed
|
|
|
|
|
|
def _provider(request: Request, resource: OwnershipResourceRequest):
|
|
registry = _registry(request)
|
|
registration = registry.ownership_provider_registration(resource.resource_type)
|
|
if (
|
|
registration is None
|
|
or registration.module_id != resource.module_id
|
|
):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Owned resource type is not available",
|
|
)
|
|
return registration.provider
|
|
|
|
|
|
def _provider_for_transfer(request: Request, transfer: OwnershipTransfer):
|
|
return _provider(
|
|
request,
|
|
OwnershipResourceRequest(
|
|
module_id=transfer.resource_module,
|
|
resource_type=transfer.resource_type,
|
|
resource_id=transfer.resource_id,
|
|
),
|
|
)
|
|
|
|
|
|
def _registry(request: Request) -> PlatformRegistry:
|
|
registry = getattr(request.app.state, "govoplan_registry", None)
|
|
if not isinstance(registry, PlatformRegistry):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="Platform registry is not available",
|
|
)
|
|
return registry
|
|
|
|
|
|
def _actor(principal: ApiPrincipal) -> OwnershipSubjectRef:
|
|
actor_id = principal.membership_id
|
|
if not actor_id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Ownership actions require a tenant membership",
|
|
)
|
|
return OwnershipSubjectRef(
|
|
type="user",
|
|
id=actor_id,
|
|
label=principal.display_name,
|
|
scopes=principal.scopes,
|
|
group_ids=principal.group_ids,
|
|
recently_authenticated=_recently_authenticated(principal),
|
|
)
|
|
|
|
|
|
def _recently_authenticated(principal: ApiPrincipal) -> bool:
|
|
auth_session = principal.auth_session
|
|
created_at = getattr(auth_session, "created_at", None)
|
|
if not isinstance(created_at, datetime):
|
|
return False
|
|
if created_at.tzinfo is None:
|
|
created_at = created_at.replace(tzinfo=timezone.utc)
|
|
elapsed = datetime.now(timezone.utc) - created_at
|
|
return timedelta(0) <= elapsed <= RECENT_AUTHENTICATION_WINDOW
|
|
|
|
|
|
def _require_recovery_authority(principal: ApiPrincipal) -> None:
|
|
if not (
|
|
principal.has("admin:settings:write")
|
|
or principal.has("system:settings:write")
|
|
):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Administrative ownership recovery authority is required",
|
|
)
|
|
|
|
|
|
def _resource(payload: OwnershipResourceRequest) -> OwnershipResourceRef:
|
|
return OwnershipResourceRef(
|
|
module_id=payload.module_id,
|
|
resource_type=payload.resource_type,
|
|
resource_id=payload.resource_id,
|
|
)
|
|
|
|
|
|
def _subject(payload: OwnershipSubjectRequest) -> OwnershipSubjectRef:
|
|
return OwnershipSubjectRef(type=payload.type, id=payload.id)
|
|
|
|
|
|
def _optional_subject(
|
|
subject_type: str | None,
|
|
subject_id: str | None,
|
|
) -> OwnershipSubjectRequest | None:
|
|
if not subject_type or not subject_id:
|
|
return None
|
|
return OwnershipSubjectRequest(type=subject_type, id=subject_id)
|
|
|
|
|
|
def _response(transfer: OwnershipTransfer) -> OwnershipTransferResponse:
|
|
return OwnershipTransferResponse(
|
|
id=transfer.id,
|
|
tenant_id=transfer.tenant_id,
|
|
resource_module=transfer.resource_module,
|
|
resource_type=transfer.resource_type,
|
|
resource_id=transfer.resource_id,
|
|
kind=transfer.kind,
|
|
status=transfer.status,
|
|
current_owner=OwnershipSubjectRequest(
|
|
type=transfer.current_owner_type,
|
|
id=transfer.current_owner_id,
|
|
),
|
|
target_owner=OwnershipSubjectRequest(
|
|
type=transfer.target_owner_type,
|
|
id=transfer.target_owner_id,
|
|
),
|
|
initiated_by=OwnershipSubjectRequest(
|
|
type=transfer.initiated_by_type,
|
|
id=transfer.initiated_by_id,
|
|
),
|
|
owner_approved_by=_optional_subject(
|
|
transfer.owner_approved_by_type,
|
|
transfer.owner_approved_by_id,
|
|
),
|
|
target_accepted_by=_optional_subject(
|
|
transfer.target_accepted_by_type,
|
|
transfer.target_accepted_by_id,
|
|
),
|
|
reason=transfer.reason,
|
|
assurance_profile=transfer.assurance_profile,
|
|
required_approvals=transfer.required_approvals,
|
|
approvals=list(transfer.approvals or []),
|
|
decisions=list(transfer.decisions or []),
|
|
expires_at=transfer.expires_at,
|
|
execute_after=transfer.execute_after,
|
|
completed_at=transfer.completed_at,
|
|
declined_at=transfer.declined_at,
|
|
cancelled_at=transfer.cancelled_at,
|
|
expired_at=transfer.expired_at,
|
|
revision=transfer.revision,
|
|
metadata=dict(transfer.metadata_ or {}),
|
|
created_at=transfer.created_at,
|
|
updated_at=transfer.updated_at,
|
|
)
|
|
|
|
|
|
__all__ = ["router"]
|