diff --git a/backend/src/api/packet_scripting_api.py b/backend/src/api/packet_scripting_api.py index 0620182..ec3f460 100644 --- a/backend/src/api/packet_scripting_api.py +++ b/backend/src/api/packet_scripting_api.py @@ -1,39 +1,40 @@ -# script_router_with_venv.py +# script_router_named.py """ -APIRouter for uploading scripts (with optional requirements.txt), -creating per-script venvs (when requirements provided), and managing -systemd services that run scripts with a queue number argument. +APIRouter: upload scripts with a supplied name, optional requirements -> create per-script venv. +If venv install fails, response includes pip output and the router deletes the uploaded files and venv. -Usage: - from fastapi import FastAPI - from script_router_with_venv import router, register_lifecycle - app = FastAPI() - app.include_router(router) - register_lifecycle(app) # optional: stops/removes manager-created units on shutdown +Endpoints: + - POST /scripts -> upload script (multipart): script file, optional requirements file, required 'name' form field + - GET /scripts -> list scripts + - GET /scripts/{name} -> download script + - POST /scripts/{name}/enable -> enable systemd service for script on given qnum + - POST /scripts/{name}/disable -> disable service for script on qnum + - GET /scripts/status -> status of all fw-script units + - GET /scripts/{name}/status -> status of units for that script Notes: - - The router does NOT interact with nft. You should create/delete nft queue rules with your separate API. - - Script files are stored under SCRIPT_DIR. - - Virtualenvs (if created) are stored under VENV_BASE/. - - Systemd units are created under /etc/systemd/system with names: fw-script--q.service - - This code must run with permissions to create venvs, write unit files and call systemctl (typically root). + - This relies on systemd and writes units to /etc/systemd/system + - Script files are stored at SCRIPT_DIR/.py + - Venv stored at VENV_BASE/ (if requirements provided) + - 'name' must match regex [A-Za-z0-9_.-]+ (no path separators) """ import os import sys +import re import uuid +import json import shutil import subprocess import time -import re import logging from typing import Optional, List, Dict -from fastapi import APIRouter, UploadFile, File, HTTPException +from fastapi import APIRouter, UploadFile, File, Form, HTTPException from fastapi.responses import FileResponse from pydantic import BaseModel -# ---------- Config ---------- +# ---------- Configuration ---------- SCRIPT_DIR = "/srv/fw-scripts" VENV_BASE = "/srv/fw-scripts/venvs" UNIT_DIR = "/etc/systemd/system" @@ -45,87 +46,91 @@ os.makedirs(VENV_BASE, exist_ok=True) # ---------- Logging ---------- logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s [%(name)s] %(message)s") -logger = logging.getLogger("script-router-venv") +logger = logging.getLogger("script-router-named") # ---------- Router ---------- router = APIRouter(prefix="/scripts", tags=["scripts"]) -# ---------- Models ---------- -class ScriptInfo(BaseModel): - id: str - name: str - path: str +# ---------- Name validation ---------- +# Accept only safe file-name characters to avoid path traversal: letters, digits, dot, underscore, hyphen +_NAME_RE = re.compile(r'^[A-Za-z0-9_.-]+$') -class EnableRequest(BaseModel): - qnum: int - service_name: Optional[str] = None - extra_args: Optional[str] = None - enable_at_boot: Optional[bool] = False +def validate_name(name: str) -> None: + if not name: + raise ValueError("name must be provided") + if not _NAME_RE.match(name): + raise ValueError("invalid name; allowed characters: letters, digits, dot, underscore, hyphen") + # prevent reserved names or dots-only + if name in (".", ".."): + raise ValueError("invalid name") -# ---------- Utilities: service names / unit paths ---------- -def make_service_name(sid: str, qnum: int) -> str: - return f"{UNIT_PREFIX}-{sid}-q{qnum}" +# ---------- Utility paths ---------- +def script_path_for(name: str) -> str: + return os.path.join(SCRIPT_DIR, f"{name}.py") + +def requirements_path_for(name: str) -> str: + return os.path.join(SCRIPT_DIR, f"{name}-requirements.txt") + +def venv_path_for(name: str) -> str: + return os.path.join(VENV_BASE, name) + +def venv_python_for(name: str) -> str: + vpy = os.path.join(venv_path_for(name), "bin", "python") + if os.path.exists(vpy): + return vpy + return "/usr/bin/python3" + +def make_service_name(name: str, qnum: int) -> str: + return f"{UNIT_PREFIX}-{name}-q{qnum}" def unit_path_for(service_name: str) -> str: return os.path.join(UNIT_DIR, service_name + ".service") # ---------- Venv helpers ---------- -def venv_path_for(sid: str) -> str: - return os.path.join(VENV_BASE, sid) - -def venv_python_for(sid: str) -> str: - vpy = os.path.join(venv_path_for(sid), "bin", "python") - if os.path.exists(vpy): - return vpy - # fallback to system python - return "/usr/bin/python3" - -def create_venv(sid: str, timeout: int = 60): - venv_dir = venv_path_for(sid) +def create_venv(name: str, timeout: int = 60) -> str: + venv_dir = venv_path_for(name) if os.path.exists(venv_dir): - logger.debug("Venv already exists for sid=%s: %s", sid, venv_dir) + logger.debug("Venv already exists for %s", name) return venv_dir os.makedirs(os.path.dirname(venv_dir), exist_ok=True) - logger.info("Creating venv for sid=%s at %s", sid, venv_dir) + logger.info("Creating venv for %s at %s", name, venv_dir) try: - subprocess.run([sys.executable, "-m", "venv", venv_dir], check=True, timeout=timeout) + subprocess.run([sys.executable, "-m", "venv", venv_dir], check=True, timeout=timeout, capture_output=True, text=True) except subprocess.CalledProcessError as e: - raise RuntimeError(f"venv creation failed: {e}") + logger.exception("venv creation failed for %s: %s", name, e.stderr if hasattr(e, "stderr") else str(e)) + raise RuntimeError("venv creation failed: " + (e.stderr or str(e))) except subprocess.TimeoutExpired: + logger.exception("venv creation timed out for %s", name) raise RuntimeError("venv creation timed out") return venv_dir -def pip_install_requirements(sid: str, requirements_path: str, timeout: int = 600) -> Dict[str, str]: +def pip_install_requirements(name: str, requirements_path: str, timeout: int = 600) -> Dict[str, str]: """ - Install requirements into the venv for sid from requirements_path. - Returns dict: { "stdout": "...", "stderr": "..." } - Raises on failure with details in exception message. + Install requirements into the venv for name from requirements_path. + Returns dict with stdout/stderr. Raises RuntimeError on failure including outputs. """ - venv_dir = create_venv(sid) + venv_dir = create_venv(name) pip_path = os.path.join(venv_dir, "bin", "pip") - # ensure pip exists and upgrade + # ensure pip exists and attempt to upgrade try: subprocess.run([pip_path, "install", "--upgrade", "pip"], check=True, capture_output=True, text=True, timeout=300) except subprocess.CalledProcessError as e: - # continue but warn - logger.warning("pip upgrade failed for sid=%s: %s", sid, e.stderr if hasattr(e, "stderr") else str(e)) - # install requirements + logger.warning("pip upgrade warning for %s: %s", name, getattr(e, "stderr", str(e))) + # run install try: p = subprocess.run([pip_path, "install", "-r", requirements_path, "--no-cache-dir"], check=True, capture_output=True, text=True, timeout=timeout) - logger.info("pip install success for sid=%s", sid) - return {"stdout": p.stdout, "stderr": p.stderr} + logger.info("pip install succeeded for %s", name) + return {"stdout": p.stdout or "", "stderr": p.stderr or ""} except subprocess.CalledProcessError as e: - logger.error("pip install failed for sid=%s: %s", sid, e.stderr if hasattr(e, "stderr") else str(e)) - # return output for debugging out = {"stdout": getattr(e, "stdout", "") or "", "stderr": getattr(e, "stderr", "") or str(e)} + logger.error("pip install failed for %s: %s", name, out["stderr"][:4000]) raise RuntimeError(jsonify_cmd_output(out)) except subprocess.TimeoutExpired: - logger.error("pip install timeout for sid=%s", sid) + logger.error("pip install timed out for %s", name) raise RuntimeError("pip install timed out") def jsonify_cmd_output(out: Dict[str, str]) -> str: - # helper to pack stdout/stderr into a single string message s = "" if out.get("stdout"): s += "STDOUT:\n" + out["stdout"] + "\n" @@ -153,7 +158,6 @@ WantedBy=multi-user.target """ with open(unit_path, "w") as fh: fh.write(unit_text) - # reload systemd subprocess.run(["systemctl", "daemon-reload"], check=True) logger.info("Wrote unit %s", unit_path) if enable_at_boot: @@ -192,18 +196,18 @@ def is_unit_active(service_name: str) -> bool: return p.returncode == 0 def list_fw_units() -> List[str]: - """Return list of fw unit names without .service suffix.""" + """List our fw-script units (without .service suffix).""" units = [] try: for fn in os.listdir(UNIT_DIR): if fn.startswith(UNIT_PREFIX + "-") and fn.endswith(".service"): - units.append(fn[:-8]) # remove .service + units.append(fn[:-8]) except FileNotFoundError: logger.warning("Unit dir %s not found", UNIT_DIR) return units -# ExecStart parse pattern: python path + script path + qnum + optional extra -_RE_EXECSTART = re.compile(r'(?P/\S*python\S*)\s+(?P