mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
Renn F
92294f90fb
Route layer now rejects 403 if the role doesn't match the router's allowed set. Choreographer still re-checks role per verb where needed, but defense in depth means a future verb that forgets the role check doesn't leak. Auditor router also gated.
45 lines
1.1 KiB
Python
45 lines
1.1 KiB
Python
"""Auditor intent-verb HTTP endpoints. Read-only.
|
|
|
|
Thin handlers; delegate to Choreographer.
|
|
"""
|
|
|
|
from typing import Annotated
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter, Depends, Header
|
|
|
|
from roboco.api.deps import get_choreographer
|
|
from roboco.api.routes.v2._role_dep import require_auditor
|
|
from roboco.api.schemas.v2.flow import IAmIdleRequest, TriageRequest
|
|
from roboco.services.gateway.choreographer import Choreographer
|
|
|
|
router = APIRouter(
|
|
prefix="/api/v2/flow/auditor",
|
|
tags=["v2-flow-auditor"],
|
|
dependencies=[require_auditor],
|
|
)
|
|
|
|
|
|
_AgentIdHeader = Annotated[UUID, Header(alias="X-Agent-ID")]
|
|
_ChoreographerDep = Annotated[Choreographer, Depends(get_choreographer)]
|
|
|
|
|
|
@router.post("/triage")
|
|
async def triage(
|
|
_body: TriageRequest,
|
|
x_agent_id: _AgentIdHeader,
|
|
choreographer: _ChoreographerDep,
|
|
) -> dict:
|
|
env = await choreographer.auditor_triage(x_agent_id)
|
|
return env.as_dict()
|
|
|
|
|
|
@router.post("/i_am_idle")
|
|
async def i_am_idle(
|
|
_body: IAmIdleRequest,
|
|
x_agent_id: _AgentIdHeader,
|
|
choreographer: _ChoreographerDep,
|
|
) -> dict:
|
|
env = await choreographer.i_am_idle(x_agent_id)
|
|
return env.as_dict()
|