test new file guard and auto deploy

backend/example_scripts/README.md

@@ -0,0 +1,23 @@
+# Example NFQUEUE Scripts
+
+Files in this folder are treated as protected example scripts by the API:
+
+- `*.py`: script source
+- `*-requirements.txt`: optional pip requirements copied and installed into the script venv
+- `*.deploy.json`: optional deployment settings for startup auto-deploy
+
+Protected behavior:
+
+- scripts are synced from this folder into `/srv/fw-scripts` on backend startup
+- scripts in this folder cannot be overwritten, disabled, or deleted via the API
+- scripts with a deploy config containing `qnum` are auto-started as systemd services
+
+Example deploy file:
+
+```json
+{
+  "qnum": 1,
+  "enable_at_boot": true,
+  "extra_args": "--log-level INFO"
+}
+```

backend/example_scripts/hello_nfqueue-requirements.txt

@@ -0,0 +1 @@
+netfilterqueue

backend/example_scripts/hello_nfqueue.deploy.json

@@ -0,0 +1,4 @@
+{
+  "qnum": 1,
+  "enable_at_boot": true
+}

backend/example_scripts/hello_nfqueue.py

@@ -0,0 +1,68 @@
+#!/usr/bin/env python3
+"""Minimal NFQUEUE example script.
+
+Expected argv:
+  argv[1] = queue number
+Any extra args are optional.
+"""
+
+import logging
+import signal
+import sys
+from typing import Optional
+
+try:
+    from netfilterqueue import NetfilterQueue
+except Exception as exc:  # pragma: no cover
+    print(f"Failed to import netfilterqueue: {exc}", file=sys.stderr)
+    sys.exit(2)
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
+logger = logging.getLogger("hello-nfqueue")
+
+_running = True
+
+
+def _stop(_sig: int, _frame: Optional[object]) -> None:
+    global _running
+    _running = False
+
+
+def _handle_packet(packet) -> None:
+    # This demo accepts all packets and logs basic metadata.
+    logger.info("packet id=%s len=%s", packet.get_id(), len(packet.get_payload()))
+    packet.accept()
+
+
+def main() -> int:
+    if len(sys.argv) < 2:
+        print("Usage: hello_nfqueue.py <qnum> [extra args...]", file=sys.stderr)
+        return 1
+
+    try:
+        qnum = int(sys.argv[1])
+    except ValueError:
+        print("qnum must be an integer", file=sys.stderr)
+        return 1
+
+    signal.signal(signal.SIGINT, _stop)
+    signal.signal(signal.SIGTERM, _stop)
+
+    nfq = NetfilterQueue()
+    logger.info("Binding to NFQUEUE %d", qnum)
+    nfq.bind(qnum, _handle_packet)
+
+    try:
+        while _running:
+            nfq.run(block=True)
+    except KeyboardInterrupt:
+        pass
+    finally:
+        logger.info("Unbinding NFQUEUE %d", qnum)
+        nfq.unbind()
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

backend/src/api/packet_scripting_api.py

@@ -15,6 +15,7 @@ Endpoints:
 import os
 import sys
 import re
+import json
 import shutil
 import subprocess
 import time
@@ -28,12 +29,15 @@ from pydantic import BaseModel, Field
 
 # ---------- Configuration ----------
 SCRIPT_DIR = "/srv/fw-scripts"
+REPO_BACKEND_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
+EXAMPLE_SCRIPT_DIR = os.path.join(REPO_BACKEND_DIR, "example_scripts")
 VENV_BASE = "/srv/fw-scripts/venvs"
 UNIT_DIR = "/etc/systemd/system"   # retained for writing new units, but discovery uses systemctl
 UNIT_PREFIX = "fw-script"
 
 # ensure dirs exist
 os.makedirs(SCRIPT_DIR, exist_ok=True)
+os.makedirs(EXAMPLE_SCRIPT_DIR, exist_ok=True)
 os.makedirs(VENV_BASE, exist_ok=True)
 
 # ---------- Logging ----------
@@ -65,6 +69,18 @@ def requirements_path_for(name: str) -> str:
     return os.path.join(SCRIPT_DIR, f"{name}-requirements.txt")
 
 
+def example_script_path_for(name: str) -> str:
+    return os.path.join(EXAMPLE_SCRIPT_DIR, f"{name}.py")
+
+
+def example_requirements_path_for(name: str) -> str:
+    return os.path.join(EXAMPLE_SCRIPT_DIR, f"{name}-requirements.txt")
+
+
+def example_deploy_config_path_for(name: str) -> str:
+    return os.path.join(EXAMPLE_SCRIPT_DIR, f"{name}.deploy.json")
+
+
 def venv_path_for(name: str) -> str:
     return os.path.join(VENV_BASE, name)
 
@@ -86,6 +102,122 @@ def unit_path_for_name(service_name: str) -> str:
     return os.path.join(UNIT_DIR, service_name + ".service")
 
 
+def is_example_script(name: str) -> bool:
+    return os.path.exists(example_script_path_for(name))
+
+
+def assert_not_example_script(name: str, action: str) -> None:
+    if is_example_script(name):
+        raise HTTPException(status_code=403, detail=f"example script '{name}' is protected and cannot be {action} via API")
+
+
+def _read_example_deploy_config(name: str) -> Dict[str, Any]:
+    cfg_path = example_deploy_config_path_for(name)
+    if not os.path.exists(cfg_path):
+        return {}
+    try:
+        with open(cfg_path, "r", encoding="utf-8") as fh:
+            data = json.load(fh)
+        if not isinstance(data, dict):
+            logger.warning("Deploy config for %s is not an object: %s", name, cfg_path)
+            return {}
+        return data
+    except Exception:
+        logger.exception("Failed reading deploy config for %s at %s", name, cfg_path)
+        return {}
+
+
+def deploy_example_scripts_on_startup() -> None:
+    """
+    Sync example scripts from EXAMPLE_SCRIPT_DIR into SCRIPT_DIR and auto-deploy units
+    if an optional '<name>.deploy.json' contains a queue number.
+
+    Optional config schema per script:
+      {
+        "qnum": 42,
+        "extra_args": "--flag value",
+                "enable_at_boot": true,
+                "service_name": "fw-script-<name>-q<qnum>"
+      }
+        service_name is optional and must start with 'fw-script-'.
+    """
+    if not os.path.isdir(EXAMPLE_SCRIPT_DIR):
+        return
+
+    for fn in os.listdir(EXAMPLE_SCRIPT_DIR):
+        if not fn.endswith(".py"):
+            continue
+
+        name = fn.rsplit(".", 1)[0]
+        src_script = os.path.join(EXAMPLE_SCRIPT_DIR, fn)
+        dst_script = script_path_for(name)
+
+        try:
+            shutil.copy2(src_script, dst_script)
+            os.chmod(dst_script, 0o700)
+            logger.info("Synced example script %s -> %s", src_script, dst_script)
+        except Exception:
+            logger.exception("Failed syncing example script %s", src_script)
+            continue
+
+        src_req = example_requirements_path_for(name)
+        dst_req = requirements_path_for(name)
+        if os.path.exists(src_req):
+            try:
+                shutil.copy2(src_req, dst_req)
+                logger.info("Synced example requirements %s -> %s", src_req, dst_req)
+                pip_install_requirements(name, dst_req)
+            except Exception:
+                logger.exception("Failed syncing/installing requirements for example %s", name)
+
+        cfg = _read_example_deploy_config(name)
+        qnum_raw = cfg.get("qnum")
+        if qnum_raw is None:
+            continue
+
+        try:
+            qnum = int(qnum_raw)
+        except (TypeError, ValueError):
+            logger.warning("Invalid qnum in deploy config for %s: %r", name, qnum_raw)
+            continue
+
+        if qnum < 0 or qnum > 65535:
+            logger.warning("Out-of-range qnum in deploy config for %s: %d", name, qnum)
+            continue
+
+        default_service_name = make_service_name(name, qnum)
+        configured_service_name = cfg.get("service_name")
+        if configured_service_name:
+            if str(configured_service_name).startswith(UNIT_PREFIX + "-"):
+                service_name = str(configured_service_name)
+            else:
+                logger.warning(
+                    "Ignoring invalid service_name for example %s: %r (must start with '%s-'). Falling back to %s",
+                    name,
+                    configured_service_name,
+                    UNIT_PREFIX,
+                    default_service_name,
+                )
+                service_name = default_service_name
+        else:
+            service_name = default_service_name
+        extra_args = cfg.get("extra_args") or ""
+        enable_at_boot = bool(cfg.get("enable_at_boot", True))
+
+        python_path = venv_python_for(name)
+        exec_start = f"{python_path} {dst_script} {qnum}"
+        if extra_args:
+            exec_start += " " + str(extra_args)
+
+        try:
+            write_unit(service_name, exec_start, description=f"FW example script {name} queue {qnum}", enable_at_boot=enable_at_boot)
+            time.sleep(0.05)
+            start_unit(service_name)
+            logger.info("Auto-deployed example script %s as %s", name, service_name)
+        except Exception:
+            logger.exception("Failed auto-deploying example script %s", name)
+
+
 # ---------- systemd interaction (systemctl-based, no fallback) ----------
 def _systemctl_unit_name(unit: str) -> str:
     """Return unit with .service suffix if missing."""
@@ -377,9 +509,13 @@ class ScriptWithStatus(ScriptInfo):
     Represents a script plus discovered unit mappings (if any).
     - mappings: list of UnitMapping for that script
     - requirements_exists: whether a requirements file exists on disk
+    - is_protected_example: script originates from repository example folder and is immutable via API
+    - requirements_is_protected_example: requirements file originates from repository example folder and is immutable via API
     """
     mappings: List[UnitMapping] = Field(default_factory=list)
     requirements_exists: bool = False
+    is_protected_example: bool = False
+    requirements_is_protected_example: bool = False
 
 
 class ScriptUploadResponse(ScriptInfo):
@@ -447,6 +583,8 @@ def list_scripts_with_status() -> List[ScriptWithStatus]:
       - name, path
       - mappings: list of UnitMapping (service, parsed ExecStart, active flag)
       - requirements_exists: boolean
+      - is_protected_example: boolean
+      - requirements_is_protected_example: boolean
     """
     out: List[ScriptWithStatus] = []
 
@@ -469,7 +607,16 @@ def list_scripts_with_status() -> List[ScriptWithStatus]:
             mappings.append(UnitMapping(service=svc, parsed=parsed, active=active))
 
         req_exists = os.path.exists(requirements_path_for(name))
-        out.append(ScriptWithStatus(name=name, path=spath, mappings=mappings, requirements_exists=req_exists))
+        out.append(
+            ScriptWithStatus(
+                name=name,
+                path=spath,
+                mappings=mappings,
+                requirements_exists=req_exists,
+                is_protected_example=is_example_script(name),
+                requirements_is_protected_example=os.path.exists(example_requirements_path_for(name)),
+            )
+        )
 
     logger.debug("Listed %d scripts with status", len(out))
     return out
@@ -493,6 +640,8 @@ async def upload_script(
     except ValueError as e:
         logger.warning("Invalid name provided: %s", name)
         raise HTTPException(status_code=400, detail=str(e))
+
+    assert_not_example_script(name, "overwritten")
     
     spath = script_path_for(name)
     if os.path.exists(spath):
@@ -566,7 +715,7 @@ async def upload_script(
             pass
         raise HTTPException(status_code=500, detail="internal error during upload")
 
-    resp = {"name": name, "path": spath}
+    resp: Dict[str, Any] = {"name": name, "path": spath}
     if pip_output is not None:
         resp["pip"] = pip_output
     return ScriptUploadResponse(**resp)
@@ -579,6 +728,10 @@ async def upload_script(
         200: {
             "content": {"application/octet-stream": {"schema": {"type": "string", "format": "binary"}}},
             "description": "Python script file (binary).",
+            "headers": {
+                "X-Script-Is-Protected-Example": {"schema": {"type": "string"}, "description": "'true' when script is a protected example"},
+                "X-Requirements-Is-Protected-Example": {"schema": {"type": "string"}, "description": "'true' when requirements file is a protected example"},
+            },
         },
         404: {"description": "Not found"},
         400: {"description": "Invalid name"},
@@ -599,8 +752,19 @@ def download_script(name: str) -> FileResponse:
         raise HTTPException(status_code=404, detail="not found")
     logger.info("Download script %s", name)
 
+    is_protected = is_example_script(name)
+    req_is_protected = os.path.exists(example_requirements_path_for(name))
+
     # Return generic octet-stream so clients treat as binary blob.
-    return FileResponse(path, media_type="application/octet-stream", filename=f"{name}.py")
+    return FileResponse(
+        path,
+        media_type="application/octet-stream",
+        filename=f"{name}.py",
+        headers={
+            "X-Script-Is-Protected-Example": "true" if is_protected else "false",
+            "X-Requirements-Is-Protected-Example": "true" if req_is_protected else "false",
+        },
+    )
 
 
 @router.get(
@@ -610,6 +774,10 @@ def download_script(name: str) -> FileResponse:
         200: {
             "content": {"application/octet-stream": {"schema": {"type": "string", "format": "binary"}}},
             "description": "requirements.txt file (binary).",
+            "headers": {
+                "X-Script-Is-Protected-Example": {"schema": {"type": "string"}, "description": "'true' when script is a protected example"},
+                "X-Requirements-Is-Protected-Example": {"schema": {"type": "string"}, "description": "'true' when requirements file is a protected example"},
+            },
         },
         404: {"description": "Not found"},
         400: {"description": "Invalid name"},
@@ -627,7 +795,17 @@ def download_requirements(name: str) -> FileResponse:
     if not os.path.exists(path):
         raise HTTPException(status_code=404, detail="requirements not found")
     logger.info("Download requirements for %s", name)
-    return FileResponse(path, media_type="application/octet-stream", filename=f"{name}-requirements.txt")
+    is_protected = is_example_script(name)
+    req_is_protected = os.path.exists(example_requirements_path_for(name))
+    return FileResponse(
+        path,
+        media_type="application/octet-stream",
+        filename=f"{name}-requirements.txt",
+        headers={
+            "X-Script-Is-Protected-Example": "true" if is_protected else "false",
+            "X-Requirements-Is-Protected-Example": "true" if req_is_protected else "false",
+        },
+    )
 
 
 @router.put(
@@ -650,6 +828,8 @@ async def upload_requirements_install(name: str, requirements: UploadFile = File
     except ValueError as e:
         raise HTTPException(status_code=400, detail=str(e))
 
+    assert_not_example_script(name, "modified")
+
     # ensure script exists (we don't allow attaching requirements to non-existing script)
     spath = script_path_for(name)
     if not os.path.exists(spath):
@@ -698,6 +878,8 @@ def delete_requirements_and_cleanup(name: str) -> Dict[str, Any]:
     except ValueError as e:
         raise HTTPException(status_code=400, detail=str(e))
 
+    assert_not_example_script(name, "modified")
+
     spath = script_path_for(name)
     if not os.path.exists(spath):
         raise HTTPException(status_code=404, detail="script not found")
@@ -744,6 +926,8 @@ def enable_script(name: str, req: EnableRequest) -> OperationResult:
     except ValueError as e:
         raise HTTPException(status_code=400, detail=str(e))
 
+    assert_not_example_script(name, "manually enabled")
+
     script_path = script_path_for(name)
     if not os.path.exists(script_path):
         raise HTTPException(status_code=404, detail="script not found")
@@ -784,6 +968,8 @@ def disable_script(name: str, qnum: int) -> OperationResult:
         validate_name(name)
     except ValueError as e:
         raise HTTPException(status_code=400, detail=str(e))
+
+    assert_not_example_script(name, "disabled")
     service_name = make_service_name(name, qnum)
     # attempt stop + remove via systemctl-based remove_unit
     try:
@@ -816,6 +1002,8 @@ def delete_script(name: str, qnum: Optional[int] = Query(None, description="If g
     except ValueError as e:
         raise HTTPException(status_code=400, detail=str(e))
 
+    assert_not_example_script(name, "deleted")
+
     removed_units: List[str] = []
     failed_units: List[str] = []
     errors: List[str] = []
@@ -895,6 +1083,11 @@ def delete_script(name: str, qnum: Optional[int] = Query(None, description="If g
 
 # ---------- Lifecycle helper ----------
 def register_lifecycle(app) -> None:
+    @app.on_event("startup")
+    def _startup_event() -> None:
+        logger.info("Startup: syncing and deploying protected example scripts from %s", EXAMPLE_SCRIPT_DIR)
+        deploy_example_scripts_on_startup()
+
     @app.on_event("shutdown")
     def _shutdown_event() -> None:
         logger.info("Shutdown: stopping/removing manager-created units with prefix %s", UNIT_PREFIX)