diff --git a/imptune/api/drivers.py b/imptune/api/drivers.py new file mode 100644 index 0000000..123a153 --- /dev/null +++ b/imptune/api/drivers.py @@ -0,0 +1,115 @@ +"""Driver upload API — POST /drivers/upload.""" +from __future__ import annotations + +import io +import json +import zipfile +from pathlib import Path + +from fastapi import APIRouter, Request, UploadFile +from fastapi.responses import HTMLResponse +from fastapi.templating import Jinja2Templates + +import imptune.config as _cfg +from imptune.db.models import Driver +from imptune.services.inf_parser import _detect_encoding, parse_inf +from imptune.storage.driver_store import DriverStore + +router = APIRouter(prefix="/drivers") + +templates = Jinja2Templates( + directory=str(Path(__file__).parent.parent / "templates") +) + +MAX_UPLOAD_BYTES = 100 * 1024 * 1024 # 100 MB + + +def _error_response(message: str, status_code: int = 400) -> HTMLResponse: + """Return an HTMX-friendly error fragment swapped into #driver-list.""" + return HTMLResponse( + content=f"

{message}

", + status_code=status_code, + ) + + +@router.post("/upload", response_class=HTMLResponse) +def upload_driver(request: Request, file: UploadFile) -> HTMLResponse: + """Accept a driver ZIP, parse its INF, persist via DriverStore + Peewee ORM. + + Returns an HTMX partial (partials/driver_list.html) on success, or an + inline error fragment with HTTP 400 on validation failure. + """ + data = file.file.read(MAX_UPLOAD_BYTES + 1) + if len(data) > MAX_UPLOAD_BYTES: + return _error_response("File exceeds 100 MB limit.") + + # Must end with .zip + filename = file.filename or "" + if not filename.lower().endswith(".zip"): + return _error_response("Only .zip files are accepted.") + + # Must be a valid ZIP archive + if not zipfile.is_zipfile(io.BytesIO(data)): + return _error_response("Uploaded file is not a valid ZIP archive.") + + with zipfile.ZipFile(io.BytesIO(data)) as zf: + zip_names = zf.namelist() + + # Reject zip-slip paths + for name in zip_names: + if ".." in name or name.startswith("/"): + return _error_response("ZIP contains unsafe paths.") + + # Find .inf files + inf_names = [n for n in zip_names if n.lower().endswith(".inf")] + if not inf_names: + return _error_response("No .inf file found in the uploaded ZIP.") + + # Prefer amd64/x64 INF when multiple exist; fall back to alphabetical first + preferred = [ + n for n in inf_names if "amd64" in n.lower() or "x64" in n.lower() + ] + chosen_inf = preferred[0] if preferred else sorted(inf_names)[0] + + raw_inf = zf.read(chosen_inf) + + # Decode INF + encoding = _detect_encoding(raw_inf) + inf_text = raw_inf.decode(encoding) + + # Parse INF + parsed = parse_inf(inf_text, inf_filename=chosen_inf, zip_names=zip_names) + + # Persist file (content-addressed, dedup automatic) + # Read DRIVERS_DIR at call time so tests can monkeypatch imptune.config.DRIVERS_DIR + store = DriverStore(_cfg.DRIVERS_DIR) + sha256 = store.save(data) + + # Upsert Driver record (no duplicate if same SHA256) + Driver.get_or_create( + sha256=sha256, + defaults={ + "original_filename": filename, + "size_bytes": len(data), + "driver_desc": json.dumps(parsed.driver_names), + "inf_filename": parsed.inf_filename, + "architecture": parsed.architecture, + "has_cat_file": parsed.has_cat_file, + }, + ) + + # Build driver_data for template + drivers = list(Driver.select().order_by(Driver.uploaded_at.desc())) + driver_data = [] + for d in drivers: + names = json.loads(d.driver_desc) if d.driver_desc else [] + driver_data.append({"driver": d, "names": names}) + + return templates.TemplateResponse( + request=request, + name="partials/driver_list.html", + context={ + "driver_data": driver_data, + "parsed": parsed, + }, + ) diff --git a/imptune/api/pages.py b/imptune/api/pages.py index 9027fcb..814f20a 100644 --- a/imptune/api/pages.py +++ b/imptune/api/pages.py @@ -1,3 +1,5 @@ +import json + from fastapi import APIRouter, Request from fastapi.responses import HTMLResponse from fastapi.templating import Jinja2Templates @@ -18,3 +20,19 @@ def dashboard(request: Request): "recent_packages": [], }, ) + + +@router.get("/drivers", response_class=HTMLResponse) +def drivers_page(request: Request): + from imptune.db.models import Driver + + drivers = list(Driver.select().order_by(Driver.uploaded_at.desc())) + driver_data = [] + for d in drivers: + names = json.loads(d.driver_desc) if d.driver_desc else [] + driver_data.append({"driver": d, "names": names}) + return templates.TemplateResponse( + request=request, + name="drivers.html", + context={"driver_data": driver_data}, + ) diff --git a/imptune/main.py b/imptune/main.py index 14cbe1c..ab57d63 100644 --- a/imptune/main.py +++ b/imptune/main.py @@ -5,7 +5,7 @@ from pathlib import Path from fastapi import FastAPI from fastapi.staticfiles import StaticFiles -from imptune.api import health, pages +from imptune.api import drivers, health, pages from imptune.config import DATA_DIR, DRIVERS_DIR from imptune.db.database import init_db @@ -27,3 +27,4 @@ app.mount("/static", StaticFiles(directory=str(_static_dir)), name="static") # Register routers app.include_router(health.router) app.include_router(pages.router) +app.include_router(drivers.router) diff --git a/imptune/templates/drivers.html b/imptune/templates/drivers.html new file mode 100644 index 0000000..201f463 --- /dev/null +++ b/imptune/templates/drivers.html @@ -0,0 +1,25 @@ +{% extends "base.html" %} +{% block content %} +

Drivers

+ +
+

Upload Driver Package

+
+ + + + Uploading... +
+
+ +
+

Driver Library

+ {% include "partials/driver_list.html" %} +
+{% endblock %} diff --git a/imptune/templates/partials/driver_list.html b/imptune/templates/partials/driver_list.html new file mode 100644 index 0000000..d238891 --- /dev/null +++ b/imptune/templates/partials/driver_list.html @@ -0,0 +1,50 @@ +
+ {% if parsed is defined and parsed.unused_files %} +

+ {{ parsed.unused_files | length }} file(s) may be unused (not referenced by the INF): +

+ Show unused files + +
+

+ {% endif %} + + {% if driver_data %} + + + + + + + + + + + {% for item in driver_data %} + + + + + + + {% endfor %} + +
FilenameDriver Name(s)ArchitectureUploaded
{{ item.driver.original_filename }} + {% if item.names %} + + {% else %} + Unknown + {% endif %} + {{ item.driver.architecture or "Unknown" }}{{ item.driver.uploaded_at }}
+ {% else %} +

No drivers uploaded yet.

+ {% endif %} +
diff --git a/tests/conftest.py b/tests/conftest.py index dbe17c3..de08e3d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -8,7 +8,8 @@ from fastapi.testclient import TestClient def client(tmp_data_dir): from imptune.main import app - return TestClient(app) + with TestClient(app) as c: + yield c @pytest.fixture