feat: driver rename, driver icons, web image + driver search

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
`<dialog>` 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) <noreply@anthropic.com>
This commit is contained in:
2026-08-05 10:44:47 +02:00
co-authored by Claude Opus 5
parent 3f9cd1f266
commit 2c06806814
34 changed files with 2516 additions and 162 deletions
+3
View File
@@ -15,3 +15,6 @@ venv/
/data/ /data/
imptune_data/ imptune_data/
*.intunewin *.intunewin
# Playwright MCP session artifacts (snapshots / screenshots)
.playwright-mcp/
+46 -2
View File
@@ -43,11 +43,21 @@ ImpTune make printer deploy packages (`.intunewin` for Intune, `.zip` for NinjaR
**Request flow:** **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) 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`) 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) 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:** **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/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/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 - `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. **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
`"<typed> 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 `<dialog>` 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. **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 **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) | | `DATA_DIR` | `/data` | Storage root (DB + drivers + icons) |
| `PORT` | `8000` | Server port | | `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` | `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: `COOKIE_SECURE=false` degrades the session instead of weakening the credential:
+4
View File
@@ -20,6 +20,10 @@ services:
# Or drop sessions entirely — no cookie, one shared store for everyone who # Or drop sessions entirely — no cookie, one shared store for everyone who
# can reach the app. Test boxes / single-person local prod only. # can reach the app. Test boxes / single-person local prod only.
# - COOKIE_SECURE=single_user # - 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: volumes:
imptune_data: imptune_data:
+1 -7
View File
@@ -1,21 +1,15 @@
"""Client CRUD API — POST /clients, GET /clients.""" """Client CRUD API — POST /clients, GET /clients."""
from __future__ import annotations from __future__ import annotations
from pathlib import Path
from fastapi import APIRouter, Form, Request from fastapi import APIRouter, Form, Request
from fastapi.responses import HTMLResponse from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from peewee import IntegrityError from peewee import IntegrityError
from imptune.db.models import Client from imptune.db.models import Client
from imptune.templating import templates
router = APIRouter(prefix="/clients") router = APIRouter(prefix="/clients")
templates = Jinja2Templates(
directory=str(Path(__file__).parent.parent / "templates")
)
def _error_response(message: str, status_code: int = 400) -> HTMLResponse: def _error_response(message: str, status_code: int = 400) -> HTMLResponse:
"""Return an HTMX-friendly error fragment swapped into #client-list.""" """Return an HTMX-friendly error fragment swapped into #client-list."""
+143 -15
View File
@@ -1,27 +1,31 @@
"""Driver upload API — POST /drivers/upload.""" """Driver API — upload, rename, and driver-library icons."""
from __future__ import annotations from __future__ import annotations
import io import io
import json import json
import zipfile import zipfile
from pathlib import Path
from fastapi import APIRouter, Form, Request, UploadFile from fastapi import APIRouter, Form, Request, UploadFile
from fastapi.responses import HTMLResponse from fastapi.responses import FileResponse, HTMLResponse, Response
from fastapi.templating import Jinja2Templates
import imptune.config as _cfg 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.services.inf_parser import _detect_encoding, parse_inf
from imptune.storage.driver_store import DriverStore from imptune.storage.driver_store import DriverStore
from imptune.templating import templates
router = APIRouter(prefix="/drivers") router = APIRouter(prefix="/drivers")
templates = Jinja2Templates(
directory=str(Path(__file__).parent.parent / "templates")
)
MAX_UPLOAD_BYTES = 100 * 1024 * 1024 # 100 MB MAX_UPLOAD_BYTES = 100 * 1024 * 1024 # 100 MB
MAX_DISPLAY_NAME = 120
def _error_response(message: str, status_code: int = 400) -> HTMLResponse: 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) @router.post("/upload", response_class=HTMLResponse)
def upload_driver( def upload_driver(
request: Request, request: Request,
@@ -103,11 +120,7 @@ def upload_driver(
) )
# Build driver_data for template # Build driver_data for template
drivers = list(Driver.select().order_by(Driver.uploaded_at.desc())) driver_data = build_driver_data(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})
# When called from the printer form, emit primary fragment + OOB select refresh # When called from the printer form, emit primary fragment + OOB select refresh
if caller == "printer_form": if caller == "printer_form":
@@ -128,5 +141,120 @@ def upload_driver(
context={ context={
"driver_data": driver_data, "driver_data": driver_data,
"parsed": parsed, "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 `<dialog>` out of the DOM.
"""
src = f"/drivers/{driver_id}/icon?v={sha256[:8]}"
return HTMLResponse(
content=(
'<p class="ok-note" x-data x-text="$store.i18n.t(\'icon_uploaded\')">'
"Icon saved</p>"
f'<div class="icon-preview"><img src="{src}" width="56" height="56" alt="">'
'<span class="meta">256&times;256 PNG</span></div>'
f'<span id="driver-thumb-{driver_id}" class="driver-thumb" hx-swap-oob="true">'
f'<img src="{src}" width="24" height="24" alt=""></span>'
),
status_code=200,
)
def _icon_error(message: str, status_code: int = 422) -> HTMLResponse:
return HTMLResponse(
content=f"<p class='error-note'>{message}</p>", 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"},
)
+75 -81
View File
@@ -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 from __future__ import annotations
import hashlib from fastapi import APIRouter, Form, Request, UploadFile
import io
from pathlib import Path
from fastapi import APIRouter, Request, UploadFile
from fastapi.responses import FileResponse, HTMLResponse, Response 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.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") 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.
'<p class="ok-note" x-data x-text="$store.i18n.t(\'icon_uploaded\')">'
"Icon uploaded successfully</p>"
f'<div class="icon-preview"><img src="/printers/{printer_id}/icon?v={sha256[:8]}"'
' width="56" height="56" alt="">'
'<span class="meta">256&times;256 PNG</span></div>'
),
status_code=200,
)
def _error(message: str, status_code: int = 422) -> HTMLResponse:
return HTMLResponse(content=f"<p class='error-note'>{message}</p>", 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) @router.post("/{printer_id}/icon", response_class=HTMLResponse)
def upload_icon(request: Request, printer_id: int, file: UploadFile) -> 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: Any raster Pillow can decode is accepted and letterboxed into the box —
- Format: PNG only only an unreadable file or one over 750 KB is rejected. Replaces the
- Dimensions: exactly 256x256 pixels printer's previous icon.
- Size: at most 750 KB
Replaces any previously uploaded icon for this printer.
Returns an HTMX-friendly HTML fragment.
""" """
# Check printer exists and belongs to this owner if _owned_printer(request, printer_id) is None:
printer = Printer.get_or_none( return HTMLResponse(content="<p>Printer not found.</p>", status_code=404)
(Printer.id == printer_id) & (Printer.owner == request.state.owner)
)
if printer is None:
return HTMLResponse(
content="<p>Printer not found.</p>",
status_code=404,
)
# Read file (read one byte extra to detect oversized files)
data = file.file.read(MAX_ICON_BYTES + 1) data = file.file.read(MAX_ICON_BYTES + 1)
if len(data) > MAX_ICON_BYTES:
return HTMLResponse(
content="<p>Icon exceeds 750 KB limit.</p>",
status_code=422,
)
# Validate with Pillow
try: try:
img = Image.open(io.BytesIO(data)) icon = set_printer_icon(printer_id, data, file.filename or "icon.png")
except Exception: except IconTooLarge as exc:
return HTMLResponse( return _error(str(exc))
content="<p>Icon must be PNG format.</p>", except ImageError:
status_code=422, return _error("That file is not a readable image.")
)
if img.format != "PNG": return _status_fragment(printer_id, icon.sha256)
return HTMLResponse(
content="<p>Icon must be PNG format.</p>",
status_code=422,
)
if img.size != (256, 256):
return HTMLResponse(
content=f"<p>Icon must be 256x256 pixels, got {img.size}.</p>",
status_code=422,
)
# Store SHA256-addressed on disk. cfg.ICONS_DIR is the same path the @router.post("/{printer_id}/icon/from-web", response_class=HTMLResponse)
# .intunewin export reads from — deriving it a second time from DATA_DIR def upload_icon_from_web(
# would silently diverge if the layout ever changes. request: Request, printer_id: int, url: str = Form(...)
sha256 = hashlib.sha256(data).hexdigest() ) -> HTMLResponse:
icons_dir = Path(cfg.ICONS_DIR) """Download a search-result image server-side and use it as the icon."""
icons_dir.mkdir(parents=True, exist_ok=True) if _owned_printer(request, printer_id) is None:
icon_path = icons_dir / sha256 return HTMLResponse(content="<p>Printer not found.</p>", status_code=404)
icon_path.write_bytes(data)
# Replace existing Icon record for this printer try:
Icon.delete().where(Icon.printer == printer_id).execute() data = websearch.fetch_image(url)
Icon.create( except websearch.WebSearchError as exc:
printer=printer_id, return _error(str(exc), status_code=400)
sha256=sha256,
original_filename=file.filename or "icon.png",
size_bytes=len(data),
)
return HTMLResponse( try:
content=( icon = set_printer_icon(printer_id, data, url.rsplit("/", 1)[-1][:120] or "web.png")
'<p class="ok-note">Icon uploaded successfully</p>' except IconTooLarge as exc:
f'<div class="icon-preview"><img src="/printers/{printer_id}/icon?v={sha256[:8]}"' return _error(str(exc))
f' width="56" height="56" alt=""><span class="meta">{img.size[0]}&times;{img.size[1]} PNG</span></div>' except ImageError:
), return _error("That URL did not return a readable image.")
status_code=200,
) return _status_fragment(printer_id, icon.sha256)
@router.get("/{printer_id}/icon") @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. Owner-scoped: a printer belonging to another owner reads as missing.
""" """
printer = Printer.get_or_none( if _owned_printer(request, printer_id) is None:
(Printer.id == printer_id) & (Printer.owner == request.state.owner)
)
if printer is None:
return Response(status_code=404) return Response(status_code=404)
icon = Icon.get_or_none(Icon.printer == printer_id) icon = Icon.get_or_none(Icon.printer == printer_id)
if icon is None: if icon is None:
return Response(status_code=404) return Response(status_code=404)
icon_path = Path(cfg.ICONS_DIR) / icon.sha256 path = icon_path(icon.sha256)
if not icon_path.exists(): if not path.exists():
return Response(status_code=404) return Response(status_code=404)
return FileResponse( return FileResponse(
icon_path, path,
media_type="image/png", media_type="image/png",
headers={"Cache-Control": "private, max-age=300"}, headers={"Cache-Control": "private, max-age=300"},
) )
+35 -14
View File
@@ -1,17 +1,14 @@
import json import json
from collections import defaultdict from collections import defaultdict
from pathlib import Path
from fastapi import APIRouter, Request from fastapi import APIRouter, Request
from fastapi.responses import HTMLResponse from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from peewee import JOIN, fn from peewee import JOIN, fn
from imptune.templating import templates
router = APIRouter() router = APIRouter()
templates = Jinja2Templates(directory=str(Path(__file__).parent.parent / "templates"))
# Group label for printers with no client — the template translates it. # Group label for printers with no client — the template translates it.
UNASSIGNED_GROUP = "Unassigned" UNASSIGNED_GROUP = "Unassigned"
@@ -57,12 +54,31 @@ def group_printers_by_client(printers) -> dict[str, list]:
return ordered return ordered
def _driver_data(drivers) -> list[dict]: def build_driver_data(drivers) -> list[dict]:
"""Attach the parsed driver names to each Driver row for template use.""" """Attach parsed driver names + icon presence to each Driver row.
return [
{"driver": d, "names": json.loads(d.driver_desc) if d.driver_desc else []} Shared by every template that lists drivers (library table, printer form
for d in drivers 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) @router.get("/", response_class=HTMLResponse)
@@ -109,13 +125,16 @@ def dashboard(request: Request):
def drivers_page(request: Request): def drivers_page(request: Request):
from imptune.db.models import Driver from imptune.db.models import Driver
from imptune.services.websearch import BRAND_SUGGESTIONS
drivers = list(Driver.select().order_by(Driver.uploaded_at.desc())) drivers = list(Driver.select().order_by(Driver.uploaded_at.desc()))
return templates.TemplateResponse( return templates.TemplateResponse(
request=request, request=request,
name="drivers.html", name="drivers.html",
context={ context={
"driver_data": _driver_data(drivers), "driver_data": build_driver_data(drivers),
"usage": printer_counts_by_driver(request.state.owner), "usage": printer_counts_by_driver(request.state.owner),
"brands": BRAND_SUGGESTIONS,
}, },
) )
@@ -145,7 +164,7 @@ def printers_page(request: Request):
context={ context={
"grouped": grouped, "grouped": grouped,
"clients": clients, "clients": clients,
"driver_data": _driver_data(all_drivers), "driver_data": build_driver_data(all_drivers),
"printer_count": printer_count, "printer_count": printer_count,
"ready_count": ready_count, "ready_count": ready_count,
}, },
@@ -161,13 +180,14 @@ def printers_new_page(request: Request):
return templates.TemplateResponse( return templates.TemplateResponse(
request=request, request=request,
name="printers_new.html", 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) @router.get("/printers/{printer_id}", response_class=HTMLResponse)
def printer_detail(request: Request, printer_id: int): def printer_detail(request: Request, printer_id: int):
from imptune.db.models import Client, Driver, Icon, Printer from imptune.db.models import Client, Driver, Icon, Printer
from imptune.services.websearch import image_query
printer = ( printer = (
Printer.select(Printer, Client, Driver) Printer.select(Printer, Client, Driver)
@@ -201,6 +221,7 @@ def printer_detail(request: Request, printer_id: int):
"driver_names": driver_names, "driver_names": driver_names,
"has_driver": has_driver, "has_driver": has_driver,
"has_icon": icon is not None, "has_icon": icon is not None,
"image_query": image_query(printer.name),
"install_cmd": install_cmd, "install_cmd": install_cmd,
"uninstall_cmd": uninstall_cmd, "uninstall_cmd": uninstall_cmd,
}, },
@@ -251,7 +272,7 @@ def client_detail(request: Request, client_id: int):
"client": client, "client": client,
"grouped": grouped, "grouped": grouped,
"clients": clients, "clients": clients,
"driver_data": _driver_data(all_drivers), "driver_data": build_driver_data(all_drivers),
"printer_count": len(printers), "printer_count": len(printers),
"ready_count": sum(1 for p in printers if p.driver_id), "ready_count": sum(1 for p in printers if p.driver_id),
}, },
+4 -15
View File
@@ -1,22 +1,16 @@
"""Printer CRUD API — POST /printers, DELETE /printers/{id}, PATCH /printers/{id}.""" """Printer CRUD API — POST /printers, DELETE /printers/{id}, PATCH /printers/{id}."""
from __future__ import annotations from __future__ import annotations
from pathlib import Path
from fastapi import APIRouter, Form, Request from fastapi import APIRouter, Form, Request
from fastapi.responses import HTMLResponse, RedirectResponse, Response from fastapi.responses import HTMLResponse, RedirectResponse, Response
from fastapi.templating import Jinja2Templates
from peewee import JOIN 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.db.models import Client, Driver, Printer
from imptune.templating import templates
router = APIRouter(prefix="/printers") router = APIRouter(prefix="/printers")
templates = Jinja2Templates(
directory=str(Path(__file__).parent.parent / "templates")
)
_VALID_DUPLEX = {"OneSided", "LongEdge", "ShortEdge"} _VALID_DUPLEX = {"OneSided", "LongEdge", "ShortEdge"}
_VALID_PAPER = {"A4", "Letter", "Legal"} _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: def _render_printer_list(request: Request) -> HTMLResponse:
"""Query printers with LEFT JOIN on client and render grouped partial.""" """Query printers with LEFT JOIN on client and render grouped partial."""
import json
owner = request.state.owner owner = request.state.owner
query = ( query = (
Printer.select(Printer, Client) Printer.select(Printer, Client)
@@ -96,11 +88,8 @@ def _render_printer_list(request: Request) -> HTMLResponse:
grouped = group_printers_by_client(query) grouped = group_printers_by_client(query)
clients = list(Client.select().where(Client.owner == owner).order_by(Client.name)) clients = list(Client.select().where(Client.owner == owner).order_by(Client.name))
all_drivers = list(Driver.select().order_by(Driver.uploaded_at.desc())) all_drivers = Driver.select().order_by(Driver.uploaded_at.desc())
driver_data = [ driver_data = build_driver_data(all_drivers)
{"driver": d, "names": json.loads(d.driver_desc) if d.driver_desc else []}
for d in all_drivers
]
return templates.TemplateResponse( return templates.TemplateResponse(
request=request, request=request,
+1 -7
View File
@@ -1,22 +1,16 @@
"""Owner session routes — backup-key download and restore-on-new-browser.""" """Owner session routes — backup-key download and restore-on-new-browser."""
from __future__ import annotations from __future__ import annotations
from pathlib import Path
from fastapi import APIRouter, Form, HTTPException, Request from fastapi import APIRouter, Form, HTTPException, Request
from fastapi.responses import HTMLResponse, PlainTextResponse, RedirectResponse from fastapi.responses import HTMLResponse, PlainTextResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
import imptune.config as cfg import imptune.config as cfg
from imptune.db.models import Owner from imptune.db.models import Owner
from imptune.services.session import COOKIE_NAME, cookie_kwargs, is_same_origin from imptune.services.session import COOKIE_NAME, cookie_kwargs, is_same_origin
from imptune.templating import templates
router = APIRouter(prefix="/session") router = APIRouter(prefix="/session")
templates = Jinja2Templates(
directory=str(Path(__file__).parent.parent / "templates")
)
def _require_cookie_sessions() -> None: def _require_cookie_sessions() -> None:
"""404 these routes in single-user mode — keys have no meaning without a cookie. """404 these routes in single-user mode — keys have no meaning without a cookie.
+113
View File
@@ -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"<p class='dim' x-data x-text=\"$store.i18n.t('{key}')\">{fallback}</p>",
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),
},
)
+5
View File
@@ -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")) 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") DB_PATH = str(Path(DATA_DIR) / "imptune.db")
DRIVERS_DIR = str(Path(DATA_DIR) / "drivers") DRIVERS_DIR = str(Path(DATA_DIR) / "drivers")
ICONS_DIR = str(Path(DATA_DIR) / "icons") ICONS_DIR = str(Path(DATA_DIR) / "icons")
+18 -2
View File
@@ -16,7 +16,7 @@ def init_db() -> None:
Closes any existing connection before re-initializing so that test Closes any existing connection before re-initializing so that test
fixtures can monkeypatch DB_PATH between test runs. 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 # Re-read DB_PATH each time so tests can patch imptune.config.DB_PATH
import imptune.config as cfg import imptune.config as cfg
@@ -33,8 +33,24 @@ def init_db() -> None:
}, },
) )
db.connect(reuse_if_open=True) 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) _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: def _migrate_owner_column(data_dir: str) -> None:
+26
View File
@@ -58,10 +58,18 @@ class Driver(BaseModel):
inf_filename = CharField(null=True) inf_filename = CharField(null=True)
architecture = CharField(null=True) architecture = CharField(null=True)
has_cat_file = BooleanField(default=False) 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: class Meta:
table_name = "driver" 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): class Printer(BaseModel):
"""Printer configuration record.""" """Printer configuration record."""
@@ -94,3 +102,21 @@ class Icon(BaseModel):
class Meta: class Meta:
table_name = "icon" 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"
+13 -1
View File
@@ -5,7 +5,18 @@ from pathlib import Path
from fastapi import FastAPI from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles from fastapi.staticfiles import StaticFiles
from imptune.api import clients, drivers, health, 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.config import DATA_DIR, DRIVERS_DIR, ICONS_DIR
from imptune.db.database import db, init_db from imptune.db.database import db, init_db
from imptune.services.session import OwnerSessionMiddleware from imptune.services.session import OwnerSessionMiddleware
@@ -40,3 +51,4 @@ app.include_router(scripts.router)
app.include_router(packages.router) app.include_router(packages.router)
app.include_router(icons.router) app.include_router(icons.router)
app.include_router(session.router) app.include_router(session.router)
app.include_router(web.router)
+93
View File
@@ -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("-", " ")
+50
View File
@@ -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()
+384
View File
@@ -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'<a[^>]+class="result__a"[^>]+href="(?P<href>[^"]+)"[^>]*>(?P<title>.*?)</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"
+102 -1
View File
@@ -1263,6 +1263,106 @@ div.error > p {
.icon-preview .meta { font-size: .812rem; color: var(--im-text-dim); } .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 Dialogs
========================================================================== */ ========================================================================== */
@@ -1406,7 +1506,8 @@ details.more-pills > summary::marker { content: ""; }
sidebar there is only room for the file and its parsed names. */ sidebar there is only room for the file and its parsed names. */
.form-aside .data-table .col-arch, .form-aside .data-table .col-arch,
.form-aside .data-table .col-used, .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 label { margin: 0; }
.uploader .hint { margin-top: 0; } .uploader .hint { margin-top: 0; }
+58
View File
@@ -224,6 +224,35 @@
replace_icon: 'Remplacer licône', replace_icon: 'Remplacer licône',
no_icon: 'Aucune icône. Intune affichera licône par défaut.', no_icon: 'Aucune icône. Intune affichera licône par défaut.',
hint_icon: 'PNG, exactement 256 × 256, 750 Ko maximum.', hint_icon: 'PNG, exactement 256 × 256, 750 Ko maximum.',
hint_any_image: 'Nimporte quelle image — redimensionnée en PNG 256 × 256. 750 Ko maximum.',
// Recherche dimages 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 dimprimante.',
// 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 naffiche que des liens : téléchargez le ZIP puis téléversez-le ci-dessus.',
driver_search_label: 'Marque ou modèle dimprimante',
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', back_to_printers_btn: 'Retour aux imprimantes',
export_locked: 'Assignez un pilote à cette imprimante pour débloquer les scripts et lexport.', export_locked: 'Assignez un pilote à cette imprimante pour débloquer les scripts et lexport.',
// Edit modal // Edit modal
@@ -410,6 +439,35 @@
replace_icon: 'Replace icon', replace_icon: 'Replace icon',
no_icon: 'No icon. Intune will show its default.', no_icon: 'No icon. Intune will show its default.',
hint_icon: 'PNG, exactly 256 × 256, 750 KB max.', 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 manufacturers 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', back_to_printers_btn: 'Back to printers',
export_locked: 'Assign a driver to this printer to unlock scripts and export.', export_locked: 'Assign a driver to this printer to unlock scripts and export.',
// Edit modal // Edit modal
+38
View File
@@ -38,6 +38,44 @@
</div> </div>
</section> </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> <h2 class="eyebrow" style="margin-bottom:.6rem" x-data x-text="$store.i18n.t('driver_library')">Driver library</h2>
{% include "partials/driver_list.html" %} {% include "partials/driver_list.html" %}
{% endblock %} {% endblock %}
@@ -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()">&times;</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&times;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>
@@ -30,17 +30,34 @@
<th class="col-arch" x-data x-text="$store.i18n.t('architecture')">Architecture</th> <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-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-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> </tr>
</thead> </thead>
<tbody> <tbody>
{% for item in driver_data %} {% for item in driver_data %}
<tr> <tr>
<td> <td>
<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> <span class="cell-name mono">{{ item.driver.original_filename }}</span>
{% endif %}
<span class="cell-sub"> <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.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 %} &middot; {{ item.driver.inf_filename }}{% endif %} {% if item.driver.inf_filename %} &middot; {{ item.driver.inf_filename }}{% endif %}
</span> </span>
</span>
</div>
</td> </td>
<td> <td>
{% if item.names %} {% if item.names %}
@@ -77,6 +94,9 @@
<span class="badge ok">{{ ico.i('check', 12) }}new</span> <span class="badge ok">{{ ico.i('check', 12) }}new</span>
{% endif %} {% endif %}
</td> </td>
<td class="actions col-actions">
{% include "partials/driver_edit_modal.html" %}
</td>
</tr> </tr>
{% endfor %} {% endfor %}
</tbody> </tbody>
@@ -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 %}
@@ -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 }}&times;{{ 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 %}
+26 -2
View File
@@ -192,16 +192,40 @@
hx-target="#icon-status" hx-swap="innerHTML" hx-target="#icon-status" hx-swap="innerHTML"
hx-encoding="multipart/form-data"> hx-encoding="multipart/form-data">
<div class="uploader"> <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')"> x-data :aria-label="$store.i18n.t('upload_icon')">
<div class="btn-row"> <div class="btn-row">
<button type="submit" class="btn ghost sm" x-data> <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> {{ ico.i('upload', 13) }}<span x-text="$store.i18n.t('{{ 'replace_icon' if has_icon else 'upload_icon' }}')">Upload icon</span>
</button> </button>
</div> </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> </div>
</form> </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> </div>
</section> </section>
</div> </div>
+17
View File
@@ -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
+84
View File
@@ -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
+46 -1
View File
@@ -26,12 +26,18 @@ def db_env(tmp_path, monkeypatch):
if not db.is_closed(): if not db.is_closed():
db.close() db.close()
return { yield {
"data_dir": data_dir, "data_dir": data_dir,
"drivers_dir": drivers_dir, "drivers_dir": drivers_dir,
"db_path": db_path, "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): def test_create_tables(db_env):
"""init_db() creates all 4 tables in a fresh SQLite file.""" """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) path = store.get_path(sha256)
assert path == Path(str(drivers_dir)) / f"{sha256}.zip" 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
+227
View File
@@ -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
)
+82
View File
@@ -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
+36 -9
View File
@@ -66,16 +66,35 @@ class TestIconUpload:
icon_file = Path(tmp_data_dir) / "icons" / sha256 icon_file = Path(tmp_data_dir) / "icons" / sha256
assert icon_file.exists() assert icon_file.exists()
def test_reject_non_png(self, client, owner, tmp_data_dir): def test_jpeg_is_converted(self, client, owner, tmp_data_dir):
"""POST with a JPEG file returns 422 with PNG format error.""" """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) printer = _create_printer(owner)
jpeg_data = _make_jpeg(256, 256) jpeg_data = _make_jpeg(256, 256)
response = client.post( response = client.post(
f"/printers/{printer.id}/icon", f"/printers/{printer.id}/icon",
files={"file": ("icon.jpg", io.BytesIO(jpeg_data), "image/jpeg")}, 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 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): def test_reject_oversized(self, client, owner, tmp_data_dir):
"""POST with PNG > 750KB returns 422 with 750 KB error.""" """POST with PNG > 750KB returns 422 with 750 KB error."""
@@ -90,16 +109,24 @@ class TestIconUpload:
assert response.status_code == 422 assert response.status_code == 422
assert "750" in response.text assert "750" in response.text
def test_reject_wrong_dimensions(self, client, owner, tmp_data_dir): def test_wrong_dimensions_are_resized(self, client, owner, tmp_data_dir):
"""POST with 128x128 PNG returns 422 with 256x256 error.""" """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) printer = _create_printer(owner)
png_data = _make_png(128, 128) png_data = _make_png(128, 400)
response = client.post( response = client.post(
f"/printers/{printer.id}/icon", 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 response.status_code == 200
assert "256x256" in response.text
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): def test_replace_existing_icon(self, client, owner, tmp_data_dir):
"""Second upload for same printer replaces the Icon record.""" """Second upload for same printer replaces the Icon record."""
+66
View File
@@ -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"")
+267
View File
@@ -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
+260
View File
@@ -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&amp;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()