60 lines
1.9 KiB
Python
60 lines
1.9 KiB
Python
"""Authenticated, project-scoped policy simulation and audit read routes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
|
|
|
from packages.hub_policy import InvocationContext
|
|
|
|
from .service import PolicyService
|
|
|
|
|
|
def _deny_by_default(_: Request) -> None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN, detail="policy authorization required"
|
|
)
|
|
|
|
|
|
def build_policy_router(
|
|
service: PolicyService, *, authorize: Callable[[Request], Any] = _deny_by_default
|
|
) -> APIRouter:
|
|
def authorized(request: Request) -> None:
|
|
authorize(request)
|
|
|
|
router = APIRouter(
|
|
prefix="/governance", tags=["governance"], dependencies=[Depends(authorized)]
|
|
)
|
|
|
|
@router.post("/simulate")
|
|
def simulate(context: InvocationContext) -> dict[str, Any]:
|
|
return service.simulate(context)
|
|
|
|
@router.get("/audit")
|
|
def audit(
|
|
request: Request, workspace: str, project: str, limit: int = 100, offset: int = 0
|
|
) -> dict[str, Any]:
|
|
del request
|
|
try:
|
|
return service.audit_search(
|
|
workspace=workspace, project=project, limit=limit, offset=offset
|
|
)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=422, detail=str(exc)) from exc
|
|
|
|
@router.get("/audit/export-metadata")
|
|
def export_metadata(workspace: str, project: str) -> dict[str, Any]:
|
|
return service.audit_export_metadata(workspace=workspace, project=project)
|
|
|
|
@router.put("/policies/{slug}")
|
|
def update_policy(slug: str, payload: dict[str, Any]) -> None:
|
|
del slug, payload
|
|
try:
|
|
service.update_policy()
|
|
except PermissionError as exc:
|
|
raise HTTPException(status_code=405, detail=str(exc)) from exc
|
|
|
|
return router
|