"""REST surface for Cloud-managed skills: admin management + host-scoped sync. Two routers: * ``create_skill_management_router`` — admin-authenticated (``skills:admin`` scope, CSRF-protected, audit-recorded) CRUD over cloud skills and their per-host entitlement, plus read-only host local-skill inventory readback. * ``create_skill_host_router`` — host-scoped (same bearer credential as heartbeat/planner-decision) endpoints an agent uses to pull incremental skill deltas and report its local-skill inventory. """ from __future__ import annotations import json from collections.abc import Callable from uuid import uuid4 from fastapi import APIRouter, HTTPException, Query, Request, status from cloud.auth import ( AuthProvider, HostAuthorizationError, Principal, SKILLS_ADMIN_SCOPE, ) from cloud.observability import current_correlation_id from cloud.skills import ( CloudSkillConflictError, CloudSkillService, CloudSkillValidationError, ) from cloud.sdk.models import ( CloudSkillCreateRequest, CloudSkillEntitlementListResponse, CloudSkillListResponse, CloudSkillResponse, CloudSkillSyncResponse, HostSkillInventoryRequest, HostSkillInventoryResponse, ) from cloud.user_auth import AuthAuditEvent from core.models import utc_now # ---------------------------------------------------------------------- # Admin management router # ---------------------------------------------------------------------- def create_skill_management_router( *, service: CloudSkillService, repository, auth_provider: AuthProvider, csrf_validator: Callable[[Request, Principal], bool], version_prefix: str = "/v1", ) -> APIRouter: router = APIRouter(prefix=version_prefix, tags=["skill-management"]) def authorize(request: Request) -> Principal: principal = auth_provider.authenticate(request) if principal is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="unauthorized", headers={"WWW-Authenticate": "Bearer"}, ) if principal.must_change_password or not principal.has_scope( SKILLS_ADMIN_SCOPE ): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"missing required scope: {SKILLS_ADMIN_SCOPE}", ) return principal def require_csrf(request: Request, principal: Principal) -> None: if not csrf_validator(request, principal): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="CSRF validation failed", ) @router.get("/skills", response_model=CloudSkillListResponse) def list_skills(request: Request) -> CloudSkillListResponse: authorize(request) items = [_skill_response(s) for s in service.list_skills()] return CloudSkillListResponse(items=items) @router.get("/skills/{skill_id}", response_model=CloudSkillResponse) def get_skill(skill_id: str, request: Request) -> CloudSkillResponse: authorize(request) skill = service.get_skill(skill_id) if skill is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Skill not found" ) return _skill_response(skill) @router.post( "/skills", response_model=CloudSkillResponse, status_code=status.HTTP_201_CREATED, ) def create_skill( payload: CloudSkillCreateRequest, request: Request, ) -> CloudSkillResponse: principal = authorize(request) require_csrf(request, principal) try: skill = service.create_skill( name=payload.name, kind=payload.kind, description=payload.description, tags=payload.tags, content=payload.content, steps_json=json.dumps(payload.steps), parameters_json=json.dumps(payload.parameters), now=utc_now(), ) except CloudSkillValidationError as exc: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc) ) from exc except CloudSkillConflictError as exc: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=str(exc) ) from exc _audit(repository, principal, skill.id, "cloud_skill_create") return _skill_response(skill) @router.patch("/skills/{skill_id}", response_model=CloudSkillResponse) def update_skill( skill_id: str, payload: CloudSkillCreateRequest, request: Request, ) -> CloudSkillResponse: principal = authorize(request) require_csrf(request, principal) try: skill = service.update_skill( skill_id, name=payload.name, kind=payload.kind, description=payload.description, tags=payload.tags, content=payload.content, steps_json=json.dumps(payload.steps), parameters_json=json.dumps(payload.parameters), now=utc_now(), ) except CloudSkillValidationError as exc: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Skill not found" ) from exc except CloudSkillConflictError as exc: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=str(exc) ) from exc _audit(repository, principal, skill.id, "cloud_skill_update") return _skill_response(skill) @router.delete("/skills/{skill_id}", status_code=status.HTTP_204_NO_CONTENT) def delete_skill(skill_id: str, request: Request) -> None: principal = authorize(request) require_csrf(request, principal) try: service.delete_skill(skill_id) except KeyError as exc: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Skill not found" ) from exc _audit(repository, principal, skill_id, "cloud_skill_delete") @router.get( "/skills/{skill_id}/entitlements", response_model=CloudSkillEntitlementListResponse, ) def list_entitlements( skill_id: str, request: Request ) -> CloudSkillEntitlementListResponse: authorize(request) return CloudSkillEntitlementListResponse( skill_id=skill_id, host_ids=service.list_hosts_for_skill(skill_id), ) @router.post( "/skills/{skill_id}/entitlements/{host_id}", status_code=status.HTTP_204_NO_CONTENT, ) def grant_entitlement(skill_id: str, host_id: str, request: Request) -> None: principal = authorize(request) require_csrf(request, principal) try: service.grant_entitlement(skill_id, host_id, now=utc_now()) except CloudSkillValidationError as exc: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=str(exc) ) from exc _audit(repository, principal, skill_id, "cloud_skill_entitlement_grant") @router.delete( "/skills/{skill_id}/entitlements/{host_id}", status_code=status.HTTP_204_NO_CONTENT, ) def revoke_entitlement(skill_id: str, host_id: str, request: Request) -> None: principal = authorize(request) require_csrf(request, principal) service.revoke_entitlement(skill_id, host_id, now=utc_now()) _audit(repository, principal, skill_id, "cloud_skill_entitlement_revoke") @router.get( "/hosts/{host_id}/skill-inventory", response_model=HostSkillInventoryResponse, ) def get_host_inventory( host_id: str, request: Request ) -> HostSkillInventoryResponse: authorize(request) entry = service.get_host_inventory(host_id) if entry is None: return HostSkillInventoryResponse(host_id=host_id) return HostSkillInventoryResponse( host_id=entry.host_id, payload=json.loads(entry.payload_json or "[]"), reported_at=entry.reported_at, ) return router # ---------------------------------------------------------------------- # Host-scoped router (agent sync + inventory report) # ---------------------------------------------------------------------- def create_skill_host_router( *, service: CloudSkillService, auth_provider: AuthProvider, version_prefix: str = "/internal/v1", ) -> APIRouter: router = APIRouter(prefix=version_prefix, tags=["skill-sync"]) def authorize_host(request: Request, host_id: str) -> None: principal = auth_provider.authenticate(request) if principal is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="unauthorized", headers={"WWW-Authenticate": "Bearer"}, ) try: principal.require_host(host_id) except HostAuthorizationError as exc: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=str(exc) ) from exc @router.get("/hosts/{host_id}/skills/sync", response_model=CloudSkillSyncResponse) def sync_skills( host_id: str, request: Request, since_version: int | None = Query(default=None, ge=0), ) -> CloudSkillSyncResponse: authorize_host(request, host_id) delta = service.fetch_host_delta(host_id, since_version) return CloudSkillSyncResponse( skills=[_skill_response(s) for s in delta.skills], removed_ids=list(delta.removed_ids), latest_version=delta.latest_version, is_full_replace=delta.is_full_replace, ) @router.post( "/hosts/{host_id}/skills/inventory", status_code=status.HTTP_204_NO_CONTENT, ) def report_inventory( host_id: str, payload: HostSkillInventoryRequest, request: Request, ) -> None: authorize_host(request, host_id) service.record_host_inventory( host_id, json.dumps(payload.skills), now=utc_now(), ) return router # ---------------------------------------------------------------------- # Helpers # ---------------------------------------------------------------------- def _skill_response(skill) -> CloudSkillResponse: return CloudSkillResponse( id=skill.id, name=skill.name, kind=skill.kind, description=skill.description, tags=list(skill.tags), revision=skill.revision, created_at=skill.created_at, updated_at=skill.updated_at, content=skill.content, steps=json.loads(skill.steps_json or "[]"), parameters=json.loads(skill.parameters_json or "{}"), ) def _audit(repository, principal: Principal, skill_id: str, action: str) -> None: repository.record_auth_audit( AuthAuditEvent( id=uuid4().hex, occurred_at=utc_now(), actor_principal_id=principal.id, target_user_id=None, action=action, outcome="success", correlation_id=current_correlation_id(), metadata={"cloud_skill_id": skill_id}, ) )