File size: 3,692 Bytes
a10e62e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 | import logging
import os
from typing import Any, Dict, List
from atom_security.analyzers.static import StaticAnalyzer
from fastapi import APIRouter, Body, Depends, HTTPException
from pydantic import BaseModel
from core.admin_endpoints import get_super_admin
from core.base_routes import BaseAPIRouter
from core.models import User
from core.skill_builder_service import SkillMetadata, skill_builder_service
logger = logging.getLogger(__name__)
router = BaseAPIRouter(prefix="/api/admin/skills", tags=["Admin Skills"])
class CreateSkillRequest(BaseModel):
name: str
description: str
instructions: str
capabilities: List[str] = []
scripts: Dict[str, str] # filename -> content
@router.post("/")
async def create_new_skill(
request: CreateSkillRequest,
admin: User = Depends(get_super_admin)
):
"""
Create a new standardized skill package (Skill Skill).
"""
try:
# Use the admin's tenant ID
# In a real SaaS, this might be the admin's organization or a specific target tenant
# For now, we assume the admin creates skills for their own tenant context
tenant_id = str(admin.tenant_id) if admin.tenant_id else "default"
# Proactive Security Audit
try:
from atom_security.analyzers.llm import LLMAnalyzer
from atom_security.analyzers.static import StaticAnalyzer
# Static Scan
static_analyzer = StaticAnalyzer()
combined_content = f"{request.instructions}\n" + "\n".join(request.scripts.values())
static_findings = static_analyzer.scan_content(combined_content)
# Optional LLM Scan (BYOK or Local)
llm_findings = []
if os.getenv("ATOM_SECURITY_ENABLE_LLM_SCAN", "false").lower() == "true":
try:
llm_analyzer = LLMAnalyzer(mode=os.getenv("ATOM_SECURITY_LLM_MODE", "local"))
llm_findings = await llm_analyzer.analyze(request.name, combined_content)
except Exception as e:
logger.error(f"LLM Scan failed: {e}")
all_findings = static_findings + llm_findings
critical_findings = [f.dict() for f in all_findings if f.severity.value in ["HIGH", "CRITICAL"]]
if critical_findings:
raise router.permission_denied_error(
action="create_skill",
resource="Skill",
details={
"message": "Skill rejected due to security policy violations.",
"findings": critical_findings
}
)
except HTTPException:
raise
except Exception as scan_error:
# Log but don't block if security module fails
logger.warning(f"Security scan error: {scan_error}")
metadata = SkillMetadata(
name=request.name,
description=request.description,
instructions=request.instructions,
capabilities=request.capabilities,
author=admin.email or "Admin"
)
result = skill_builder_service.create_skill_package(
tenant_id=tenant_id,
metadata=metadata,
scripts=request.scripts
)
if not result["success"]:
raise router.validation_error("skill_creation", result["message"])
return router.success_response(
data=result,
message="Skill created successfully"
)
except HTTPException:
raise
except Exception as e:
raise router.internal_error(str(e))
|