From 2c068068142e03e624e83483bbf7f23b5afc4273 Mon Sep 17 00:00:00 2001 From: Kawa Date: Wed, 5 Aug 2026 10:44:47 +0200 Subject: [PATCH] feat: driver rename, driver icons, web image + driver search MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Driver rename and icons: `Driver.display_name` plus a `DriverIcon` table, both global/shared like the `Driver` row they hang off, so a rename or an icon is what every Owner sees. The rename/icon dialog keeps its forms as siblings (nested forms are invalid HTML) and the icon routes return an `hx-swap-oob` thumbnail refresh rather than re-rendering the table, which would tear the open `` out of the DOM. Web image picker: `GET /web/images` renders a pickable grid for a printer or a driver icon, with the search term prefilled from the entity name and editable. Picking one downloads it server-side and normalizes it. Driver download search: `GET /web/drivers` searches for a vendor-wide driver (the term is rewritten into the vendor's real product name for 15 brands) or for the exact model as typed. Links only — nothing is downloaded, and the fragment says the results are unvetted. Icon uploads no longer reject off-size or non-PNG files: `normalize_icon()` letterboxes any decodable raster into a 256x256 PNG. An already-exact 256x256 PNG is returned byte-identical, because icon storage is content-addressed and re-encoding would move the file on every save. `fetch_image()` makes the request from the server, so `assert_fetchable()` refuses any URL resolving to a private, loopback, or link-local address, and re-runs on every redirect. ImpTune sits on the same LAN as the printers it configures; an unguarded fetcher would be a port scanner for anyone who can reach the UI. DuckDuckGo is scraped, not called through an API — no key needed, but fragile, so both search functions swallow parse failures and return [] instead of 500ing a page. `WEB_SEARCH=false` disables every outbound request and hides the controls, for air-gapped installs. Also: one shared `Jinja2Templates` in `templating.py` instead of five per-router instances, so a template global is declared once; `_add_missing_columns()` in `database.py` adds new nullable columns to a pre-existing table, which `create_tables(safe=True)` skips; `db_env` in test_db.py now closes its connection on teardown, or the next test's ORM writes land in the previous test's DB file. Co-Authored-By: Claude Opus 5 (1M context) --- .gitignore | 3 + CLAUDE.md | 48 ++- docker-compose.yml | 4 + imptune/api/clients.py | 8 +- imptune/api/drivers.py | 158 ++++++- imptune/api/icons.py | 156 ++++--- imptune/api/pages.py | 49 ++- imptune/api/printers.py | 19 +- imptune/api/session.py | 8 +- imptune/api/web.py | 113 ++++++ imptune/config.py | 5 + imptune/db/database.py | 20 +- imptune/db/models.py | 26 ++ imptune/main.py | 14 +- imptune/services/icons.py | 93 +++++ imptune/services/image_utils.py | 50 +++ imptune/services/websearch.py | 384 ++++++++++++++++++ imptune/static/app.css | 103 ++++- imptune/templates/base.html | 58 +++ imptune/templates/drivers.html | 38 ++ .../templates/partials/driver_edit_modal.html | 109 +++++ imptune/templates/partials/driver_list.html | 30 +- .../partials/driver_search_results.html | 29 ++ imptune/templates/partials/image_results.html | 30 ++ imptune/templates/printer_detail.html | 28 +- imptune/templating.py | 17 + tests/e2e/test_driver_rename.py | 84 ++++ tests/test_db.py | 47 ++- tests/test_driver_icon.py | 227 +++++++++++ tests/test_driver_rename.py | 82 ++++ tests/test_icon_upload.py | 45 +- tests/test_image_utils.py | 66 +++ tests/test_web_routes.py | 267 ++++++++++++ tests/test_websearch.py | 260 ++++++++++++ 34 files changed, 2516 insertions(+), 162 deletions(-) create mode 100644 imptune/api/web.py create mode 100644 imptune/services/icons.py create mode 100644 imptune/services/image_utils.py create mode 100644 imptune/services/websearch.py create mode 100644 imptune/templates/partials/driver_edit_modal.html create mode 100644 imptune/templates/partials/driver_search_results.html create mode 100644 imptune/templates/partials/image_results.html create mode 100644 imptune/templating.py create mode 100644 tests/e2e/test_driver_rename.py create mode 100644 tests/test_driver_icon.py create mode 100644 tests/test_driver_rename.py create mode 100644 tests/test_image_utils.py create mode 100644 tests/test_web_routes.py create mode 100644 tests/test_websearch.py diff --git a/.gitignore b/.gitignore index 8d04241..98698a1 100644 --- a/.gitignore +++ b/.gitignore @@ -15,3 +15,6 @@ venv/ /data/ imptune_data/ *.intunewin + +# Playwright MCP session artifacts (snapshots / screenshots) +.playwright-mcp/ diff --git a/CLAUDE.md b/CLAUDE.md index 8bfd25e..b64cb69 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -43,11 +43,21 @@ ImpTune make printer deploy packages (`.intunewin` for Intune, `.zip` for NinjaR **Request flow:** 1. Driver upload → `api/drivers.py` → `services/inf_parser.py` parse INF → `storage/driver_store.py` store by SHA256 → Peewee `Driver` record (shared/global — visible to every Owner) 2. Printer config → `api/printers.py` → `db/models.py` `Printer` record (links Driver FK, scoped to `request.state.owner`) -3. Icon upload → `api/icons.py` → Pillow validate PNG 256×256 → SHA256 storage → `Icon` record +3. Icon upload → `api/icons.py` → `services/image_utils.normalize_icon()` resize any raster to 256×256 PNG → SHA256 storage → `Icon` record 4. Package export → `api/packages.py` → `generators/script_generator.py` render Jinja2 PS1 templates → `generators/intunewin_builder.py` encrypt ZIP (AES-256-CBC + HMAC-SHA256) **Key modules:** -- `imptune/config.py` — `DATA_DIR`, `DB_PATH`, `DRIVERS_DIR`, `ICONS_DIR`, `COOKIE_SECURE` from env +- `imptune/config.py` — `DATA_DIR`, `DB_PATH`, `DRIVERS_DIR`, `ICONS_DIR`, `COOKIE_SECURE`, `WEB_SEARCH` from env +- `imptune/templating.py` — the *only* `Jinja2Templates` instance; every router + imports `templates` from it. Template globals (`web_search_enabled`) are + declared once there, as callables so a monkeypatched `cfg` takes effect +- `imptune/services/image_utils.py` — `normalize_icon()`: any Pillow-decodable + raster → 256×256 PNG, letterboxed (aspect kept, transparent padding). An + already-exact 256×256 PNG is returned **byte-identical**, because icon storage + is content-addressed and re-encoding would move the file on every save +- `imptune/services/icons.py` — shared icon storage for `Icon` (printer) and + `DriverIcon`; 750 KB cap on the *source* bytes +- `imptune/services/websearch.py` — the only code that leaves the box - `imptune/db/database.py` — SQLite WAL mode + `foreign_keys=1`; all models inherit `BaseModel`; `init_db()` also backfills `owner_id` on pre-per-owner-scoping DBs into a synthetic legacy `Owner` (key written to `{DATA_DIR}/legacy_owner_key.txt`) - `imptune/services/session.py` — `OwnerSessionMiddleware` resolves `request.state.owner` from the `imptune_owner_key` cookie, creating one on first visit (skips `/health`) - `imptune/services/inf_parser.py` — auto-detect encoding (UTF-16/UTF-8/cp1252), resolve `%TOKEN%` from `[Strings]`, handle multi-model INFs @@ -56,6 +66,39 @@ ImpTune make printer deploy packages (`.intunewin` for Intune, `.zip` for NinjaR **Per-owner storage:** `Printer`/`Client` (groups) are scoped to an `Owner` identified by an opaque bearer key in a cookie — no accounts. `Driver` stays global/shared. Every route taking a `printer_id`/`client_id` must filter/check `.owner == request.state.owner` (404, not 403, on mismatch) — printer IDs are small sequential ints, so a list-only filter isn't enough. Onboarding modal (`templates/base.html`, gated on `request.state.is_new_owner`) offers "download backup key" (`GET /session/key/download`, marks `Owner.is_permanent`) vs. temporary; `/session/restore` re-attaches a browser to a previously downloaded key. In tests, use the `owner` fixture (`tests/conftest.py`) when creating `Printer`/`Client` rows directly via the ORM so the `client` fixture's cookie-scoped requests can see them. +**Web lookups (`services/websearch.py`, `api/web.py`):** DuckDuckGo is *scraped*, +not called through an API — no key, but fragile by nature, so `search_images()` +and `search_pages()` swallow parse/transport failures and return `[]` instead of +500ing a page. Image search needs a per-query `vqd` token scraped from the HTML +first, *and* the `_XHR_HEADERS` set (`Accept`, `X-Requested-With`, `Sec-Fetch-*`) +on the `i.js` call — with a valid token but no fetch metadata it answers **403**. `GET /web/images?q&target=printer|driver&id=` and `GET /web/drivers?q&mode=generic|exact` +return HTML fragments (never JSON), and ownership is checked *before* a search is +spent on the id. `fetch_image()` downloads server-side, so `assert_fetchable()` +refuses any URL resolving to a private/loopback/link-local address — ImpTune sits +on the same LAN as the printers, and an unguarded fetcher is a port scanner for +anyone who can reach the UI. Redirects re-run the guard via +`_GuardedRedirectHandler`. Driver search returns **links only** — nothing is +downloaded, and `partials/driver_search_results.html` must keep saying so. +`generic_driver_query()` maps a detected brand to that vendor's real universal-driver +product name (`GENERIC_DRIVER_TERMS`); an unknown brand falls back to +`" universal print driver download"`. + +**Driver rename + driver icons:** `Driver.display_name` (nullable) and the +`DriverIcon` table. Both are **global/shared** like `Driver` itself — a rename is +visible to every Owner, and `GET /drivers/{id}/icon` is deliberately not +owner-scoped. `PATCH /drivers/{id}` swaps the whole `#driver-list`; the icon +routes return a small status fragment *plus* an `hx-swap-oob` refresh of +`#driver-thumb-{id}`, because the dialog stays open after picking an icon and +re-rendering the table would tear the open `` out of the DOM. +`partials/driver_edit_modal.html` keeps the rename form and the icon forms as +*siblings* (nested forms are invalid HTML) — the footer's Save reaches the rename +form through `form="driver-rename-{id}"`. + +**Schema changes on an existing DB:** `create_tables(safe=True)` skips a table +that already exists, so a new field on an old model needs an entry in +`database._add_missing_columns()` — that is what puts `display_name` on a +pre-rename `driver` table. Tests: `test_db.py::test_init_db_adds_display_name_*`. + **UI stack:** Pico CSS + HTMX 2 + Alpine.js 3 + Jinja2 server-side templates. **Design layer (`static/app.css`):** a token + component layer over Pico. Tokens @@ -115,6 +158,7 @@ on narrow screens or in the add-printer sidebar (`.form-aside`). |-----|---------|---------| | `DATA_DIR` | `/data` | Storage root (DB + drivers + icons) | | `PORT` | `8000` | Server port | +| `WEB_SEARCH` | `true` | `false` disables every outbound request (image search, driver-page search, image download) and hides the search controls — `templating.py` exposes it to templates as `web_search_enabled()` | | `COOKIE_SECURE` | `true` | Three-way session mode, parsed by `config.parse_cookie_mode()` into `(COOKIE_SECURE, SINGLE_USER)`: `true` = Secure + 10-year cookie; `false` = plain-HTTP serving (browser drops a Secure cookie → new Owner per request), cookie becomes **memory-only** (no `Max-Age`); `single_user` (or `single-user`/`single`) = no cookie at all, one shared Owner — see below. | `COOKIE_SECURE=false` degrades the session instead of weakening the credential: diff --git a/docker-compose.yml b/docker-compose.yml index 8ccad74..a63112f 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -20,6 +20,10 @@ services: # Or drop sessions entirely — no cookie, one shared store for everyone who # can reach the app. Test boxes / single-person local prod only. # - COOKIE_SECURE=single_user + # Image search / driver-page search / image download are the only outbound + # requests ImpTune makes. Set false on an air-gapped host and the UI hides + # those controls instead of timing out on each one. + # - WEB_SEARCH=false volumes: imptune_data: \ No newline at end of file diff --git a/imptune/api/clients.py b/imptune/api/clients.py index 5a77a7c..49184d4 100644 --- a/imptune/api/clients.py +++ b/imptune/api/clients.py @@ -1,21 +1,15 @@ """Client CRUD API — POST /clients, GET /clients.""" from __future__ import annotations -from pathlib import Path - from fastapi import APIRouter, Form, Request from fastapi.responses import HTMLResponse -from fastapi.templating import Jinja2Templates from peewee import IntegrityError from imptune.db.models import Client +from imptune.templating import templates router = APIRouter(prefix="/clients") -templates = Jinja2Templates( - directory=str(Path(__file__).parent.parent / "templates") -) - def _error_response(message: str, status_code: int = 400) -> HTMLResponse: """Return an HTMX-friendly error fragment swapped into #client-list.""" diff --git a/imptune/api/drivers.py b/imptune/api/drivers.py index b8f154c..7574d7f 100644 --- a/imptune/api/drivers.py +++ b/imptune/api/drivers.py @@ -1,27 +1,31 @@ -"""Driver upload API — POST /drivers/upload.""" +"""Driver API — upload, rename, and driver-library icons.""" from __future__ import annotations import io import json import zipfile -from pathlib import Path - from fastapi import APIRouter, Form, Request, UploadFile -from fastapi.responses import HTMLResponse -from fastapi.templating import Jinja2Templates +from fastapi.responses import FileResponse, HTMLResponse, Response import imptune.config as _cfg -from imptune.db.models import Driver +from imptune.api.pages import build_driver_data, printer_counts_by_driver +from imptune.db.models import Driver, DriverIcon +from imptune.services import websearch +from imptune.services.icons import ( + MAX_ICON_BYTES, + ImageError, + IconTooLarge, + icon_path, + set_driver_icon, +) from imptune.services.inf_parser import _detect_encoding, parse_inf from imptune.storage.driver_store import DriverStore +from imptune.templating import templates router = APIRouter(prefix="/drivers") -templates = Jinja2Templates( - directory=str(Path(__file__).parent.parent / "templates") -) - MAX_UPLOAD_BYTES = 100 * 1024 * 1024 # 100 MB +MAX_DISPLAY_NAME = 120 def _error_response(message: str, status_code: int = 400) -> HTMLResponse: @@ -32,6 +36,19 @@ def _error_response(message: str, status_code: int = 400) -> HTMLResponse: ) +def _render_driver_list(request: Request, **extra) -> HTMLResponse: + """Re-render the whole `#driver-list` table — the target every form swaps.""" + drivers = Driver.select().order_by(Driver.uploaded_at.desc()) + context = { + "driver_data": build_driver_data(drivers), + "usage": printer_counts_by_driver(request.state.owner), + } + context.update(extra) + return templates.TemplateResponse( + request=request, name="partials/driver_list.html", context=context + ) + + @router.post("/upload", response_class=HTMLResponse) def upload_driver( request: Request, @@ -103,11 +120,7 @@ def upload_driver( ) # 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}) + driver_data = build_driver_data(Driver.select().order_by(Driver.uploaded_at.desc())) # When called from the printer form, emit primary fragment + OOB select refresh if caller == "printer_form": @@ -128,5 +141,120 @@ def upload_driver( context={ "driver_data": driver_data, "parsed": parsed, + "usage": printer_counts_by_driver(request.state.owner), }, ) + + +@router.patch("/{driver_id}", response_class=HTMLResponse) +def rename_driver( + request: Request, driver_id: int, display_name: str = Form("") +) -> HTMLResponse: + """Set or clear a driver's display name. + + Drivers are global/shared, so this rename is what every Owner sees — the + same as the driver row itself. An empty value clears the rename and the + listing falls back to the ZIP filename. + """ + driver = Driver.get_or_none(Driver.id == driver_id) + if driver is None: + return _error_response(f"Driver {driver_id} not found.", status_code=404) + + name = display_name.strip() + if len(name) > MAX_DISPLAY_NAME: + return _error_response( + f"Name must be at most {MAX_DISPLAY_NAME} characters." + ) + + driver.display_name = name or None + driver.save() + return _render_driver_list(request) + + +def _driver_icon_status(driver_id: int, sha256: str) -> HTMLResponse: + """Inline confirmation for the rename/icon dialog, plus an OOB row refresh. + + The dialog stays open after picking an icon, so the thumbnail in the table + row behind it is swapped out of band instead of re-rendering the table and + tearing the open `` out of the DOM. + """ + src = f"/drivers/{driver_id}/icon?v={sha256[:8]}" + return HTMLResponse( + content=( + '

' + "Icon saved

" + f'
' + '256×256 PNG
' + f'' + f'' + ), + status_code=200, + ) + + +def _icon_error(message: str, status_code: int = 422) -> HTMLResponse: + return HTMLResponse( + content=f"

{message}

", status_code=status_code + ) + + +@router.post("/{driver_id}/icon", response_class=HTMLResponse) +def upload_driver_icon( + request: Request, driver_id: int, file: UploadFile +) -> HTMLResponse: + """Attach an icon to a driver. Normalized to 256x256 PNG like printer icons.""" + if Driver.get_or_none(Driver.id == driver_id) is None: + return _icon_error("Driver not found.", status_code=404) + + data = file.file.read(MAX_ICON_BYTES + 1) + try: + icon = set_driver_icon(driver_id, data, file.filename or "icon.png") + except IconTooLarge as exc: + return _icon_error(str(exc)) + except ImageError: + return _icon_error("That file is not a readable image.") + + return _driver_icon_status(driver_id, icon.sha256) + + +@router.post("/{driver_id}/icon/from-web", response_class=HTMLResponse) +def driver_icon_from_web( + request: Request, driver_id: int, url: str = Form(...) +) -> HTMLResponse: + """Download a search-result image server-side and use it as the driver icon.""" + if Driver.get_or_none(Driver.id == driver_id) is None: + return _icon_error("Driver not found.", status_code=404) + + try: + data = websearch.fetch_image(url) + except websearch.WebSearchError as exc: + return _icon_error(str(exc), status_code=400) + + try: + icon = set_driver_icon( + driver_id, data, url.rsplit("/", 1)[-1][:120] or "web.png" + ) + except IconTooLarge as exc: + return _icon_error(str(exc)) + except ImageError: + return _icon_error("That URL did not return a readable image.") + + return _driver_icon_status(driver_id, icon.sha256) + + +@router.get("/{driver_id}/icon") +def get_driver_icon(driver_id: int) -> Response: + """Serve a driver icon. Not owner-scoped — the driver library is shared.""" + icon = DriverIcon.get_or_none(DriverIcon.driver == driver_id) + if icon is None: + return Response(status_code=404) + + path = icon_path(icon.sha256) + if not path.exists(): + return Response(status_code=404) + + return FileResponse( + path, + media_type="image/png", + headers={"Cache-Control": "private, max-age=300"}, + ) diff --git a/imptune/api/icons.py b/imptune/api/icons.py index 7decd0b..fbce689 100644 --- a/imptune/api/icons.py +++ b/imptune/api/icons.py @@ -1,99 +1,96 @@ -"""Icon upload API — POST /printers/{printer_id}/icon.""" +"""Printer icon API — upload, fetch-from-web, and serve.""" from __future__ import annotations -import hashlib -import io -from pathlib import Path - -from fastapi import APIRouter, Request, UploadFile +from fastapi import APIRouter, Form, Request, UploadFile from fastapi.responses import FileResponse, HTMLResponse, Response -from PIL import Image -import imptune.config as cfg from imptune.db.models import Icon, Printer +from imptune.services import websearch +from imptune.services.icons import ( + MAX_ICON_BYTES, + ImageError, + IconTooLarge, + icon_path, + set_printer_icon, +) router = APIRouter(prefix="/printers") -MAX_ICON_BYTES = 750 * 1024 # 750 KB + +def _status_fragment(printer_id: int, sha256: str) -> HTMLResponse: + """The `#icon-status` block, re-rendered after a successful save. + + `?v=` busts the 5-minute private cache the GET route sets — without it the + preview keeps showing the icon that was just replaced. + """ + return HTMLResponse( + content=( + # Swapped-in fragments are normally English-only; this one is on the + # everyday path, so it goes through the i18n store with the English + # text as its fallback body. + '

' + "Icon uploaded successfully

" + f'
' + '256×256 PNG
' + ), + status_code=200, + ) + + +def _error(message: str, status_code: int = 422) -> HTMLResponse: + return HTMLResponse(content=f"

{message}

", status_code=status_code) + + +def _owned_printer(request: Request, printer_id: int) -> Printer | None: + return Printer.get_or_none( + (Printer.id == printer_id) & (Printer.owner == request.state.owner) + ) @router.post("/{printer_id}/icon", response_class=HTMLResponse) def upload_icon(request: Request, printer_id: int, file: UploadFile) -> HTMLResponse: - """Accept a printer icon PNG, validate it, store it, and update the Icon record. + """Accept a printer icon, normalize it to 256x256 PNG, and store it. - Validation rules: - - Format: PNG only - - Dimensions: exactly 256x256 pixels - - Size: at most 750 KB - - Replaces any previously uploaded icon for this printer. - Returns an HTMX-friendly HTML fragment. + Any raster Pillow can decode is accepted and letterboxed into the box — + only an unreadable file or one over 750 KB is rejected. Replaces the + printer's previous icon. """ - # Check printer exists and belongs to this owner - printer = Printer.get_or_none( - (Printer.id == printer_id) & (Printer.owner == request.state.owner) - ) - if printer is None: - return HTMLResponse( - content="

Printer not found.

", - status_code=404, - ) + if _owned_printer(request, printer_id) is None: + return HTMLResponse(content="

Printer not found.

", status_code=404) - # Read file (read one byte extra to detect oversized files) data = file.file.read(MAX_ICON_BYTES + 1) - if len(data) > MAX_ICON_BYTES: - return HTMLResponse( - content="

Icon exceeds 750 KB limit.

", - status_code=422, - ) - - # Validate with Pillow try: - img = Image.open(io.BytesIO(data)) - except Exception: - return HTMLResponse( - content="

Icon must be PNG format.

", - status_code=422, - ) + icon = set_printer_icon(printer_id, data, file.filename or "icon.png") + except IconTooLarge as exc: + return _error(str(exc)) + except ImageError: + return _error("That file is not a readable image.") - if img.format != "PNG": - return HTMLResponse( - content="

Icon must be PNG format.

", - status_code=422, - ) + return _status_fragment(printer_id, icon.sha256) - if img.size != (256, 256): - return HTMLResponse( - content=f"

Icon must be 256x256 pixels, got {img.size}.

", - status_code=422, - ) - # Store SHA256-addressed on disk. cfg.ICONS_DIR is the same path the - # .intunewin export reads from — deriving it a second time from DATA_DIR - # would silently diverge if the layout ever changes. - sha256 = hashlib.sha256(data).hexdigest() - icons_dir = Path(cfg.ICONS_DIR) - icons_dir.mkdir(parents=True, exist_ok=True) - icon_path = icons_dir / sha256 - icon_path.write_bytes(data) +@router.post("/{printer_id}/icon/from-web", response_class=HTMLResponse) +def upload_icon_from_web( + request: Request, printer_id: int, url: str = Form(...) +) -> HTMLResponse: + """Download a search-result image server-side and use it as the icon.""" + if _owned_printer(request, printer_id) is None: + return HTMLResponse(content="

Printer not found.

", status_code=404) - # Replace existing Icon record for this printer - Icon.delete().where(Icon.printer == printer_id).execute() - Icon.create( - printer=printer_id, - sha256=sha256, - original_filename=file.filename or "icon.png", - size_bytes=len(data), - ) + try: + data = websearch.fetch_image(url) + except websearch.WebSearchError as exc: + return _error(str(exc), status_code=400) - return HTMLResponse( - content=( - '

Icon uploaded successfully

' - f'
{img.size[0]}×{img.size[1]} PNG
' - ), - status_code=200, - ) + try: + icon = set_printer_icon(printer_id, data, url.rsplit("/", 1)[-1][:120] or "web.png") + except IconTooLarge as exc: + return _error(str(exc)) + except ImageError: + return _error("That URL did not return a readable image.") + + return _status_fragment(printer_id, icon.sha256) @router.get("/{printer_id}/icon") @@ -102,22 +99,19 @@ def get_icon(request: Request, printer_id: int) -> Response: Owner-scoped: a printer belonging to another owner reads as missing. """ - printer = Printer.get_or_none( - (Printer.id == printer_id) & (Printer.owner == request.state.owner) - ) - if printer is None: + if _owned_printer(request, printer_id) is None: return Response(status_code=404) icon = Icon.get_or_none(Icon.printer == printer_id) if icon is None: return Response(status_code=404) - icon_path = Path(cfg.ICONS_DIR) / icon.sha256 - if not icon_path.exists(): + path = icon_path(icon.sha256) + if not path.exists(): return Response(status_code=404) return FileResponse( - icon_path, + path, media_type="image/png", headers={"Cache-Control": "private, max-age=300"}, ) diff --git a/imptune/api/pages.py b/imptune/api/pages.py index 9426af3..34b3057 100644 --- a/imptune/api/pages.py +++ b/imptune/api/pages.py @@ -1,17 +1,14 @@ import json from collections import defaultdict -from pathlib import Path from fastapi import APIRouter, Request from fastapi.responses import HTMLResponse -from fastapi.templating import Jinja2Templates from peewee import JOIN, fn +from imptune.templating import templates router = APIRouter() -templates = Jinja2Templates(directory=str(Path(__file__).parent.parent / "templates")) - # Group label for printers with no client — the template translates it. UNASSIGNED_GROUP = "Unassigned" @@ -57,12 +54,31 @@ def group_printers_by_client(printers) -> dict[str, list]: return ordered -def _driver_data(drivers) -> list[dict]: - """Attach the parsed driver names to each Driver row for template use.""" - return [ - {"driver": d, "names": json.loads(d.driver_desc) if d.driver_desc else []} - for d in drivers - ] +def build_driver_data(drivers) -> list[dict]: + """Attach parsed driver names + icon presence to each Driver row. + + Shared by every template that lists drivers (library table, printer form + select, edit modal) so a rename or a new icon shows up in all of them. + """ + from imptune.services.icons import driver_icon_ids, driver_search_text + from imptune.services.websearch import image_query + + drivers = list(drivers) + with_icons = driver_icon_ids(drivers) + data = [] + for d in drivers: + names = json.loads(d.driver_desc) if d.driver_desc else [] + search_text = driver_search_text(d, names) + data.append( + { + "driver": d, + "names": names, + "has_icon": d.id in with_icons, + "search_text": search_text, + "image_query": image_query(search_text), + } + ) + return data @router.get("/", response_class=HTMLResponse) @@ -109,13 +125,16 @@ def dashboard(request: Request): def drivers_page(request: Request): from imptune.db.models import Driver + from imptune.services.websearch import BRAND_SUGGESTIONS + drivers = list(Driver.select().order_by(Driver.uploaded_at.desc())) return templates.TemplateResponse( request=request, name="drivers.html", context={ - "driver_data": _driver_data(drivers), + "driver_data": build_driver_data(drivers), "usage": printer_counts_by_driver(request.state.owner), + "brands": BRAND_SUGGESTIONS, }, ) @@ -145,7 +164,7 @@ def printers_page(request: Request): context={ "grouped": grouped, "clients": clients, - "driver_data": _driver_data(all_drivers), + "driver_data": build_driver_data(all_drivers), "printer_count": printer_count, "ready_count": ready_count, }, @@ -161,13 +180,14 @@ def printers_new_page(request: Request): return templates.TemplateResponse( request=request, name="printers_new.html", - context={"clients": clients, "driver_data": _driver_data(all_drivers)}, + context={"clients": clients, "driver_data": build_driver_data(all_drivers)}, ) @router.get("/printers/{printer_id}", response_class=HTMLResponse) def printer_detail(request: Request, printer_id: int): from imptune.db.models import Client, Driver, Icon, Printer + from imptune.services.websearch import image_query printer = ( Printer.select(Printer, Client, Driver) @@ -201,6 +221,7 @@ def printer_detail(request: Request, printer_id: int): "driver_names": driver_names, "has_driver": has_driver, "has_icon": icon is not None, + "image_query": image_query(printer.name), "install_cmd": install_cmd, "uninstall_cmd": uninstall_cmd, }, @@ -251,7 +272,7 @@ def client_detail(request: Request, client_id: int): "client": client, "grouped": grouped, "clients": clients, - "driver_data": _driver_data(all_drivers), + "driver_data": build_driver_data(all_drivers), "printer_count": len(printers), "ready_count": sum(1 for p in printers if p.driver_id), }, diff --git a/imptune/api/printers.py b/imptune/api/printers.py index a0b0350..9176c6d 100644 --- a/imptune/api/printers.py +++ b/imptune/api/printers.py @@ -1,22 +1,16 @@ """Printer CRUD API — POST /printers, DELETE /printers/{id}, PATCH /printers/{id}.""" from __future__ import annotations -from pathlib import Path - from fastapi import APIRouter, Form, Request from fastapi.responses import HTMLResponse, RedirectResponse, Response -from fastapi.templating import Jinja2Templates from peewee import JOIN -from imptune.api.pages import group_printers_by_client +from imptune.api.pages import build_driver_data, group_printers_by_client from imptune.db.models import Client, Driver, Printer +from imptune.templating import templates router = APIRouter(prefix="/printers") -templates = Jinja2Templates( - directory=str(Path(__file__).parent.parent / "templates") -) - _VALID_DUPLEX = {"OneSided", "LongEdge", "ShortEdge"} _VALID_PAPER = {"A4", "Letter", "Legal"} @@ -84,8 +78,6 @@ def _resolve_driver(raw: str) -> tuple[int | None, HTMLResponse | None]: def _render_printer_list(request: Request) -> HTMLResponse: """Query printers with LEFT JOIN on client and render grouped partial.""" - import json - owner = request.state.owner query = ( Printer.select(Printer, Client) @@ -96,11 +88,8 @@ def _render_printer_list(request: Request) -> HTMLResponse: grouped = group_printers_by_client(query) clients = list(Client.select().where(Client.owner == owner).order_by(Client.name)) - all_drivers = list(Driver.select().order_by(Driver.uploaded_at.desc())) - driver_data = [ - {"driver": d, "names": json.loads(d.driver_desc) if d.driver_desc else []} - for d in all_drivers - ] + all_drivers = Driver.select().order_by(Driver.uploaded_at.desc()) + driver_data = build_driver_data(all_drivers) return templates.TemplateResponse( request=request, diff --git a/imptune/api/session.py b/imptune/api/session.py index 0336f82..05d766e 100644 --- a/imptune/api/session.py +++ b/imptune/api/session.py @@ -1,22 +1,16 @@ """Owner session routes — backup-key download and restore-on-new-browser.""" from __future__ import annotations -from pathlib import Path - from fastapi import APIRouter, Form, HTTPException, Request from fastapi.responses import HTMLResponse, PlainTextResponse, RedirectResponse -from fastapi.templating import Jinja2Templates import imptune.config as cfg from imptune.db.models import Owner from imptune.services.session import COOKIE_NAME, cookie_kwargs, is_same_origin +from imptune.templating import templates router = APIRouter(prefix="/session") -templates = Jinja2Templates( - directory=str(Path(__file__).parent.parent / "templates") -) - def _require_cookie_sessions() -> None: """404 these routes in single-user mode — keys have no meaning without a cookie. diff --git a/imptune/api/web.py b/imptune/api/web.py new file mode 100644 index 0000000..beacc11 --- /dev/null +++ b/imptune/api/web.py @@ -0,0 +1,113 @@ +"""Web lookup routes — image picker and driver-download search. + +Both return HTML fragments for HTMX, never JSON: the results are only ever +rendered into a page, and keeping the shaping server-side means the templates +own the "unverified source" warnings. +""" +from __future__ import annotations + +from fastapi import APIRouter, Request +from fastapi.responses import HTMLResponse + +from imptune.db.models import Driver, Printer +from imptune.services import websearch +from imptune.templating import templates + +router = APIRouter(prefix="/web") + +MAX_RESULTS = 12 +_TARGETS = ("printer", "driver") + + +def _notice(key: str, fallback: str, status_code: int = 200) -> HTMLResponse: + """A one-line status fragment that still speaks French. + + Server-rendered fragments are normally English-only, but these are the + everyday path (empty search box, disabled feature), so they go through the + i18n store like the rest of the picker — Alpine initializes the swapped-in + node, and the English body is the fallback if it does not. + """ + return HTMLResponse( + content=f"

{fallback}

", + status_code=status_code, + ) + + +@router.get("/images", response_class=HTMLResponse) +def search_images( + request: Request, + q: str = "", + target: str = "printer", + id: int = 0, +) -> HTMLResponse: + """Image results for `q`, each pickable as the icon of `target`/`id`. + + `target` decides which POST endpoint the result buttons hit, so the picker + is the same partial for a printer icon and a driver icon. + """ + if not websearch.enabled(): + return _notice("web_search_off", "Web lookups are disabled on this server.") + if target not in _TARGETS: + return _notice("unknown_target", "Unknown search target.", status_code=400) + + # Confirm the caller may write to this entity before spending a search on + # it — an unowned printer id must not even reveal that it exists. + if target == "printer": + owned = Printer.get_or_none( + (Printer.id == id) & (Printer.owner == request.state.owner) + ) + if owned is None: + return _notice("printer_not_found", "Printer not found.", status_code=404) + post_url = f"/printers/{id}/icon/from-web" + status_target = "#icon-status" + else: + if Driver.get_or_none(Driver.id == id) is None: + return _notice("driver_not_found", "Driver not found.", status_code=404) + post_url = f"/drivers/{id}/icon/from-web" + status_target = f"#driver-icon-status-{id}" + + query = q.strip() + if not query: + return _notice("type_to_search_image", "Type something to search for.") + + hits = websearch.search_images(query, limit=MAX_RESULTS) + return templates.TemplateResponse( + request=request, + name="partials/image_results.html", + context={ + "hits": hits, + "query": query, + "post_url": post_url, + "status_target": status_target, + }, + ) + + +@router.get("/drivers", response_class=HTMLResponse) +def search_driver_pages(request: Request, q: str = "", mode: str = "generic") -> HTMLResponse: + """Search the web for a driver download page. + + `mode=generic` rewrites the term into the vendor's universal-driver product + name (HP UPD, Xerox Global, …); `mode=exact` searches the text as typed. + Results are plain links — nothing is downloaded, and the partial says so. + """ + if not websearch.enabled(): + return _notice("web_search_off", "Web lookups are disabled on this server.") + + typed = q.strip() + if not typed: + return _notice("type_to_search_driver", "Type a printer model or brand to search for.") + + query = websearch.generic_driver_query(typed) if mode == "generic" else typed + hits = websearch.search_pages(query, limit=MAX_RESULTS) + return templates.TemplateResponse( + request=request, + name="partials/driver_search_results.html", + context={ + "hits": hits, + "query": query, + "typed": typed, + "mode": mode, + "brand": websearch.detect_brand(typed), + }, + ) diff --git a/imptune/config.py b/imptune/config.py index 24713a5..482f39f 100644 --- a/imptune/config.py +++ b/imptune/config.py @@ -30,6 +30,11 @@ def parse_cookie_mode(raw: str) -> tuple[bool, bool]: COOKIE_SECURE, SINGLE_USER = parse_cookie_mode(os.environ.get("COOKIE_SECURE", "true")) +# Outbound lookups (image search, driver-page search, image download). The only +# feature that leaves the box — set false for an air-gapped deployment and the +# UI hides every search control instead of timing out on each one. +WEB_SEARCH = os.environ.get("WEB_SEARCH", "true").strip().lower() != "false" + DB_PATH = str(Path(DATA_DIR) / "imptune.db") DRIVERS_DIR = str(Path(DATA_DIR) / "drivers") ICONS_DIR = str(Path(DATA_DIR) / "icons") diff --git a/imptune/db/database.py b/imptune/db/database.py index e7e8f47..74fe855 100644 --- a/imptune/db/database.py +++ b/imptune/db/database.py @@ -16,7 +16,7 @@ def init_db() -> None: Closes any existing connection before re-initializing so that test fixtures can monkeypatch DB_PATH between test runs. """ - from imptune.db.models import Client, Driver, Owner, Printer, Icon + from imptune.db.models import Client, Driver, DriverIcon, Owner, Printer, Icon # Re-read DB_PATH each time so tests can patch imptune.config.DB_PATH import imptune.config as cfg @@ -33,8 +33,24 @@ def init_db() -> None: }, ) db.connect(reuse_if_open=True) - db.create_tables([Owner, Client, Driver, Printer, Icon], safe=True) + db.create_tables([Owner, Client, Driver, Printer, Icon, DriverIcon], safe=True) _migrate_owner_column(cfg.DATA_DIR) + _add_missing_columns() + + +def _add_missing_columns() -> None: + """Add columns introduced after a DB was first created. + + `create_tables(safe=True)` skips an existing table entirely, so a new field + on an old model never lands without this. Plain ADD COLUMN of a nullable + field — no backfill needed. + """ + added: dict[str, str] = {"driver": "display_name VARCHAR(255)"} + for table, column_def in added.items(): + column = column_def.split()[0] + columns = {row[1] for row in db.execute_sql(f"PRAGMA table_info({table})")} + if column not in columns: + db.execute_sql(f"ALTER TABLE {table} ADD COLUMN {column_def}") def _migrate_owner_column(data_dir: str) -> None: diff --git a/imptune/db/models.py b/imptune/db/models.py index 3fdd432..0f46ab5 100644 --- a/imptune/db/models.py +++ b/imptune/db/models.py @@ -58,10 +58,18 @@ class Driver(BaseModel): inf_filename = CharField(null=True) architecture = CharField(null=True) has_cat_file = BooleanField(default=False) + # Operator-chosen label. Drivers are global/shared, so a rename is visible + # to every Owner — same as the rest of this row. + display_name = CharField(null=True) class Meta: table_name = "driver" + @property + def label(self) -> str: + """What the UI shows: the rename if there is one, else the ZIP name.""" + return self.display_name or self.original_filename + class Printer(BaseModel): """Printer configuration record.""" @@ -94,3 +102,21 @@ class Icon(BaseModel): class Meta: table_name = "icon" + + +class DriverIcon(BaseModel): + """Icon image for a driver package (one per driver). + + Separate table rather than a nullable FK on `Icon`: printer icons ship in + the `.intunewin` export and driver icons are library decoration only, so + the two never share a query. Global/shared, like `Driver` itself. + """ + + driver = ForeignKeyField(Driver, unique=True, backref="icons", on_delete="CASCADE") + sha256 = CharField() + original_filename = CharField() + size_bytes = IntegerField() + uploaded_at = DateTimeField(default=_utcnow) + + class Meta: + table_name = "driver_icon" diff --git a/imptune/main.py b/imptune/main.py index a6dfb16..6c5670f 100644 --- a/imptune/main.py +++ b/imptune/main.py @@ -5,7 +5,18 @@ from pathlib import Path from fastapi import FastAPI from fastapi.staticfiles import StaticFiles -from imptune.api import clients, drivers, health, icons, pages, packages, printers, scripts, session +from imptune.api import ( + clients, + drivers, + health, + icons, + packages, + pages, + printers, + scripts, + session, + web, +) from imptune.config import DATA_DIR, DRIVERS_DIR, ICONS_DIR from imptune.db.database import db, init_db from imptune.services.session import OwnerSessionMiddleware @@ -40,3 +51,4 @@ app.include_router(scripts.router) app.include_router(packages.router) app.include_router(icons.router) app.include_router(session.router) +app.include_router(web.router) diff --git a/imptune/services/icons.py b/imptune/services/icons.py new file mode 100644 index 0000000..6d9b6a9 --- /dev/null +++ b/imptune/services/icons.py @@ -0,0 +1,93 @@ +"""Icon storage shared by printer icons and driver icons. + +Both live in `ICONS_DIR`, content-addressed by the SHA256 of the *normalized* +bytes, so two entities that end up with the same 256x256 PNG share one file. +""" +from __future__ import annotations + +import hashlib +from pathlib import Path + +import imptune.config as cfg +from imptune.db.models import Driver, DriverIcon, Icon +from imptune.services.image_utils import ImageError, normalize_icon + +MAX_ICON_BYTES = 750 * 1024 # 750 KB, on the bytes as uploaded + +__all__ = [ + "ImageError", + "IconTooLarge", + "MAX_ICON_BYTES", + "driver_icon_ids", + "icon_path", + "set_driver_icon", + "set_printer_icon", + "store_bytes", +] + + +class IconTooLarge(ValueError): + """Raised when the source file is over `MAX_ICON_BYTES`.""" + + +def icon_path(sha256: str) -> Path: + """On-disk location of a stored icon. Read `cfg` late — tests patch it.""" + return Path(cfg.ICONS_DIR) / sha256 + + +def store_bytes(data: bytes) -> tuple[str, int]: + """Normalize `data` to a 256x256 PNG, write it, return `(sha256, size)`. + + Raises `IconTooLarge` or `ImageError` before anything touches the disk. + """ + if len(data) > MAX_ICON_BYTES: + raise IconTooLarge(f"Icon exceeds {MAX_ICON_BYTES // 1024} KB limit.") + + png = normalize_icon(data) + sha256 = hashlib.sha256(png).hexdigest() + path = icon_path(sha256) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(png) + return sha256, len(png) + + +def set_printer_icon(printer_id: int, data: bytes, filename: str) -> Icon: + """Replace this printer's icon. One `Icon` row per printer, always.""" + sha256, size = store_bytes(data) + Icon.delete().where(Icon.printer == printer_id).execute() + return Icon.create( + printer=printer_id, + sha256=sha256, + original_filename=filename or "icon.png", + size_bytes=size, + ) + + +def set_driver_icon(driver_id: int, data: bytes, filename: str) -> DriverIcon: + """Replace this driver's icon. Driver rows are global, so this icon is too.""" + sha256, size = store_bytes(data) + DriverIcon.delete().where(DriverIcon.driver == driver_id).execute() + return DriverIcon.create( + driver=driver_id, + sha256=sha256, + original_filename=filename or "icon.png", + size_bytes=size, + ) + + +def driver_icon_ids(drivers=None) -> set[int]: + """Ids of drivers that have an icon, for the library listing.""" + query = DriverIcon.select(DriverIcon.driver) + if drivers is not None: + ids = [d.id for d in drivers] or [0] + query = query.where(DriverIcon.driver.in_(ids)) + return {row.driver_id for row in query} + + +def driver_search_text(driver: Driver, names: list[str]) -> str: + """Best guess at what this driver is, for prefilling a search box.""" + if driver.display_name: + return driver.display_name + if names: + return names[0] + return Path(driver.original_filename).stem.replace("_", " ").replace("-", " ") diff --git a/imptune/services/image_utils.py b/imptune/services/image_utils.py new file mode 100644 index 0000000..3e89561 --- /dev/null +++ b/imptune/services/image_utils.py @@ -0,0 +1,50 @@ +"""Icon normalization — anything Pillow can decode becomes a 256x256 PNG. + +Intune wants exactly 256x256 PNG, but nothing a human picks (a photo from the +web, a vendor logo, a screenshot) arrives that way. Rather than reject it, fit +it into the box: aspect ratio preserved, transparent letterbox around it. +""" +from __future__ import annotations + +import io + +from PIL import Image, ImageOps, UnidentifiedImageError + +ICON_SIZE = (256, 256) + + +class ImageError(ValueError): + """Raised when the bytes are not a decodable raster image.""" + + +def normalize_icon(data: bytes) -> bytes: + """Return `data` as exactly-256x256 PNG bytes. + + A file that *already* is a 256x256 PNG is returned byte-identical — the + icon store is content-addressed by SHA256, so re-encoding an unchanged + upload would move it to a new path on every save for no reason. + + Raises `ImageError` if the bytes cannot be decoded as an image. + """ + try: + img = Image.open(io.BytesIO(data)) + img.load() + except (UnidentifiedImageError, OSError, ValueError) as exc: + raise ImageError("Not a readable image file.") from exc + + if img.format == "PNG" and img.size == ICON_SIZE: + return data + + # `contain` scales down to fit inside the box without cropping; a smaller + # source is left at its own size rather than blown up into mush. + fitted = ImageOps.contain(img.convert("RGBA"), ICON_SIZE, Image.LANCZOS) + + canvas = Image.new("RGBA", ICON_SIZE, (0, 0, 0, 0)) + canvas.paste( + fitted, + ((ICON_SIZE[0] - fitted.width) // 2, (ICON_SIZE[1] - fitted.height) // 2), + ) + + out = io.BytesIO() + canvas.save(out, format="PNG", optimize=True) + return out.getvalue() diff --git a/imptune/services/websearch.py b/imptune/services/websearch.py new file mode 100644 index 0000000..7a5dfef --- /dev/null +++ b/imptune/services/websearch.py @@ -0,0 +1,384 @@ +"""Web lookups: image search, driver-page search, and guarded image download. + +The only place in ImpTune that talks to the internet. Everything here is +best-effort: a failed lookup returns an empty list or raises `WebSearchError`, +and the rest of the app keeps working offline. + +DuckDuckGo is scraped rather than called through an API — no key, no signup. +That makes it fragile by nature: `search_images` / `search_pages` swallow parse +failures and return `[]` instead of a 500. Kill the whole feature with +`WEB_SEARCH=false`. + +Fetching a URL the user picked means the *server* makes the request, so +`fetch_image` refuses anything that resolves to a private address. ImpTune runs +on the same LAN as the printers it configures — an unguarded fetcher would be a +port scanner for whoever can reach the UI. +""" +from __future__ import annotations + +import html +import ipaddress +import json +import re +import socket +import urllib.error +import urllib.request +from dataclasses import dataclass +from urllib.parse import parse_qs, quote_plus, urlparse + +import imptune.config as cfg + +USER_AGENT = ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/125.0 Safari/537.36" +) +TIMEOUT_SECONDS = 8 +MAX_IMAGE_BYTES = 8 * 1024 * 1024 +MAX_HTML_BYTES = 2 * 1024 * 1024 +MAX_REDIRECTS = 3 + +DDG_HOME = "https://duckduckgo.com/" +DDG_IMAGES = "https://duckduckgo.com/i.js" +DDG_HTML = "https://html.duckduckgo.com/html/" + +# Brands with a real vendor-wide driver. The search term is what a technician +# would type; the live results are whatever the engine returns for it. +GENERIC_DRIVER_TERMS: dict[str, str] = { + "hp": "HP Universal Print Driver PCL6 download", + "konica": "Konica Minolta Universal PCL Print Driver download", + "konica minolta": "Konica Minolta Universal PCL Print Driver download", + "xerox": "Xerox Global Print Driver download", + "ricoh": "Ricoh PCL6 Universal Print Driver download", + "lexmark": "Lexmark Universal Print Driver download", + "brother": "Brother universal printer driver download", + "canon": "Canon Generic PCL6 Printer Driver download", + "epson": "Epson Universal Print Driver download", + "kyocera": "Kyocera Classic Universal Print Driver download", + "sharp": "Sharp Universal Print Driver download", + "toshiba": "Toshiba Universal Printer 2 driver download", + "oki": "OKI PCL universal print driver download", + "samsung": "Samsung Universal Print Driver download", + "dell": "Dell universal printer driver download", +} + + +# Display spellings for the search box's datalist, in the order a French shop +# is likeliest to need them. +BRAND_SUGGESTIONS = ( + "HP", + "Konica Minolta", + "Xerox", + "Ricoh", + "Canon", + "Kyocera", + "Brother", + "Lexmark", + "Epson", + "Sharp", + "Toshiba", + "OKI", + "Samsung", + "Dell", +) + + +class WebSearchError(RuntimeError): + """A lookup or download could not be completed.""" + + +@dataclass(frozen=True) +class ImageHit: + """One image search result.""" + + url: str + thumbnail: str + title: str + width: int + height: int + source: str + + +@dataclass(frozen=True) +class PageHit: + """One web page search result.""" + + url: str + title: str + snippet: str + host: str + + +def enabled() -> bool: + """False when the deployment opted out of outbound requests.""" + return bool(getattr(cfg, "WEB_SEARCH", True)) + + +def _require_enabled() -> None: + if not enabled(): + raise WebSearchError("Web lookups are disabled (WEB_SEARCH=false).") + + +# -------------------------------------------------------------------------- +# SSRF guard +# -------------------------------------------------------------------------- + +def _is_public_ip(raw: str) -> bool: + try: + ip = ipaddress.ip_address(raw) + except ValueError: + return False + return not ( + ip.is_private + or ip.is_loopback + or ip.is_link_local + or ip.is_multicast + or ip.is_reserved + or ip.is_unspecified + ) + + +def assert_fetchable(url: str) -> None: + """Raise `WebSearchError` unless `url` is an https(s) URL on a public host. + + Resolution happens here *and* the socket connects by hostname afterwards, so + a DNS-rebinding host could still slip through between the two. Accepted: + the payload is decoded by Pillow and discarded unless it is an image. + """ + parsed = urlparse(url) + if parsed.scheme not in ("http", "https"): + raise WebSearchError("Only http and https URLs can be fetched.") + if not parsed.hostname: + raise WebSearchError("URL has no host.") + + try: + infos = socket.getaddrinfo(parsed.hostname, None) + except socket.gaierror as exc: + raise WebSearchError(f"Cannot resolve {parsed.hostname}.") from exc + + for info in infos: + if not _is_public_ip(info[4][0]): + raise WebSearchError( + f"{parsed.hostname} resolves to a private address — refused." + ) + + +class _GuardedRedirectHandler(urllib.request.HTTPRedirectHandler): + """Re-run the SSRF guard on every redirect target.""" + + def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: D102 + assert_fetchable(newurl) + return super().redirect_request(req, fp, code, msg, headers, newurl) + + +_opener = urllib.request.build_opener(_GuardedRedirectHandler) + + +def _get( + url: str, + *, + referer: str | None = None, + max_bytes: int, + extra_headers: dict[str, str] | None = None, +) -> bytes: + """GET `url` with the SSRF guard applied and the response body capped.""" + assert_fetchable(url) + headers = {"User-Agent": USER_AGENT, "Accept-Language": "en-US,en;q=0.7"} + if referer: + headers["Referer"] = referer + if extra_headers: + headers.update(extra_headers) + request = urllib.request.Request(url, headers=headers) + try: + with _opener.open(request, timeout=TIMEOUT_SECONDS) as response: + return response.read(max_bytes + 1)[: max_bytes + 1] + except urllib.error.URLError as exc: + raise WebSearchError(f"Request failed: {exc}") from exc + except (TimeoutError, socket.timeout) as exc: # noqa: UP041 — socket.timeout on 3.9 paths + raise WebSearchError("Request timed out.") from exc + + +# -------------------------------------------------------------------------- +# Image search +# -------------------------------------------------------------------------- + +_VQD_RE = re.compile(rb"vqd=[\"']?([\w-]{8,})[\"']?") + +# The image endpoint is an XHR route: without the fetch metadata a real browser +# would send, it answers 403 even with a valid vqd token. Verified header set — +# dropping any of these brings the 403 back. +_XHR_HEADERS = { + "Accept": "application/json, text/javascript, */*; q=0.01", + "X-Requested-With": "XMLHttpRequest", + "Sec-Fetch-Dest": "empty", + "Sec-Fetch-Mode": "cors", + "Sec-Fetch-Site": "same-origin", +} + + +def _image_vqd(query: str) -> str: + """Scrape the per-query token DuckDuckGo's image endpoint demands.""" + body = _get( + f"{DDG_HOME}?q={quote_plus(query)}&iax=images&ia=images", + max_bytes=MAX_HTML_BYTES, + ) + match = _VQD_RE.search(body) + if not match: + raise WebSearchError("Could not read the search token.") + return match.group(1).decode("ascii", "ignore") + + +def search_images(query: str, limit: int = 12) -> list[ImageHit]: + """Image results for `query`, newest-first as the engine ranked them. + + Returns `[]` on any parse or transport failure — a dead scrape must not + take a page down with it. + """ + _require_enabled() + query = query.strip() + if not query: + return [] + + try: + vqd = _image_vqd(query) + raw = _get( + f"{DDG_IMAGES}?l=us-en&o=json&q={quote_plus(query)}&vqd={vqd}&f=,,,&p=1", + referer=DDG_HOME, + max_bytes=MAX_HTML_BYTES, + extra_headers=_XHR_HEADERS, + ) + payload = json.loads(raw.decode("utf-8", "replace")) + except (WebSearchError, json.JSONDecodeError): + return [] + + hits: list[ImageHit] = [] + for item in payload.get("results", []): + url = item.get("image") or "" + if not url.startswith(("http://", "https://")): + continue + hits.append( + ImageHit( + url=url, + thumbnail=item.get("thumbnail") or url, + title=(item.get("title") or "").strip(), + width=int(item.get("width") or 0), + height=int(item.get("height") or 0), + source=urlparse(url).hostname or "", + ) + ) + if len(hits) >= limit: + break + return hits + + +def fetch_image(url: str) -> bytes: + """Download `url` and return its bytes (caller validates it is an image).""" + _require_enabled() + data = _get(url, max_bytes=MAX_IMAGE_BYTES) + if len(data) > MAX_IMAGE_BYTES: + raise WebSearchError("Image is larger than 8 MB.") + if not data: + raise WebSearchError("Server returned an empty response.") + return data + + +# -------------------------------------------------------------------------- +# Page search (driver downloads) +# -------------------------------------------------------------------------- + +_RESULT_RE = re.compile( + r']+class="result__a"[^>]+href="(?P[^"]+)"[^>]*>(?P.*?)</a>', + re.DOTALL, +) +_SNIPPET_RE = re.compile( + r'<a[^>]+class="result__snippet"[^>]*>(?P<snippet>.*?)</a>', re.DOTALL +) +_TAG_RE = re.compile(r"<[^>]+>") + + +def _text(fragment: str) -> str: + return html.unescape(_TAG_RE.sub("", fragment)).strip() + + +def _snippet_text(fragment: str) -> str: + """Snippet text, minus the literal "undefined " DuckDuckGo's own template + prefixes onto some results.""" + text = _text(fragment) + return text[len("undefined ") :].lstrip() if text.startswith("undefined ") else text + + +def _unwrap(href: str) -> str: + """Unwrap DuckDuckGo's `/l/?uddg=` click-tracking redirect.""" + if href.startswith("//"): + href = "https:" + href + parsed = urlparse(href) + if "duckduckgo.com" in (parsed.hostname or "") and parsed.path.startswith("/l/"): + target = parse_qs(parsed.query).get("uddg", [""])[0] + if target: + return target + return href + + +def search_pages(query: str, limit: int = 8) -> list[PageHit]: + """Web page results for `query`. `[]` on any failure, same as image search.""" + _require_enabled() + query = query.strip() + if not query: + return [] + + try: + body = _get( + f"{DDG_HTML}?q={quote_plus(query)}", max_bytes=MAX_HTML_BYTES + ).decode("utf-8", "replace") + except WebSearchError: + return [] + + snippets = [_snippet_text(m.group("snippet")) for m in _SNIPPET_RE.finditer(body)] + hits: list[PageHit] = [] + for index, match in enumerate(_RESULT_RE.finditer(body)): + url = _unwrap(html.unescape(match.group("href"))) + if not url.startswith(("http://", "https://")): + continue + hits.append( + PageHit( + url=url, + title=_text(match.group("title")) or url, + snippet=snippets[index] if index < len(snippets) else "", + host=urlparse(url).hostname or "", + ) + ) + if len(hits) >= limit: + break + return hits + + +# -------------------------------------------------------------------------- +# Query builders +# -------------------------------------------------------------------------- + +def detect_brand(text: str) -> str | None: + """Longest brand in `GENERIC_DRIVER_TERMS` that appears in `text`.""" + lowered = (text or "").lower() + matches = [brand for brand in GENERIC_DRIVER_TERMS if brand in lowered] + return max(matches, key=len) if matches else None + + +def generic_driver_query(text: str) -> str: + """Search term for a vendor-wide driver, derived from a model name. + + A recognised brand gets that vendor's actual product name ("HP Universal + Print Driver PCL6"); anything else falls back to the raw model plus + "universal print driver download", which is what a technician would type. + """ + text = (text or "").strip() + brand = detect_brand(text) + if brand: + return GENERIC_DRIVER_TERMS[brand] + return f"{text} universal print driver download".strip() + + +def image_query(text: str) -> str: + """Default search term for a printer or driver icon.""" + text = (text or "").strip() + if not text: + return "" + return f"{text} printer" diff --git a/imptune/static/app.css b/imptune/static/app.css index 72c236f..e997d0a 100644 --- a/imptune/static/app.css +++ b/imptune/static/app.css @@ -1263,6 +1263,106 @@ div.error > p { .icon-preview .meta { font-size: .812rem; color: var(--im-text-dim); } +.error-note { + margin: 0 0 .8rem; + font-size: .875rem; + color: var(--im-danger); +} + +/* Driver-library thumbnail: fixed box whether or not an icon exists, so the + name column does not shift when one is added out of band. */ +.cell-with-thumb { + display: flex; + align-items: flex-start; + gap: .6rem; +} + +.driver-thumb { + flex: none; + width: 24px; + height: 24px; + margin-top: .1rem; + border-radius: var(--im-r-sm); + background: var(--im-surface-3); + overflow: hidden; +} + +.driver-thumb img { width: 24px; height: 24px; display: block; } + +/* ========================================================================== + Web lookups — image picker and driver-page results + ========================================================================== */ + +.web-picker { + margin-top: .9rem; + padding-top: .9rem; + border-top: 1px dashed var(--im-line); +} + +.web-picker label { margin: 0 0 .5rem; } +.web-picker .btn-row { margin: 0; } + +.image-grid { + display: grid; + grid-template-columns: repeat(auto-fill, minmax(84px, 1fr)); + gap: .5rem; + margin: .6rem 0; +} + +/* Each result is a button, not an <img> with a click handler — keyboard + selection and focus rings come for free. */ +.image-option { + position: relative; + display: block; + width: 100%; + aspect-ratio: 1; + padding: 0; + border: 1px solid var(--im-line); + border-radius: var(--im-r-sm); + background: var(--im-surface-3); + overflow: hidden; + cursor: pointer; +} + +.image-option:hover, +.image-option:focus-visible { border-color: var(--im-accent); } + +.image-option img { + width: 100%; + height: 100%; + object-fit: contain; + display: block; +} + +.image-option .image-meta { + position: absolute; + inset: auto 0 0 0; + padding: .1rem .25rem; + font-size: .625rem; + line-height: 1.4; + color: #fff; + background: rgba(0, 0, 0, .55); +} + +.link-list { + list-style: none; + margin: 0; + padding: 0; + display: flex; + flex-direction: column; + gap: .7rem; +} + +.link-list li { + padding-bottom: .7rem; + border-bottom: 1px solid var(--im-line); +} + +.link-list li:last-child { border-bottom: 0; padding-bottom: 0; } +.link-list a { font-size: .906rem; font-weight: 560; } +.link-list p { margin: .2rem 0 0; font-size: .812rem; } +.link-list .cell-sub { display: block; } + /* ========================================================================== Dialogs ========================================================================== */ @@ -1406,7 +1506,8 @@ details.more-pills > summary::marker { content: ""; } sidebar there is only room for the file and its parsed names. */ .form-aside .data-table .col-arch, .form-aside .data-table .col-used, -.form-aside .data-table .col-added { display: none; } +.form-aside .data-table .col-added, +.form-aside .data-table .col-actions { display: none; } .uploader label { margin: 0; } .uploader .hint { margin-top: 0; } diff --git a/imptune/templates/base.html b/imptune/templates/base.html index 6044e68..66daf81 100644 --- a/imptune/templates/base.html +++ b/imptune/templates/base.html @@ -224,6 +224,35 @@ replace_icon: 'Remplacer l’icône', no_icon: 'Aucune icône. Intune affichera l’icône par défaut.', hint_icon: 'PNG, exactement 256 × 256, 750 Ko maximum.', + hint_any_image: 'N’importe quelle image — redimensionnée en PNG 256 × 256. 750 Ko maximum.', + // Recherche d’images sur le web + search_web_image: 'Chercher une image sur le web', + search_btn: 'Chercher', + searching: 'Recherche…', + image_results_for: 'Résultats pour', + image_pick_hint: 'Cliquez sur une image : elle est téléchargée puis redimensionnée en PNG 256 × 256.', + no_image_results: 'Aucun résultat. Essayez un autre terme.', + web_search_off: 'Les recherches web sont désactivées sur ce serveur.', + unknown_target: 'Cible de recherche inconnue.', + printer_not_found: 'Imprimante introuvable.', + driver_not_found: 'Pilote introuvable.', + type_to_search_image: 'Saisissez un terme à rechercher.', + type_to_search_driver: 'Saisissez une marque ou un modèle d’imprimante.', + // Pilotes : renommage et icône + edit_driver_title: 'Modifier le pilote', + section_driver_name: 'Nom', + driver_display_name: 'Nom affiché', + hint_driver_rename: 'Visible par tout le monde — la bibliothèque de pilotes est partagée. Laissez vide pour afficher le nom du ZIP.', + no_driver_icon: 'Aucune icône.', + // Recherche de pilotes sur le web + find_driver_online: 'Trouver un pilote en ligne', + find_driver_intro: 'Cherchez un pilote universel du constructeur, ou un modèle précis. ImpTune n’affiche que des liens : téléchargez le ZIP puis téléversez-le ci-dessus.', + driver_search_label: 'Marque ou modèle d’imprimante', + search_generic_btn: 'Chercher le pilote universel', + search_exact_btn: 'Chercher le modèle exact', + searched_for: 'Recherche effectuée pour', + driver_search_warning: 'Résultats web bruts, sans filtrage ni validation. Ne téléchargez un pilote que depuis le site du constructeur, puis téléversez le ZIP ci-dessus.', + no_driver_results: 'Aucun résultat. Essayez le nom exact du modèle.', back_to_printers_btn: 'Retour aux imprimantes', export_locked: 'Assignez un pilote à cette imprimante pour débloquer les scripts et l’export.', // Edit modal @@ -410,6 +439,35 @@ replace_icon: 'Replace icon', no_icon: 'No icon. Intune will show its default.', hint_icon: 'PNG, exactly 256 × 256, 750 KB max.', + hint_any_image: 'Any image — resized to 256 × 256 PNG. 750 KB max.', + // Web image search + search_web_image: 'Search the web for an image', + search_btn: 'Search', + searching: 'Searching…', + image_results_for: 'Results for', + image_pick_hint: 'Click an image — it is downloaded and resized to 256 × 256 PNG.', + no_image_results: 'No results. Try another search term.', + web_search_off: 'Web lookups are disabled on this server.', + unknown_target: 'Unknown search target.', + printer_not_found: 'Printer not found.', + driver_not_found: 'Driver not found.', + type_to_search_image: 'Type something to search for.', + type_to_search_driver: 'Type a printer model or brand to search for.', + // Driver rename + icon + edit_driver_title: 'Edit driver', + section_driver_name: 'Name', + driver_display_name: 'Display name', + hint_driver_rename: 'Shared with everyone — the driver library is global. Leave empty to show the ZIP filename.', + no_driver_icon: 'No icon yet.', + // Driver web search + find_driver_online: 'Find a driver online', + find_driver_intro: 'Search for a vendor-wide driver, or for an exact model. ImpTune only shows links — you download the ZIP and upload it above.', + driver_search_label: 'Brand or printer model', + search_generic_btn: 'Find generic driver', + search_exact_btn: 'Search exact model', + searched_for: 'Searched for', + driver_search_warning: 'Unfiltered web results, not a vetted list. Download drivers only from the manufacturer’s own site, then upload the ZIP above.', + no_driver_results: 'No results. Try the exact model name instead.', back_to_printers_btn: 'Back to printers', export_locked: 'Assign a driver to this printer to unlock scripts and export.', // Edit modal diff --git a/imptune/templates/drivers.html b/imptune/templates/drivers.html index 37f87ac..d563996 100644 --- a/imptune/templates/drivers.html +++ b/imptune/templates/drivers.html @@ -38,6 +38,44 @@ </div> </section> +{% if web_search_enabled() %} +<section class="card"> + <div class="card-head"> + {{ ico.i('search') }} + <div> + <h2 x-data x-text="$store.i18n.t('find_driver_online')">Find a driver online</h2> + <p class="sub" x-data x-text="$store.i18n.t('find_driver_intro')">Search for a vendor-wide driver, or for an exact model. ImpTune only shows links — you download the ZIP and upload it above.</p> + </div> + </div> + <div class="card-body"> + {# Two submit buttons, one form: the clicked button's `mode` value is what + gets sent, so "generic" and "exact" are one round-trip apart. #} + <form hx-get="/web/drivers" + hx-target="#driver-search-results" + hx-swap="innerHTML" + hx-indicator="#driver-search-spinner"> + <label> + <span class="label-text" x-data x-text="$store.i18n.t('driver_search_label')">Brand or printer model</span> + <input type="search" name="q" list="driver-brands" placeholder="HP LaserJet M404"> + </label> + <datalist id="driver-brands"> + {% for brand in brands %}<option value="{{ brand }}">{% endfor %} + </datalist> + <div class="btn-row"> + <button type="submit" class="btn" name="mode" value="generic" x-data> + {{ ico.i('search', 14) }}<span x-text="$store.i18n.t('search_generic_btn')">Find generic driver</span> + </button> + <button type="submit" class="btn ghost" name="mode" value="exact" x-data + x-text="$store.i18n.t('search_exact_btn')">Search exact model</button> + <span id="driver-search-spinner" class="htmx-indicator" aria-busy="true" + x-data x-text="$store.i18n.t('searching')">Searching…</span> + </div> + </form> + <div id="driver-search-results"></div> + </div> +</section> +{% endif %} + <h2 class="eyebrow" style="margin-bottom:.6rem" x-data x-text="$store.i18n.t('driver_library')">Driver library</h2> {% include "partials/driver_list.html" %} {% endblock %} diff --git a/imptune/templates/partials/driver_edit_modal.html b/imptune/templates/partials/driver_edit_modal.html new file mode 100644 index 0000000..616ea38 --- /dev/null +++ b/imptune/templates/partials/driver_edit_modal.html @@ -0,0 +1,109 @@ +{# + Rename + icon dialog for one driver row. `item` is a build_driver_data() entry. + + The rename form and the icon forms are siblings, not nested — a form inside a + form is invalid HTML and the browser drops the inner one. The footer's Save + reaches the rename form through `form="..."`. + + Trigger is located by `button[onclick*='showModal']` in E2E specs, never by its + label: the default language follows navigator.language. +#} +{% import "partials/icons.html" as ico %} +{% set d = item.driver %} +<button class="btn ghost sm" + onclick="document.getElementById('driver-modal-{{ d.id }}').showModal()" + x-data> + {{ ico.i('pencil', 14) }}<span x-text="$store.i18n.t('edit')">Edit</span> +</button> + +<dialog id="driver-modal-{{ d.id }}"> + <article> + <div class="dialog-head"> + <h3 x-data x-text="$store.i18n.t('edit_driver_title')">Edit driver</h3> + <button type="button" class="icon-btn close" x-data + :aria-label="$store.i18n.t('cancel')" + onclick="document.getElementById('driver-modal-{{ d.id }}').close()">×</button> + </div> + + <div class="dialog-body"> + <form id="driver-rename-{{ d.id }}" + hx-patch="/drivers/{{ d.id }}" + hx-target="#driver-list" + hx-swap="outerHTML" + hx-on::after-request="document.getElementById('driver-modal-{{ d.id }}').close()"> + <fieldset class="form-section"> + <legend x-data x-text="$store.i18n.t('section_driver_name')">Name</legend> + <label> + <span class="label-text" x-data x-text="$store.i18n.t('driver_display_name')">Display name</span> + <input type="text" name="display_name" maxlength="120" + value="{{ d.display_name or '' }}" + placeholder="{{ d.original_filename }}"> + </label> + <span class="hint" x-data x-text="$store.i18n.t('hint_driver_rename')">Shared with everyone — the driver library is global. Leave empty to show the ZIP filename.</span> + </fieldset> + </form> + + <fieldset class="form-section"> + <legend x-data x-text="$store.i18n.t('icon_section')">Icon</legend> + <div id="driver-icon-status-{{ d.id }}"> + {% if item.has_icon %} + <div class="icon-preview"> + <img src="/drivers/{{ d.id }}/icon" width="56" height="56" alt=""> + <span class="meta">256×256 PNG</span> + </div> + {% else %} + <p class="dim" x-data x-text="$store.i18n.t('no_driver_icon')">No icon yet.</p> + {% endif %} + </div> + + <form hx-post="/drivers/{{ d.id }}/icon" + hx-target="#driver-icon-status-{{ d.id }}" + hx-swap="innerHTML" + hx-encoding="multipart/form-data"> + <div class="uploader"> + <input type="file" name="file" accept="image/*" required + x-data :aria-label="$store.i18n.t('upload_icon')"> + <div class="btn-row"> + <button type="submit" class="btn ghost sm" x-data> + {{ ico.i('upload', 13) }}<span x-text="$store.i18n.t('upload_icon')">Upload icon</span> + </button> + </div> + <p class="hint" x-data x-text="$store.i18n.t('hint_any_image')">Any image — resized to 256 × 256 PNG. 750 KB max.</p> + </div> + </form> + + {% if web_search_enabled() %} + <div class="web-picker"> + <form hx-get="/web/images" + hx-target="#driver-image-results-{{ d.id }}" + hx-swap="innerHTML" + hx-indicator="#driver-image-spinner-{{ d.id }}"> + <input type="hidden" name="target" value="driver"> + <input type="hidden" name="id" value="{{ d.id }}"> + <label> + <span class="label-text" x-data x-text="$store.i18n.t('search_web_image')">Search the web for an image</span> + <input type="search" name="q" value="{{ item.image_query }}"> + </label> + <div class="btn-row"> + <button type="submit" class="btn ghost sm" x-data> + {{ ico.i('search', 13) }}<span x-text="$store.i18n.t('search_btn')">Search</span> + </button> + <span id="driver-image-spinner-{{ d.id }}" class="htmx-indicator" aria-busy="true" + x-data x-text="$store.i18n.t('searching')">Searching…</span> + </div> + </form> + <div id="driver-image-results-{{ d.id }}"></div> + </div> + {% endif %} + </fieldset> + </div> + + <div class="dialog-foot"> + <button type="button" class="btn ghost" x-data + onclick="document.getElementById('driver-modal-{{ d.id }}').close()" + x-text="$store.i18n.t('cancel')">Cancel</button> + <button type="submit" class="btn" form="driver-rename-{{ d.id }}" x-data + x-text="$store.i18n.t('save')">Save</button> + </div> + </article> +</dialog> diff --git a/imptune/templates/partials/driver_list.html b/imptune/templates/partials/driver_list.html index 2e88d08..26b4d54 100644 --- a/imptune/templates/partials/driver_list.html +++ b/imptune/templates/partials/driver_list.html @@ -30,17 +30,34 @@ <th class="col-arch" x-data x-text="$store.i18n.t('architecture')">Architecture</th> <th class="col-used" x-data x-text="$store.i18n.t('th_used_by')">Used by</th> <th class="col-added" x-data x-text="$store.i18n.t('uploaded_at')">Added</th> + <th class="col-actions"><span x-data x-text="$store.i18n.t('th_actions')">Actions</span></th> </tr> </thead> <tbody> {% for item in driver_data %} <tr> <td> - <span class="cell-name mono">{{ item.driver.original_filename }}</span> - <span class="cell-sub"> - {% if item.driver.size_bytes >= 1048576 %}{{ (item.driver.size_bytes / 1048576) | round(1) }} MB{% elif item.driver.size_bytes >= 1024 %}{{ (item.driver.size_bytes / 1024) | round(0) | int }} KB{% else %}{{ item.driver.size_bytes }} B{% endif %} - {% if item.driver.inf_filename %} · {{ item.driver.inf_filename }}{% endif %} - </span> + <div class="cell-with-thumb"> + {# Placeholder span keeps the id addressable so an icon saved in + the dialog can be swapped in out of band. #} + <span id="driver-thumb-{{ item.driver.id }}" class="driver-thumb"> + {% if item.has_icon %} + <img src="/drivers/{{ item.driver.id }}/icon" width="24" height="24" alt=""> + {% endif %} + </span> + <span> + {% if item.driver.display_name %} + <span class="cell-name">{{ item.driver.display_name }}</span> + <span class="cell-sub mono">{{ item.driver.original_filename }}</span> + {% else %} + <span class="cell-name mono">{{ item.driver.original_filename }}</span> + {% endif %} + <span class="cell-sub"> + {% if item.driver.size_bytes >= 1048576 %}{{ (item.driver.size_bytes / 1048576) | round(1) }} MB{% elif item.driver.size_bytes >= 1024 %}{{ (item.driver.size_bytes / 1024) | round(0) | int }} KB{% else %}{{ item.driver.size_bytes }} B{% endif %} + {% if item.driver.inf_filename %} · {{ item.driver.inf_filename }}{% endif %} + </span> + </span> + </div> </td> <td> {% if item.names %} @@ -77,6 +94,9 @@ <span class="badge ok">{{ ico.i('check', 12) }}new</span> {% endif %} </td> + <td class="actions col-actions"> + {% include "partials/driver_edit_modal.html" %} + </td> </tr> {% endfor %} </tbody> diff --git a/imptune/templates/partials/driver_search_results.html b/imptune/templates/partials/driver_search_results.html new file mode 100644 index 0000000..90b21e6 --- /dev/null +++ b/imptune/templates/partials/driver_search_results.html @@ -0,0 +1,29 @@ +{# + Driver download search results — links only. ImpTune never downloads a driver + for you: these are raw web results, and the warning below says so plainly. +#} +{% if hits %} +<p class="hint"> + <span x-data x-text="$store.i18n.t('searched_for')">Searched for</span> + <span class="mono">{{ query }}</span> +</p> + +<div class="notice warn"> + <p x-data x-text="$store.i18n.t('driver_search_warning')"> + Unfiltered web results, not a vetted list. Download drivers only from the + manufacturer's own site, then upload the ZIP above. + </p> +</div> + +<ul class="link-list"> + {% for h in hits %} + <li> + <a href="{{ h.url }}" target="_blank" rel="noopener noreferrer nofollow">{{ h.title }}</a> + <span class="cell-sub mono">{{ h.host }}</span> + {% if h.snippet %}<p class="dim">{{ h.snippet }}</p>{% endif %} + </li> + {% endfor %} +</ul> +{% else %} +<p class="dim" x-data x-text="$store.i18n.t('no_driver_results')">No results. Try the exact model name instead.</p> +{% endif %} diff --git a/imptune/templates/partials/image_results.html b/imptune/templates/partials/image_results.html new file mode 100644 index 0000000..d5eb544 --- /dev/null +++ b/imptune/templates/partials/image_results.html @@ -0,0 +1,30 @@ +{# + Web image picker. Each result is a button that POSTs its URL to `post_url`; + the server downloads and normalizes it, then returns the icon-status block + that `status_target` points at. + + Thumbnails load straight from the search engine's CDN, so the browser talks + to it directly — `referrerpolicy` keeps ImpTune's own URL out of that request. +#} +{% if hits %} +<p class="hint"> + <span x-data x-text="$store.i18n.t('image_results_for')">Results for</span> + <span class="mono">{{ query }}</span> +</p> +<div class="image-grid"> + {% for h in hits %} + <button type="button" class="image-option" + hx-post="{{ post_url }}" + hx-vals='{{ {"url": h.url} | tojson }}' + hx-target="{{ status_target }}" + hx-swap="innerHTML" + title="{{ h.title }} — {{ h.source }}"> + <img src="{{ h.thumbnail }}" alt="" loading="lazy" referrerpolicy="no-referrer"> + <span class="image-meta mono">{{ h.width }}×{{ h.height }}</span> + </button> + {% endfor %} +</div> +<p class="hint" x-data x-text="$store.i18n.t('image_pick_hint')">Pick one — it is downloaded and resized to 256 × 256 PNG.</p> +{% else %} +<p class="dim" x-data x-text="$store.i18n.t('no_image_results')">No results. Try another search term.</p> +{% endif %} diff --git a/imptune/templates/printer_detail.html b/imptune/templates/printer_detail.html index 8fb41ee..112e5f1 100644 --- a/imptune/templates/printer_detail.html +++ b/imptune/templates/printer_detail.html @@ -192,16 +192,40 @@ hx-target="#icon-status" hx-swap="innerHTML" hx-encoding="multipart/form-data"> <div class="uploader"> - <input type="file" name="file" accept="image/png" required + <input type="file" name="file" accept="image/*" required x-data :aria-label="$store.i18n.t('upload_icon')"> <div class="btn-row"> <button type="submit" class="btn ghost sm" x-data> {{ ico.i('upload', 13) }}<span x-text="$store.i18n.t('{{ 'replace_icon' if has_icon else 'upload_icon' }}')">Upload icon</span> </button> </div> - <p class="hint" x-data x-text="$store.i18n.t('hint_icon')">PNG, exactly 256 × 256, 750 KB max.</p> + <p class="hint" x-data x-text="$store.i18n.t('hint_any_image')">Any image — resized to 256 × 256 PNG. 750 KB max.</p> </div> </form> + + {% if web_search_enabled() %} + <div class="web-picker"> + <form hx-get="/web/images" + hx-target="#printer-image-results" + hx-swap="innerHTML" + hx-indicator="#printer-image-spinner"> + <input type="hidden" name="target" value="printer"> + <input type="hidden" name="id" value="{{ printer.id }}"> + <label> + <span class="label-text" x-data x-text="$store.i18n.t('search_web_image')">Search the web for an image</span> + <input type="search" name="q" value="{{ image_query }}"> + </label> + <div class="btn-row"> + <button type="submit" class="btn ghost sm" x-data> + {{ ico.i('search', 13) }}<span x-text="$store.i18n.t('search_btn')">Search</span> + </button> + <span id="printer-image-spinner" class="htmx-indicator" aria-busy="true" + x-data x-text="$store.i18n.t('searching')">Searching…</span> + </div> + </form> + <div id="printer-image-results"></div> + </div> + {% endif %} </div> </section> </div> diff --git a/imptune/templating.py b/imptune/templating.py new file mode 100644 index 0000000..6ebb294 --- /dev/null +++ b/imptune/templating.py @@ -0,0 +1,17 @@ +"""The one Jinja2 environment every router renders through. + +Each `api/*` module used to build its own `Jinja2Templates`, which meant a +template global had to be declared five times or a page rendered by the wrong +router would blow up on an undefined name. One instance, one place. +""" +from pathlib import Path + +from fastapi.templating import Jinja2Templates + +from imptune.services import websearch + +templates = Jinja2Templates(directory=str(Path(__file__).parent / "templates")) + +# Passed as the function, not its result: `WEB_SEARCH` is read per call, so a +# test (or a config reload) that flips it takes effect without a restart. +templates.env.globals["web_search_enabled"] = websearch.enabled diff --git a/tests/e2e/test_driver_rename.py b/tests/e2e/test_driver_rename.py new file mode 100644 index 0000000..e659707 --- /dev/null +++ b/tests/e2e/test_driver_rename.py @@ -0,0 +1,84 @@ +"""E2E: the driver library's rename dialog — open, pre-fill, submit, row updates.""" +from __future__ import annotations + +import io +import zipfile + +# Same reasoning as test_printer_edit.py: the button label goes through the i18n +# store, which follows navigator.language, so only a structural hook is portable. +EDIT_BUTTON = "button[onclick*='showModal']" + +INF = """[Version] +Signature="$Windows NT$" +Class=Printer +Provider=%Vendor% + +[Manufacturer] +%Vendor%=Models,NTamd64 + +[Models.NTamd64] +"E2E Rename Printer"=Install,USBPRINT\\E2E + +[Strings] +Vendor="E2E Vendor" +""" + + +def _driver_zip() -> bytes: + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as zf: + zf.writestr("e2e_rename.inf", INF) + return buf.getvalue() + + +def _upload_driver(live_server: str, owner_key: str) -> None: + import httpx + + from imptune.services.session import COOKIE_NAME + + with httpx.Client( + base_url=live_server, follow_redirects=True, cookies={COOKIE_NAME: owner_key} + ) as api: + response = api.post( + "/drivers/upload", + files={"file": ("e2e_rename_pkg.zip", _driver_zip(), "application/zip")}, + ) + assert response.status_code == 200, response.text + + +def test_driver_rename_dialog_prefills_the_zip_name( + page, live_server: str, _e2e_owner_key: str +) -> None: + """With no rename yet, the input is empty and the ZIP name is its placeholder.""" + _upload_driver(live_server, _e2e_owner_key) + + page.goto(f"{live_server}/drivers", wait_until="domcontentloaded") + row = page.locator("tr", has=page.locator("text=e2e_rename_pkg.zip")) + row.first.wait_for() + row.first.locator(EDIT_BUTTON).click() + + page.wait_for_selector("dialog[open]") + field = page.locator("dialog[open] input[name='display_name']") + assert field.input_value() == "" + assert field.get_attribute("placeholder") == "e2e_rename_pkg.zip" + + +def test_driver_rename_updates_the_row(page, live_server: str, _e2e_owner_key: str) -> None: + """Saving swaps the table in place and shows the new label over the ZIP name.""" + _upload_driver(live_server, _e2e_owner_key) + + page.goto(f"{live_server}/drivers", wait_until="domcontentloaded") + row = page.locator("tr", has=page.locator("text=e2e_rename_pkg.zip")) + row.first.wait_for() + row.first.locator(EDIT_BUTTON).click() + page.wait_for_selector("dialog[open]") + + page.fill("dialog[open] input[name='display_name']", "Ground floor MFP") + # The dialog holds several submit buttons (icon upload, web search); the + # footer's Save is the one bound to the rename form. + page.click("dialog[open] .dialog-foot button[type='submit']") + + page.wait_for_selector("text=Ground floor MFP") + assert page.locator("dialog[open]").count() == 0 + # Renaming does not hide what is actually stored on disk. + assert page.locator("text=e2e_rename_pkg.zip").count() >= 1 diff --git a/tests/test_db.py b/tests/test_db.py index 7d6836c..6d0f435 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -26,12 +26,18 @@ def db_env(tmp_path, monkeypatch): if not db.is_closed(): db.close() - return { + yield { "data_dir": data_dir, "drivers_dir": drivers_dir, "db_path": db_path, } + # A connection left open here stays bound to *this* test's file, and the next + # test's ORM writes would land in it instead of its own tmp DB — the app + # thread would then 404 on rows the test just created. + if not db.is_closed(): + db.close() + def test_create_tables(db_env): """init_db() creates all 4 tables in a fresh SQLite file.""" @@ -140,3 +146,42 @@ def test_driver_store_get_path(db_env): path = store.get_path(sha256) assert path == Path(str(drivers_dir)) / f"{sha256}.zip" + + +def test_init_db_adds_display_name_to_a_preexisting_driver_table(db_env): + """A DB created before the rename feature gains the column, keeping its rows. + + `create_tables(safe=True)` skips a table that already exists, so a new field + on an old model only lands through `_add_missing_columns`. + """ + from imptune.db.database import db, init_db + + init_db() + db.execute_sql("ALTER TABLE driver DROP COLUMN display_name") + db.execute_sql( + "INSERT INTO driver (sha256, original_filename, size_bytes, uploaded_at, " + "has_cat_file) VALUES ('legacy', 'old.zip', 10, '2024-01-01 00:00:00', 0)" + ) + db.close() + + init_db() + + columns = {row[1] for row in db.execute_sql("PRAGMA table_info(driver)")} + assert "display_name" in columns + + from imptune.db.models import Driver + + legacy = Driver.get(Driver.sha256 == "legacy") + assert legacy.display_name is None + assert legacy.label == "old.zip" + + +def test_init_db_creates_the_driver_icon_table(db_env): + from imptune.db.database import db, init_db + + init_db() + tables = { + row[0] + for row in db.execute_sql("SELECT name FROM sqlite_master WHERE type='table'") + } + assert "driver_icon" in tables diff --git a/tests/test_driver_icon.py b/tests/test_driver_icon.py new file mode 100644 index 0000000..8d6d9b1 --- /dev/null +++ b/tests/test_driver_icon.py @@ -0,0 +1,227 @@ +"""Driver icons — upload, fetch-from-web, and serve. Global, like Driver rows.""" +from __future__ import annotations + +import io +import json + +import pytest +from PIL import Image + + +def _png(width: int = 256, height: int = 256, color: str = "red") -> bytes: + buf = io.BytesIO() + Image.new("RGBA", (width, height), color=color).save(buf, format="PNG") + return buf.getvalue() + + +def _create_driver(sha: str = "b" * 64): + from imptune.db.models import Driver + + return Driver.create( + sha256=sha, + original_filename="konica_c300i.zip", + size_bytes=4096, + driver_desc=json.dumps(["KONICA MINOLTA C300i PCL"]), + inf_filename="kocpl.inf", + architecture="amd64", + ) + + +class TestDriverIconUpload: + def test_upload_creates_the_record_and_file(self, client, tmp_data_dir): + from imptune.db.models import DriverIcon + + driver = _create_driver() + response = client.post( + f"/drivers/{driver.id}/icon", + files={"file": ("logo.png", io.BytesIO(_png()), "image/png")}, + ) + assert response.status_code == 200 + + icon = DriverIcon.get(DriverIcon.driver == driver.id) + assert icon.original_filename == "logo.png" + assert (tmp_data_dir / "icons" / icon.sha256).exists() + + def test_off_size_image_is_normalized(self, client, tmp_data_dir): + from imptune.db.models import DriverIcon + + driver = _create_driver() + client.post( + f"/drivers/{driver.id}/icon", + files={"file": ("wide.png", io.BytesIO(_png(600, 120)), "image/png")}, + ) + icon = DriverIcon.get(DriverIcon.driver == driver.id) + with Image.open(tmp_data_dir / "icons" / icon.sha256) as img: + assert img.size == (256, 256) + + def test_second_upload_replaces_the_first(self, client, tmp_data_dir): + from imptune.db.models import DriverIcon + + driver = _create_driver() + client.post( + f"/drivers/{driver.id}/icon", + files={"file": ("a.png", io.BytesIO(_png(color="red")), "image/png")}, + ) + client.post( + f"/drivers/{driver.id}/icon", + files={"file": ("b.png", io.BytesIO(_png(color="green")), "image/png")}, + ) + icons = list(DriverIcon.select().where(DriverIcon.driver == driver.id)) + assert len(icons) == 1 + assert icons[0].original_filename == "b.png" + + def test_response_carries_an_oob_row_refresh(self, client, tmp_data_dir): + """The dialog stays open, so the table thumbnail is swapped out of band.""" + driver = _create_driver() + response = client.post( + f"/drivers/{driver.id}/icon", + files={"file": ("logo.png", io.BytesIO(_png()), "image/png")}, + ) + assert f'id="driver-thumb-{driver.id}"' in response.text + assert 'hx-swap-oob="true"' in response.text + + def test_rejects_an_undecodable_file(self, client, tmp_data_dir): + driver = _create_driver() + response = client.post( + f"/drivers/{driver.id}/icon", + files={"file": ("x.png", io.BytesIO(b"nope"), "image/png")}, + ) + assert response.status_code == 422 + + def test_rejects_an_oversized_file(self, client, tmp_data_dir): + driver = _create_driver() + oversized = _png() + b"\x00" * (750 * 1024) + response = client.post( + f"/drivers/{driver.id}/icon", + files={"file": ("big.png", io.BytesIO(oversized), "image/png")}, + ) + assert response.status_code == 422 + assert "750" in response.text + + def test_404_for_unknown_driver(self, client, tmp_data_dir): + response = client.post( + "/drivers/99999/icon", + files={"file": ("logo.png", io.BytesIO(_png()), "image/png")}, + ) + assert response.status_code == 404 + + +class TestDriverIconFromWeb: + def test_downloads_and_stores_the_picked_image(self, client, tmp_data_dir, monkeypatch): + from imptune.db.models import DriverIcon + from imptune.services import websearch + + driver = _create_driver() + monkeypatch.setattr(websearch, "fetch_image", lambda url: _png(400, 400)) + + response = client.post( + f"/drivers/{driver.id}/icon/from-web", + data={"url": "https://example.com/pictures/c300i.png"}, + ) + assert response.status_code == 200 + + icon = DriverIcon.get(DriverIcon.driver == driver.id) + assert icon.original_filename == "c300i.png" + with Image.open(tmp_data_dir / "icons" / icon.sha256) as img: + assert img.size == (256, 256) + + def test_reports_a_refused_url(self, client, tmp_data_dir, monkeypatch): + from imptune.services import websearch + + driver = _create_driver() + + def refuse(url): + raise websearch.WebSearchError("10.0.0.5 resolves to a private address — refused.") + + monkeypatch.setattr(websearch, "fetch_image", refuse) + response = client.post( + f"/drivers/{driver.id}/icon/from-web", data={"url": "http://10.0.0.5/x.png"} + ) + assert response.status_code == 400 + assert "private address" in response.text + + def test_reports_a_page_that_is_not_an_image(self, client, tmp_data_dir, monkeypatch): + from imptune.services import websearch + + driver = _create_driver() + monkeypatch.setattr(websearch, "fetch_image", lambda url: b"<html>404</html>") + response = client.post( + f"/drivers/{driver.id}/icon/from-web", data={"url": "https://example.com/x"} + ) + assert response.status_code == 422 + assert "readable image" in response.text + + +class TestDriverIconServe: + def test_serves_the_png(self, client, tmp_data_dir): + driver = _create_driver() + client.post( + f"/drivers/{driver.id}/icon", + files={"file": ("logo.png", io.BytesIO(_png()), "image/png")}, + ) + response = client.get(f"/drivers/{driver.id}/icon") + assert response.status_code == 200 + assert response.headers["content-type"] == "image/png" + + def test_visible_to_another_owner(self, client, tmp_data_dir): + """The driver library is shared, so its icons are not owner-scoped.""" + from fastapi.testclient import TestClient + + from imptune.main import app + + driver = _create_driver() + client.post( + f"/drivers/{driver.id}/icon", + files={"file": ("logo.png", io.BytesIO(_png()), "image/png")}, + ) + with TestClient(app) as other: + assert other.get(f"/drivers/{driver.id}/icon").status_code == 200 + + def test_404_without_an_icon(self, client, tmp_data_dir): + driver = _create_driver() + assert client.get(f"/drivers/{driver.id}/icon").status_code == 404 + + def test_404_when_the_file_vanished(self, client, tmp_data_dir): + from imptune.db.models import DriverIcon + + driver = _create_driver() + client.post( + f"/drivers/{driver.id}/icon", + files={"file": ("logo.png", io.BytesIO(_png()), "image/png")}, + ) + icon = DriverIcon.get(DriverIcon.driver == driver.id) + (tmp_data_dir / "icons" / icon.sha256).unlink() + assert client.get(f"/drivers/{driver.id}/icon").status_code == 404 + + +def test_library_row_shows_the_thumbnail(client, tmp_data_dir): + driver = _create_driver() + client.post( + f"/drivers/{driver.id}/icon", + files={"file": ("logo.png", io.BytesIO(_png()), "image/png")}, + ) + page = client.get("/drivers") + assert f'src="/drivers/{driver.id}/icon"' in page.text + + +def test_printer_and_driver_icons_share_one_stored_file(client, owner, tmp_data_dir): + """Content-addressed storage: identical bytes land on the same path.""" + from imptune.db.models import DriverIcon, Icon, Printer + + driver = _create_driver() + printer = Printer.create( + name="P", ip_address="10.0.0.1", port_name="IP_10.0.0.1", owner=owner + ) + png = _png(300, 300) + client.post( + f"/drivers/{driver.id}/icon", + files={"file": ("a.png", io.BytesIO(png), "image/png")}, + ) + client.post( + f"/printers/{printer.id}/icon", + files={"file": ("b.png", io.BytesIO(png), "image/png")}, + ) + assert ( + DriverIcon.get(DriverIcon.driver == driver.id).sha256 + == Icon.get(Icon.printer == printer.id).sha256 + ) diff --git a/tests/test_driver_rename.py b/tests/test_driver_rename.py new file mode 100644 index 0000000..0f384ec --- /dev/null +++ b/tests/test_driver_rename.py @@ -0,0 +1,82 @@ +"""Driver rename — PATCH /drivers/{id}. Global/shared, like the Driver row.""" +from __future__ import annotations + +import json + + +def _create_driver(**overrides): + from imptune.db.models import Driver + + fields = { + "sha256": "a" * 64, + "original_filename": "hp_m404_x64.zip", + "size_bytes": 2048, + "driver_desc": json.dumps(["HP LaserJet M404 PCL-6"]), + "inf_filename": "hpm404.inf", + "architecture": "amd64", + } + fields.update(overrides) + return Driver.create(**fields) + + +class TestDriverRename: + def test_sets_display_name(self, client, tmp_data_dir): + from imptune.db.models import Driver + + driver = _create_driver() + response = client.patch( + f"/drivers/{driver.id}", data={"display_name": "HP UPD PCL6 (étage 2)"} + ) + assert response.status_code == 200 + assert Driver.get_by_id(driver.id).display_name == "HP UPD PCL6 (étage 2)" + + def test_renamed_driver_shows_both_names_in_the_list(self, client, tmp_data_dir): + driver = _create_driver() + response = client.patch( + f"/drivers/{driver.id}", data={"display_name": "Accounting MFP"} + ) + assert "Accounting MFP" in response.text + # The ZIP name stays visible as the sub-line — it is what is on disk. + assert "hp_m404_x64.zip" in response.text + + def test_empty_value_clears_the_rename(self, client, tmp_data_dir): + from imptune.db.models import Driver + + driver = _create_driver(display_name="Old label") + client.patch(f"/drivers/{driver.id}", data={"display_name": " "}) + assert Driver.get_by_id(driver.id).display_name is None + + def test_rename_is_visible_to_another_owner(self, client, tmp_data_dir): + """Drivers are shared, so a rename is not scoped to the renamer.""" + from fastapi.testclient import TestClient + + from imptune.main import app + + driver = _create_driver() + client.patch(f"/drivers/{driver.id}", data={"display_name": "Shared label"}) + + with TestClient(app) as other: # fresh cookie jar → a different Owner + page = other.get("/drivers") + assert "Shared label" in page.text + + def test_404_for_unknown_driver(self, client, tmp_data_dir): + response = client.patch("/drivers/99999", data={"display_name": "x"}) + assert response.status_code == 404 + + def test_rejects_an_overlong_name(self, client, tmp_data_dir): + from imptune.db.models import Driver + + driver = _create_driver() + response = client.patch( + f"/drivers/{driver.id}", data={"display_name": "x" * 121} + ) + assert response.status_code == 400 + assert Driver.get_by_id(driver.id).display_name is None + + def test_missing_field_clears_rather_than_erroring(self, client, tmp_data_dir): + from imptune.db.models import Driver + + driver = _create_driver(display_name="Old label") + response = client.patch(f"/drivers/{driver.id}", data={}) + assert response.status_code == 200 + assert Driver.get_by_id(driver.id).display_name is None diff --git a/tests/test_icon_upload.py b/tests/test_icon_upload.py index 67cd363..8f8cb39 100644 --- a/tests/test_icon_upload.py +++ b/tests/test_icon_upload.py @@ -66,16 +66,35 @@ class TestIconUpload: icon_file = Path(tmp_data_dir) / "icons" / sha256 assert icon_file.exists() - def test_reject_non_png(self, client, owner, tmp_data_dir): - """POST with a JPEG file returns 422 with PNG format error.""" + def test_jpeg_is_converted(self, client, owner, tmp_data_dir): + """A JPEG is accepted and re-encoded as a 256x256 PNG, not rejected.""" + from pathlib import Path + + from imptune.db.models import Icon + printer = _create_printer(owner) jpeg_data = _make_jpeg(256, 256) response = client.post( f"/printers/{printer.id}/icon", files={"file": ("icon.jpg", io.BytesIO(jpeg_data), "image/jpeg")}, ) + assert response.status_code == 200 + + icon = Icon.get(Icon.printer == printer.id) + stored = Path(tmp_data_dir) / "icons" / icon.sha256 + with Image.open(stored) as img: + assert img.format == "PNG" + assert img.size == (256, 256) + + def test_reject_undecodable_file(self, client, owner, tmp_data_dir): + """A file Pillow cannot open is still refused.""" + printer = _create_printer(owner) + response = client.post( + f"/printers/{printer.id}/icon", + files={"file": ("icon.png", io.BytesIO(b"not an image at all"), "image/png")}, + ) assert response.status_code == 422 - assert "PNG" in response.text + assert "readable image" in response.text def test_reject_oversized(self, client, owner, tmp_data_dir): """POST with PNG > 750KB returns 422 with 750 KB error.""" @@ -90,16 +109,24 @@ class TestIconUpload: assert response.status_code == 422 assert "750" in response.text - def test_reject_wrong_dimensions(self, client, owner, tmp_data_dir): - """POST with 128x128 PNG returns 422 with 256x256 error.""" + def test_wrong_dimensions_are_resized(self, client, owner, tmp_data_dir): + """An off-size PNG is letterboxed into 256x256 instead of rejected.""" + from pathlib import Path + + from imptune.db.models import Icon + printer = _create_printer(owner) - png_data = _make_png(128, 128) + png_data = _make_png(128, 400) response = client.post( f"/printers/{printer.id}/icon", - files={"file": ("small.png", io.BytesIO(png_data), "image/png")}, + files={"file": ("tall.png", io.BytesIO(png_data), "image/png")}, ) - assert response.status_code == 422 - assert "256x256" in response.text + assert response.status_code == 200 + + icon = Icon.get(Icon.printer == printer.id) + stored = Path(tmp_data_dir) / "icons" / icon.sha256 + with Image.open(stored) as img: + assert img.size == (256, 256) def test_replace_existing_icon(self, client, owner, tmp_data_dir): """Second upload for same printer replaces the Icon record.""" diff --git a/tests/test_image_utils.py b/tests/test_image_utils.py new file mode 100644 index 0000000..cd67b80 --- /dev/null +++ b/tests/test_image_utils.py @@ -0,0 +1,66 @@ +"""Icon normalization — anything decodable becomes a 256x256 PNG.""" +from __future__ import annotations + +import io + +import pytest +from PIL import Image + +from imptune.services.image_utils import ICON_SIZE, ImageError, normalize_icon + + +def _png(width: int, height: int, mode: str = "RGBA") -> bytes: + buf = io.BytesIO() + Image.new(mode, (width, height), color="red").save(buf, format="PNG") + return buf.getvalue() + + +def _jpeg(width: int, height: int) -> bytes: + buf = io.BytesIO() + Image.new("RGB", (width, height), color="blue").save(buf, format="JPEG") + return buf.getvalue() + + +def test_exact_png_passes_through_byte_identical(): + """The icon store is content-addressed — re-encoding would move the file.""" + data = _png(*ICON_SIZE) + assert normalize_icon(data) is data + + +@pytest.mark.parametrize( + "source", + [_png(64, 64), _png(1024, 1024), _png(1024, 128), _jpeg(300, 200)], + ids=["small", "large", "wide", "jpeg"], +) +def test_everything_else_becomes_a_256_png(source): + out = normalize_icon(source) + with Image.open(io.BytesIO(out)) as img: + assert img.format == "PNG" + assert img.size == ICON_SIZE + + +def test_aspect_ratio_is_kept_not_stretched(): + """A 400x100 source keeps its 4:1 shape, letterboxed in a square canvas. + + Checked through the alpha channel: the padding stays fully transparent, so + the opaque band is 64px tall in a 256px canvas. + """ + out = normalize_icon(_png(400, 100)) + with Image.open(io.BytesIO(out)) as img: + alpha = img.convert("RGBA").split()[3] + opaque_rows = [ + y for y in range(256) if any(alpha.getpixel((x, y)) for x in range(256)) + ] + assert len(opaque_rows) == 64 + # ...and it is centered, not flush to the top. + assert opaque_rows[0] == 96 + + +def test_undecodable_bytes_raise(): + with pytest.raises(ImageError): + normalize_icon(b"this is not an image") + + +def test_empty_bytes_raise(): + with pytest.raises(ImageError): + normalize_icon(b"") diff --git a/tests/test_web_routes.py b/tests/test_web_routes.py new file mode 100644 index 0000000..8ff93d3 --- /dev/null +++ b/tests/test_web_routes.py @@ -0,0 +1,267 @@ +"""/web/images and /web/drivers — the HTMX fragments behind the search UIs. + +`websearch.search_images` / `search_pages` are monkeypatched throughout: these +tests are about routing, ownership, and rendering, not about the scrape. +""" +from __future__ import annotations + +import io +import json + +import pytest +from PIL import Image + +from imptune.services.websearch import ImageHit, PageHit + + +@pytest.fixture +def stub_images(monkeypatch): + """Record the query the route asked for, return two canned hits.""" + calls: list[tuple[str, int]] = [] + + def fake(query, limit=12): + calls.append((query, limit)) + return [ + ImageHit( + url="https://cdn.example.com/a.png", + thumbnail="https://tn.example.com/a.png", + title="Printer A", + width=800, + height=600, + source="cdn.example.com", + ), + ImageHit( + url="https://cdn.example.com/b.jpg", + thumbnail="https://tn.example.com/b.jpg", + title="Printer B", + width=400, + height=400, + source="cdn.example.com", + ), + ] + + from imptune.api import web + + monkeypatch.setattr(web.websearch, "search_images", fake) + return calls + + +@pytest.fixture +def stub_pages(monkeypatch): + calls: list[tuple[str, int]] = [] + + def fake(query, limit=8): + calls.append((query, limit)) + return [ + PageHit( + url="https://support.hp.com/upd", + title="HP Universal Print Driver", + snippet="Download the UPD for Windows.", + host="support.hp.com", + ) + ] + + from imptune.api import web + + monkeypatch.setattr(web.websearch, "search_pages", fake) + return calls + + +def _printer(owner): + from imptune.db.models import Printer + + return Printer.create( + name="HP LaserJet M404", + ip_address="10.0.0.7", + port_name="IP_10.0.0.7", + owner=owner, + ) + + +def _driver(): + from imptune.db.models import Driver + + return Driver.create( + sha256="c" * 64, + original_filename="hp_upd.zip", + size_bytes=1024, + driver_desc=json.dumps(["HP Universal Printing PCL 6"]), + ) + + +class TestImageSearchRoute: + def test_renders_pickable_results_for_a_printer(self, client, owner, tmp_data_dir, stub_images): + printer = _printer(owner) + response = client.get( + "/web/images", params={"q": "hp m404 printer", "target": "printer", "id": printer.id} + ) + assert response.status_code == 200 + assert stub_images == [("hp m404 printer", 12)] + assert f'hx-post="/printers/{printer.id}/icon/from-web"' in response.text + assert 'hx-target="#icon-status"' in response.text + assert "https://cdn.example.com/a.png" in response.text + # Thumbnails load from the engine's CDN — no referrer leak. + assert 'referrerpolicy="no-referrer"' in response.text + + def test_renders_pickable_results_for_a_driver(self, client, owner, tmp_data_dir, stub_images): + driver = _driver() + response = client.get( + "/web/images", params={"q": "hp upd", "target": "driver", "id": driver.id} + ) + assert response.status_code == 200 + assert f'hx-post="/drivers/{driver.id}/icon/from-web"' in response.text + assert f'hx-target="#driver-icon-status-{driver.id}"' in response.text + + def test_another_owners_printer_reads_as_missing(self, client, owner, tmp_data_dir, stub_images): + """404, not 403 — and no search is spent on it.""" + from imptune.db.models import Owner, Printer + from imptune.services.session import generate_key + + stranger = Owner.create(key=generate_key()) + theirs = Printer.create( + name="Theirs", ip_address="10.0.0.9", port_name="IP_10_0_0_9", owner=stranger + ) + response = client.get( + "/web/images", params={"q": "x", "target": "printer", "id": theirs.id} + ) + assert response.status_code == 404 + assert stub_images == [] + + def test_unknown_driver_is_404(self, client, tmp_data_dir, stub_images): + response = client.get( + "/web/images", params={"q": "x", "target": "driver", "id": 99999} + ) + assert response.status_code == 404 + + def test_unknown_target_is_400(self, client, tmp_data_dir, stub_images): + response = client.get("/web/images", params={"q": "x", "target": "wat", "id": 1}) + assert response.status_code == 400 + + def test_blank_query_does_not_search(self, client, owner, tmp_data_dir, stub_images): + printer = _printer(owner) + response = client.get( + "/web/images", params={"q": " ", "target": "printer", "id": printer.id} + ) + assert response.status_code == 200 + assert stub_images == [] + + def test_no_results_renders_a_hint(self, client, owner, tmp_data_dir, monkeypatch): + from imptune.api import web + + monkeypatch.setattr(web.websearch, "search_images", lambda q, limit=12: []) + printer = _printer(owner) + response = client.get( + "/web/images", params={"q": "zzz", "target": "printer", "id": printer.id} + ) + assert response.status_code == 200 + assert "No results" in response.text + + def test_disabled_deployment_says_so(self, client, owner, tmp_data_dir, monkeypatch): + import imptune.config as cfg + + monkeypatch.setattr(cfg, "WEB_SEARCH", False) + printer = _printer(owner) + response = client.get( + "/web/images", params={"q": "x", "target": "printer", "id": printer.id} + ) + assert response.status_code == 200 + assert "disabled" in response.text + + +class TestDriverSearchRoute: + def test_generic_mode_rewrites_the_query(self, client, tmp_data_dir, stub_pages): + response = client.get( + "/web/drivers", params={"q": "HP LaserJet M404dn", "mode": "generic"} + ) + assert response.status_code == 200 + assert stub_pages == [("HP Universal Print Driver PCL6 download", 12)] + assert "support.hp.com/upd" in response.text + + def test_exact_mode_searches_the_text_as_typed(self, client, tmp_data_dir, stub_pages): + response = client.get( + "/web/drivers", params={"q": "HP LaserJet M404dn", "mode": "exact"} + ) + assert stub_pages == [("HP LaserJet M404dn", 12)] + + def test_results_carry_the_unvetted_warning(self, client, tmp_data_dir, stub_pages): + """The links are raw search results — the fragment must say so.""" + response = client.get("/web/drivers", params={"q": "hp", "mode": "generic"}) + assert "driver_search_warning" in response.text + assert 'rel="noopener noreferrer nofollow"' in response.text + assert 'target="_blank"' in response.text + + def test_blank_query_does_not_search(self, client, tmp_data_dir, stub_pages): + response = client.get("/web/drivers", params={"q": " "}) + assert response.status_code == 200 + assert stub_pages == [] + + def test_disabled_deployment_says_so(self, client, tmp_data_dir, monkeypatch): + import imptune.config as cfg + + monkeypatch.setattr(cfg, "WEB_SEARCH", False) + response = client.get("/web/drivers", params={"q": "hp"}) + assert "disabled" in response.text + + +class TestSearchUiVisibility: + def test_drivers_page_offers_the_search_card(self, client, tmp_data_dir): + page = client.get("/drivers") + assert 'hx-get="/web/drivers"' in page.text + assert "Konica Minolta" in page.text # brand datalist + + def test_printer_detail_offers_the_image_picker(self, client, owner, tmp_data_dir): + printer = _printer(owner) + page = client.get(f"/printers/{printer.id}") + assert 'hx-get="/web/images"' in page.text + # Prefilled from the printer name, so one click is enough. + assert 'value="HP LaserJet M404 printer"' in page.text + + def test_every_search_control_disappears_when_disabled( + self, client, owner, tmp_data_dir, monkeypatch + ): + import imptune.config as cfg + + monkeypatch.setattr(cfg, "WEB_SEARCH", False) + printer = _printer(owner) + + assert 'hx-get="/web/drivers"' not in client.get("/drivers").text + assert 'hx-get="/web/images"' not in client.get(f"/printers/{printer.id}").text + + +def test_picked_image_becomes_the_printer_icon(client, owner, tmp_data_dir, monkeypatch): + """End to end through the picker's POST: download, normalize, store, preview.""" + from imptune.db.models import Icon + from imptune.services import websearch + + printer = _printer(owner) + buf = io.BytesIO() + Image.new("RGB", (500, 300), color="orange").save(buf, format="JPEG") + monkeypatch.setattr(websearch, "fetch_image", lambda url: buf.getvalue()) + + response = client.post( + f"/printers/{printer.id}/icon/from-web", + data={"url": "https://cdn.example.com/photos/m404.jpg"}, + ) + assert response.status_code == 200 + assert f'src="/printers/{printer.id}/icon?v=' in response.text + + icon = Icon.get(Icon.printer == printer.id) + with Image.open(tmp_data_dir / "icons" / icon.sha256) as img: + assert img.format == "PNG" + assert img.size == (256, 256) + + +def test_from_web_rejects_an_unowned_printer(client, tmp_data_dir, monkeypatch): + from imptune.db.models import Owner, Printer + from imptune.services import websearch + from imptune.services.session import generate_key + + monkeypatch.setattr(websearch, "fetch_image", lambda url: b"unused") + stranger = Owner.create(key=generate_key()) + theirs = Printer.create( + name="Theirs", ip_address="10.0.0.9", port_name="IP_10_0_0_9", owner=stranger + ) + response = client.post( + f"/printers/{theirs.id}/icon/from-web", data={"url": "https://example.com/x.png"} + ) + assert response.status_code == 404 diff --git a/tests/test_websearch.py b/tests/test_websearch.py new file mode 100644 index 0000000..e8dfb64 --- /dev/null +++ b/tests/test_websearch.py @@ -0,0 +1,260 @@ +"""Web lookup service — scrape parsing, query building, and the SSRF guard. + +Nothing here touches the network: `_get` is monkeypatched, and the one test that +needs a "public" hostname fakes the DNS answer. +""" +from __future__ import annotations + +import json + +import pytest + +from imptune.services import websearch + +# Captured before any monkeypatching so `TestFetchGuard` can put the real guard +# back after the module-wide autouse fixture has stubbed it out. +_REAL_ASSERT_FETCHABLE = websearch.assert_fetchable + + +@pytest.fixture(autouse=True) +def _allow_any_host(monkeypatch): + """Neutralize the SSRF guard for parsing tests — it has its own tests below.""" + monkeypatch.setattr(websearch, "assert_fetchable", lambda url: None) + + +def _canned(monkeypatch, mapping: dict[str, bytes]): + """Serve `_get` from a {url-substring: body} table.""" + + def fake_get(url, *, referer=None, max_bytes=0, extra_headers=None): + for fragment, body in mapping.items(): + if fragment in url: + return body + raise websearch.WebSearchError(f"unexpected url {url}") + + monkeypatch.setattr(websearch, "_get", fake_get) + + +# -------------------------------------------------------------------------- +# Image search +# -------------------------------------------------------------------------- + +IMAGE_PAYLOAD = json.dumps( + { + "results": [ + { + "image": "https://cdn.example.com/m404.png", + "thumbnail": "https://tn.example.com/m404.png", + "title": "HP LaserJet M404", + "width": 800, + "height": 600, + "url": "https://example.com/page", + }, + { + "image": "data:image/png;base64,AAAA", + "thumbnail": "https://tn.example.com/skip.png", + "title": "inline", + "width": 1, + "height": 1, + }, + { + "image": "https://cdn.example.com/second.jpg", + "thumbnail": "", + "title": "", + "width": None, + "height": None, + }, + ] + } +).encode() + + +def test_search_images_parses_results(monkeypatch): + _canned( + monkeypatch, + {"duckduckgo.com/?q=": b'... vqd="4-12345678" ...', "i.js": IMAGE_PAYLOAD}, + ) + hits = websearch.search_images("hp laserjet m404") + + # The data: URI is dropped — only fetchable http(s) images survive. + assert [h.url for h in hits] == [ + "https://cdn.example.com/m404.png", + "https://cdn.example.com/second.jpg", + ] + assert hits[0].width == 800 + assert hits[0].source == "cdn.example.com" + # A result with no thumbnail falls back to the full image. + assert hits[1].thumbnail == "https://cdn.example.com/second.jpg" + assert hits[1].width == 0 + + +def test_search_images_honours_limit(monkeypatch): + _canned( + monkeypatch, + {"duckduckgo.com/?q=": b'vqd="4-12345678"', "i.js": IMAGE_PAYLOAD}, + ) + assert len(websearch.search_images("x", limit=1)) == 1 + + +def test_search_images_returns_empty_when_token_is_missing(monkeypatch): + """A markup change upstream must degrade to "no results", not a 500.""" + _canned(monkeypatch, {"duckduckgo.com/?q=": b"<html>redesigned</html>"}) + assert websearch.search_images("anything") == [] + + +def test_search_images_returns_empty_on_bad_json(monkeypatch): + _canned( + monkeypatch, + {"duckduckgo.com/?q=": b'vqd="4-12345678"', "i.js": b"<!doctype html>"}, + ) + assert websearch.search_images("anything") == [] + + +def test_search_images_skips_the_request_for_a_blank_query(monkeypatch): + def explode(*args, **kwargs): + raise AssertionError("should not hit the network") + + monkeypatch.setattr(websearch, "_get", explode) + assert websearch.search_images(" ") == [] + + +# -------------------------------------------------------------------------- +# Page search +# -------------------------------------------------------------------------- + +HTML_RESULTS = b""" +<div class="result"> + <a rel="nofollow" class="result__a" + href="//duckduckgo.com/l/?uddg=https%3A%2F%2Fsupport.hp.com%2Fupd&rut=abc"> + HP <b>Universal Print Driver</b> + </a> + <a class="result__snippet">Download the <b>UPD</b> for Windows.</a> +</div> +<div class="result"> + <a rel="nofollow" class="result__a" href="https://direct.example.com/drivers">Direct link</a> + <a class="result__snippet">Second snippet</a> +</div> +""" + + +def test_search_pages_unwraps_the_redirect_and_strips_markup(monkeypatch): + _canned(monkeypatch, {"html.duckduckgo.com": HTML_RESULTS}) + hits = websearch.search_pages("hp upd") + + assert hits[0].url == "https://support.hp.com/upd" + assert hits[0].title == "HP Universal Print Driver" + assert hits[0].snippet == "Download the UPD for Windows." + assert hits[0].host == "support.hp.com" + assert hits[1].url == "https://direct.example.com/drivers" + + +def test_search_pages_returns_empty_on_transport_failure(monkeypatch): + def fail(*args, **kwargs): + raise websearch.WebSearchError("boom") + + monkeypatch.setattr(websearch, "_get", fail) + assert websearch.search_pages("hp upd") == [] + + +# -------------------------------------------------------------------------- +# SSRF guard +# -------------------------------------------------------------------------- + +class TestFetchGuard: + @pytest.fixture(autouse=True) + def _real_guard(self, monkeypatch): + monkeypatch.setattr(websearch, "assert_fetchable", _REAL_ASSERT_FETCHABLE) + + @pytest.mark.parametrize( + "url", + [ + "file:///etc/passwd", + "ftp://example.com/x", + "gopher://example.com", + ], + ) + def test_non_http_schemes_refused(self, url): + with pytest.raises(websearch.WebSearchError): + websearch.assert_fetchable(url) + + @pytest.mark.parametrize( + "url", + [ + "http://127.0.0.1/admin", + "http://localhost:8000/", + "http://10.0.0.5/printer", + "http://192.168.1.1/", + "http://169.254.169.254/latest/meta-data/", + "http://[::1]/", + ], + ) + def test_private_targets_refused(self, url): + """ImpTune sits on the printer LAN — this is the whole point of the guard.""" + with pytest.raises(websearch.WebSearchError, match="private|resolve"): + websearch.assert_fetchable(url) + + def test_public_target_allowed(self, monkeypatch): + monkeypatch.setattr( + websearch.socket, + "getaddrinfo", + lambda host, port: [(2, 1, 6, "", ("93.184.216.34", 0))], + ) + websearch.assert_fetchable("https://example.com/logo.png") + + def test_a_host_resolving_to_both_is_refused(self, monkeypatch): + """One private answer in the set is enough to refuse the whole host.""" + monkeypatch.setattr( + websearch.socket, + "getaddrinfo", + lambda host, port: [ + (2, 1, 6, "", ("93.184.216.34", 0)), + (2, 1, 6, "", ("127.0.0.1", 0)), + ], + ) + with pytest.raises(websearch.WebSearchError): + websearch.assert_fetchable("https://rebind.example.com/x") + + +# -------------------------------------------------------------------------- +# Query builders and the kill switch +# -------------------------------------------------------------------------- + +@pytest.mark.parametrize( + ("typed", "expected"), + [ + ("HP LaserJet M404dn", "HP Universal Print Driver PCL6 download"), + ("bizhub C300i (Konica Minolta)", "Konica Minolta Universal PCL Print Driver download"), + ("Xerox VersaLink C405", "Xerox Global Print Driver download"), + ], +) +def test_generic_driver_query_uses_the_vendor_product_name(typed, expected): + assert websearch.generic_driver_query(typed) == expected + + +def test_generic_driver_query_falls_back_to_the_typed_model(): + assert ( + websearch.generic_driver_query("Acme 9000") + == "Acme 9000 universal print driver download" + ) + + +def test_detect_brand_prefers_the_longest_match(): + """"konica" and "konica minolta" both match — the specific one wins.""" + assert websearch.detect_brand("Konica Minolta bizhub") == "konica minolta" + assert websearch.detect_brand("Acme 9000") is None + + +def test_image_query_appends_printer(): + assert websearch.image_query("HP M404") == "HP M404 printer" + assert websearch.image_query(" ") == "" + + +def test_disabled_flag_blocks_every_lookup(monkeypatch): + monkeypatch.setattr(websearch.cfg, "WEB_SEARCH", False) + assert websearch.enabled() is False + for call in ( + lambda: websearch.search_images("x"), + lambda: websearch.search_pages("x"), + lambda: websearch.fetch_image("https://example.com/x.png"), + ): + with pytest.raises(websearch.WebSearchError, match="disabled"): + call()