agentic-mobile-control/packages/cloud-platform/cloud/sdk/skill_api.py

"""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},
        )
    )