140 lines
4.5 KiB
Python
140 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Annotated, Any
|
|
|
|
from fastapi import APIRouter, Depends, Request
|
|
from sqlmodel import Session
|
|
|
|
from ..connectors.authentik import AuthentikConnector
|
|
from ..core.security import require_operator_request
|
|
from ..db.session import get_session
|
|
from ..schemas.forge_oidc import ForgeOidcProvisionRequestV1, ForgeOidcProvisionResultV1
|
|
from ..schemas.forge_secret_delivery import (
|
|
ForgeControllerReaderProvisionRequestV1,
|
|
ForgeControllerReaderProvisionResultV1,
|
|
)
|
|
from ..services.forge_controller_reader import provision_forge_controller_reader
|
|
from ..services.forge_oidc_provisioning import provision_forge_oidc
|
|
from .connector_runtime import load_connector_runtime
|
|
|
|
router = APIRouter(prefix='/api/apps/authentik')
|
|
|
|
|
|
def _shape(payload: dict[str, Any]) -> dict[str, Any]:
|
|
data = payload.get('data')
|
|
if not isinstance(data, dict):
|
|
data = {key: value for key, value in payload.items() if key not in {'success', 'status', 'message'}}
|
|
return {
|
|
**payload,
|
|
'success': bool(payload.get('success')),
|
|
'status': str(payload.get('status', 'ok')),
|
|
'message': str(payload.get('message', 'ok')),
|
|
'data': data,
|
|
}
|
|
|
|
|
|
def _connector() -> AuthentikConnector:
|
|
return AuthentikConnector()
|
|
|
|
|
|
def _require_forge_oidc_operator(request: Request) -> str:
|
|
return require_operator_request(request)
|
|
|
|
|
|
@router.get('/health', dependencies=[Depends(load_connector_runtime)])
|
|
def health():
|
|
return _shape(_connector().health())
|
|
|
|
|
|
@router.get('/system', dependencies=[Depends(load_connector_runtime)])
|
|
def system_status():
|
|
return _shape(_connector().system_status())
|
|
|
|
|
|
@router.get('/applications', dependencies=[Depends(load_connector_runtime)])
|
|
def applications():
|
|
return _shape(_connector().applications())
|
|
|
|
|
|
@router.get('/providers', dependencies=[Depends(load_connector_runtime)])
|
|
def providers():
|
|
return _shape(_connector().providers())
|
|
|
|
|
|
@router.get('/providers/proxy', dependencies=[Depends(load_connector_runtime)])
|
|
def proxy_providers():
|
|
return _shape(_connector().proxy_providers())
|
|
|
|
|
|
@router.get('/providers/oauth2', dependencies=[Depends(load_connector_runtime)])
|
|
def oauth2_providers():
|
|
return _shape(_connector().oauth2_providers())
|
|
|
|
|
|
@router.get('/outposts', dependencies=[Depends(load_connector_runtime)])
|
|
def outposts():
|
|
return _shape(_connector().outposts())
|
|
|
|
|
|
@router.get('/outposts/health', dependencies=[Depends(load_connector_runtime)])
|
|
def outpost_health():
|
|
return _shape(_connector().outpost_health())
|
|
|
|
|
|
@router.get('/flows', dependencies=[Depends(load_connector_runtime)])
|
|
def flows_summary():
|
|
return _shape(_connector().flows_summary())
|
|
|
|
|
|
@router.get('/policies', dependencies=[Depends(load_connector_runtime)])
|
|
def policies_summary():
|
|
return _shape(_connector().policies_summary())
|
|
|
|
|
|
@router.get('/events', dependencies=[Depends(load_connector_runtime)])
|
|
def events(limit: int = 25):
|
|
return _shape(_connector().events(limit))
|
|
|
|
|
|
@router.get('/auth-coverage', dependencies=[Depends(load_connector_runtime)])
|
|
def auth_coverage():
|
|
return _shape(_connector().auth_coverage())
|
|
|
|
|
|
@router.get('/security-warnings', dependencies=[Depends(load_connector_runtime)])
|
|
def security_warnings():
|
|
return _shape(_connector().security_warnings())
|
|
|
|
|
|
@router.post(
|
|
'/forge/oidc/provision',
|
|
response_model=ForgeOidcProvisionResultV1,
|
|
)
|
|
def provision_forge_oidc_application(
|
|
payload: ForgeOidcProvisionRequestV1,
|
|
request: Request,
|
|
session: Annotated[Session, Depends(get_session)],
|
|
actor_identity: Annotated[str, Depends(_require_forge_oidc_operator)],
|
|
):
|
|
"""Run the fixed, manually authorized Forge OIDC provisioning action."""
|
|
# The payload is intentionally empty: callers cannot select a provider,
|
|
# redirect URI, mapping, groups, or secret. The service repeats the exact
|
|
# policy check before making any connector request.
|
|
_ = payload
|
|
return provision_forge_oidc(session, actor_identity=actor_identity)
|
|
|
|
|
|
@router.post(
|
|
'/forge/controller-reader/provision',
|
|
response_model=ForgeControllerReaderProvisionResultV1,
|
|
)
|
|
def provision_forge_controller_reader_account(
|
|
payload: ForgeControllerReaderProvisionRequestV1,
|
|
request: Request,
|
|
session: Annotated[Session, Depends(get_session)],
|
|
actor_identity: Annotated[str, Depends(_require_forge_oidc_operator)],
|
|
):
|
|
"""Provision only the fixed read-only Forge controller service account."""
|
|
_ = payload, request
|
|
return provision_forge_controller_reader(session, actor_identity=actor_identity)
|