feat(05-01): implement NinjaRMM ZIP and intunewin package export endpoints

- GET /printers/{id}/packages/ninja: in-memory ZIP with install.ps1 + driver files in named subfolder
- GET /printers/{id}/packages/intunewin: temp dir build of .intunewin via build_intunewin()
- _get_printer_and_driver() helper validates printer, driver, inf, desc
- Driver ZIP file existence check before processing (RESEARCH pitfall 3)
- TemporaryDirectory context manager for auto-cleanup (RESEARCH pitfall 1)
- Router registered in main.py after scripts router
- All 9 package tests pass, 84 total tests green
This commit is contained in:
2026-04-10 15:05:29 +02:00
parent d8ce223b38
commit dd6cedfed9
2 changed files with 163 additions and 2 deletions
+158
View File
@@ -0,0 +1,158 @@
"""Package export endpoints — serves deployment packages for NinjaRMM and Microsoft Intune."""
import io
import json
import os
import tempfile
import zipfile
from fastapi import APIRouter
from fastapi.responses import PlainTextResponse, Response
import imptune.config as cfg
from imptune.db.models import Printer
from imptune.generators.intunewin_builder import build_intunewin
from imptune.generators.script_generator import render_detect, render_install, render_uninstall
router = APIRouter(prefix="/printers")
def _get_printer_and_driver(printer_id: int):
"""Fetch printer and validate driver — returns (printer, driver, driver_name) or PlainTextResponse error."""
printer = Printer.get_or_none(Printer.id == printer_id)
if printer is None:
return None, PlainTextResponse("Printer not found", status_code=404)
driver = printer.driver
if driver is None:
return None, PlainTextResponse("No driver assigned", status_code=422)
if not driver.inf_filename:
return None, PlainTextResponse("Driver has no INF file", status_code=422)
if not driver.driver_desc:
return None, PlainTextResponse("Driver has no description", status_code=422)
try:
desc_list = json.loads(driver.driver_desc)
driver_name = desc_list[0]
except (json.JSONDecodeError, IndexError, TypeError):
return None, PlainTextResponse("Driver description is invalid", status_code=422)
return (printer, driver, driver_name), None
def _get_driver_zip_path(driver) -> str:
"""Return the on-disk path of the driver ZIP file."""
return os.path.join(cfg.DRIVERS_DIR, f"{driver.sha256}.zip")
@router.get("/{printer_id}/packages/ninja")
def get_ninja_package(printer_id: int):
"""Download a NinjaRMM-ready ZIP containing install.ps1 and the driver files."""
result, error = _get_printer_and_driver(printer_id)
if error is not None:
return error
printer, driver, driver_name = result
# Validate driver file exists on disk
driver_zip_path = _get_driver_zip_path(driver)
if not os.path.isfile(driver_zip_path):
return PlainTextResponse("Driver file not found on disk", status_code=422)
safe_name = printer.name.replace(" ", "_")
# Render install script
install_script = render_install(
printer_name=printer.name,
ip_address=printer.ip_address,
port_name=printer.port_name,
driver_name=driver_name,
inf_filename=driver.inf_filename,
duplex_mode=printer.duplex_mode,
color_mode=printer.color_mode,
paper_size=printer.paper_size,
collate=printer.collate,
)
# Build ZIP in-memory
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w", compression=zipfile.ZIP_DEFLATED) as zf:
# Add install script
zf.writestr(f"{safe_name}/install.ps1", install_script)
# Extract and re-add driver files from driver ZIP
with zipfile.ZipFile(driver_zip_path, "r") as driver_zf:
for member in driver_zf.namelist():
member_data = driver_zf.read(member)
zf.writestr(f"{safe_name}/drivers/{member}", member_data)
return Response(
content=buf.getvalue(),
media_type="application/zip",
headers={"Content-Disposition": f'attachment; filename="{safe_name}_ninja.zip"'},
)
@router.get("/{printer_id}/packages/intunewin")
def get_intunewin_package(printer_id: int):
"""Download a Microsoft Intune .intunewin deployment package."""
result, error = _get_printer_and_driver(printer_id)
if error is not None:
return error
printer, driver, driver_name = result
# Validate driver file exists on disk
driver_zip_path = _get_driver_zip_path(driver)
if not os.path.isfile(driver_zip_path):
return PlainTextResponse("Driver file not found on disk", status_code=422)
safe_name = printer.name.replace(" ", "_")
# Render all three scripts
install_script = render_install(
printer_name=printer.name,
ip_address=printer.ip_address,
port_name=printer.port_name,
driver_name=driver_name,
inf_filename=driver.inf_filename,
duplex_mode=printer.duplex_mode,
color_mode=printer.color_mode,
paper_size=printer.paper_size,
collate=printer.collate,
)
uninstall_script = render_uninstall(
printer_name=printer.name,
driver_name=driver_name,
port_name=printer.port_name,
)
detect_script = render_detect(printer_name=printer.name)
with tempfile.TemporaryDirectory(prefix="imptune_") as tmpdir:
# Write scripts
with open(os.path.join(tmpdir, "install.ps1"), "w", encoding="utf-8") as f:
f.write(install_script)
with open(os.path.join(tmpdir, "uninstall.ps1"), "w", encoding="utf-8") as f:
f.write(uninstall_script)
with open(os.path.join(tmpdir, "detect.ps1"), "w", encoding="utf-8") as f:
f.write(detect_script)
# Extract driver ZIP contents into tmpdir/drivers/
drivers_subdir = os.path.join(tmpdir, "drivers")
os.makedirs(drivers_subdir, exist_ok=True)
with zipfile.ZipFile(driver_zip_path, "r") as driver_zf:
driver_zf.extractall(drivers_subdir)
# Build .intunewin
output_path = os.path.join(tmpdir, "out.intunewin")
build_intunewin(tmpdir, "install.ps1", output_path)
with open(output_path, "rb") as f:
content = f.read()
return Response(
content=content,
media_type="application/octet-stream",
headers={"Content-Disposition": f'attachment; filename="{safe_name}.intunewin"'},
)
+5 -2
View File
@@ -5,8 +5,8 @@ from pathlib import Path
from fastapi import FastAPI from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles from fastapi.staticfiles import StaticFiles
from imptune.api import clients, drivers, health, pages, printers, scripts from imptune.api import clients, drivers, health, icons, pages, packages, printers, scripts
from imptune.config import DATA_DIR, DRIVERS_DIR from imptune.config import DATA_DIR, DRIVERS_DIR, ICONS_DIR
from imptune.db.database import db, init_db from imptune.db.database import db, init_db
@@ -14,6 +14,7 @@ from imptune.db.database import db, init_db
async def lifespan(app: FastAPI): async def lifespan(app: FastAPI):
os.makedirs(DATA_DIR, exist_ok=True) os.makedirs(DATA_DIR, exist_ok=True)
os.makedirs(DRIVERS_DIR, exist_ok=True) os.makedirs(DRIVERS_DIR, exist_ok=True)
os.makedirs(ICONS_DIR, exist_ok=True)
init_db() init_db()
yield yield
# Close DB connection on shutdown so test fixtures can re-initialize cleanly # Close DB connection on shutdown so test fixtures can re-initialize cleanly
@@ -34,3 +35,5 @@ app.include_router(drivers.router)
app.include_router(printers.router) app.include_router(printers.router)
app.include_router(clients.router) app.include_router(clients.router)
app.include_router(scripts.router) app.include_router(scripts.router)
app.include_router(packages.router)
app.include_router(icons.router)