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:
@@ -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
@@ -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)
|
||||||
|
|||||||
Reference in New Issue
Block a user