"""Web UI and JSON API.""" from __future__ import annotations import base64 import hmac import logging import os import threading from contextlib import asynccontextmanager from pathlib import Path import paramiko import segno from fastapi import FastAPI, HTTPException, Request from fastapi.responses import HTMLResponse, JSONResponse, Response from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from . import materials, newspool, tagcodec from .bridge import Bridge from .sshsource import CanvasRecordReader, HostKeyMismatch, KeyStore, SSHError, fingerprint from .spoolman import SpoolmanError, SpoolmanNotFound, summary from .storage import SettingsStore log = logging.getLogger("web") HERE = Path(__file__).parent VERSION = os.environ.get("APP_VERSION", "dev") def create_app(data_dir: Path | None = None, *, bridge: Bridge | None = None, start_loop: bool = True, env: dict | None = None) -> FastAPI: env = os.environ if env is None else env data_dir = Path(data_dir or env.get("DATA_DIR", "/data")) data_dir.mkdir(parents=True, exist_ok=True) settings = bridge.settings_store if bridge else SettingsStore(data_dir, env) keys = bridge.keys if bridge else KeyStore(data_dir) bridge = bridge or Bridge(data_dir, settings, keys) stop = threading.Event() tested_host_keys: dict[str, dict] = {} # printer id -> host key from its last connection test @asynccontextmanager async def lifespan(app): thread = None if start_loop: thread = threading.Thread(target=bridge.run, args=(stop,), name="bridge", daemon=True) thread.start() yield stop.set() bridge.wake.set() if thread: thread.join(timeout=15) app = FastAPI(title="CC2 Spoolman Bridge", lifespan=lifespan, docs_url=None, redoc_url=None, openapi_url=None) app.state.bridge = bridge templates = Jinja2Templates(directory=str(HERE / "templates")) app.mount("/static", StaticFiles(directory=str(HERE / "static")), name="static") user, password = env.get("BRIDGE_USER", ""), env.get("BRIDGE_PASSWORD", "") @app.middleware("http") async def guard(request: Request, call_next): if user and password and request.url.path != "/healthz": ok = False auth = request.headers.get("authorization", "") if auth.lower().startswith("basic "): try: u, _, p = base64.b64decode(auth[6:]).decode("utf-8").partition(":") ok = hmac.compare_digest(u, user) & hmac.compare_digest(p, password) except (ValueError, UnicodeDecodeError): ok = False if not ok: return Response(status_code=401, headers={"WWW-Authenticate": 'Basic realm="CC2 Bridge"'}) # Writes only from our own pages: a custom header cannot be sent cross-site without CORS. if request.method not in ("GET", "HEAD") and request.headers.get("x-bridge-request") != "1": return JSONResponse({"error": "Fehlender X-Bridge-Request-Header"}, status_code=403) response = await call_next(request) response.headers["X-Content-Type-Options"] = "nosniff" response.headers["Referrer-Policy"] = "no-referrer" response.headers["X-Frame-Options"] = "DENY" return response def page(request: Request, name: str, **ctx): return templates.TemplateResponse(request, name, {"version": VERSION, "active": name, **ctx}) # ---------- pages ---------- @app.get("/", response_class=HTMLResponse) def index(request: Request): return page(request, "status.html") @app.get("/s/{smid}", response_class=HTMLResponse) def spool_page(request: Request, smid: int): return page(request, "spool.html", spool_id=smid) @app.get("/log", response_class=HTMLResponse) def log_page(request: Request): return page(request, "log.html") @app.get("/manifest.webmanifest") def manifest(): return JSONResponse({ "name": "CC2 Bridge", "short_name": "CC2 Bridge", "lang": "de", "start_url": "/", "scope": "/", "display": "standalone", "background_color": "#EDEFF2", "theme_color": "#14181F", "icons": [{"src": "/static/icon-192.png", "sizes": "192x192", "type": "image/png"}, {"src": "/static/icon-512.png", "sizes": "512x512", "type": "image/png"}, {"src": "/static/icon.svg", "sizes": "any", "type": "image/svg+xml"}], }, media_type="application/manifest+json") @app.get("/tags", response_class=HTMLResponse) def tags(request: Request): return page(request, "tags.html") @app.get("/settings", response_class=HTMLResponse) def settings_page(request: Request): return page(request, "settings.html") @app.get("/healthz") def healthz(): return {"ok": True, "version": VERSION, "last_poll": bridge.snapshot.get("last_poll")} # ---------- status ---------- @app.get("/api/status") def status(): return {**bridge.snapshot, "dry_run": settings.get().dry_run, "log": bridge.log.recent(200)} def printer_or_404(pid: str): p = settings.get().printer(pid) if p is None: raise HTTPException(404, "Unbekannter Drucker") return p def close_readers(): for w in list(bridge.workers.values()): w.reader.close() @app.post("/api/printers/{pid}/cc2/enable") async def cc2_enable(pid: str, request: Request): printer_or_404(pid) body = await _json(request) if body.get("confirm") is not True: raise HTTPException(400, "Bestätigung fehlt") bridge.commands.put(("enable", pid, None)) bridge.wake.set() return {"ok": True, "dry_run": settings.get().dry_run} @app.post("/api/printers/{pid}/slots/{slot}/{action}") def slot_action(pid: str, slot: int, action: str): printer_or_404(pid) if slot not in range(4) or action not in ("reassign", "release"): raise HTTPException(404) bridge.commands.put((action, pid, slot)) bridge.wake.set() return {"ok": True} # ---------- settings ---------- @app.get("/api/settings") def get_settings(): return settings.get().to_dict() @app.post("/api/settings") async def post_settings(request: Request): body = await _json(request) before = {p.id for p in settings.get().printers} allowed = set(settings.get().to_dict()) try: s = settings.save({k: v for k, v in body.items() if k in allowed}) except ValueError as e: raise HTTPException(400, str(e)) from None for pid in before - {p.id for p in s.printers}: keys.forget_host_key(pid) bridge.log.add("info", "Einstellungen gespeichert" + (" (Dry-Run aktiv)" if s.dry_run else " (Schreiben AKTIV)")) bridge.wake.set() return s.to_dict() # ---------- SSH ---------- @app.get("/api/ssh") def ssh_info(): return {"key": keys.key_info(), "host_keys": {p.id: keys.host_key(p.id) for p in settings.get().printers}} @app.post("/api/ssh/key") async def ssh_key(request: Request): body = await _json(request) text = body.get("key") if not isinstance(text, str) or len(text) > 20000: raise HTTPException(400, "Schlüssel fehlt") try: keys.save_key(text) except SSHError as e: raise HTTPException(400, str(e)) from None bridge.log.add("info", "SSH-Schlüssel gespeichert") close_readers() return {"key": keys.key_info()} @app.delete("/api/ssh/key") def ssh_key_delete(): keys.delete_key() close_readers() bridge.log.add("info", "SSH-Schlüssel gelöscht") return {"key": None} @app.post("/api/printers/{pid}/ssh/test") def ssh_test(pid: str): p = printer_or_404(pid) reader = CanvasRecordReader(keys, p.id, p.host, p.ssh_port, p.ssh_user) try: result = reader.test() except HostKeyMismatch as e: return JSONResponse({"ok": False, "error": str(e), "host_key": {"fingerprint": fingerprint(e.key)}}, status_code=200) except SSHError as e: return {"ok": False, "error": str(e)} hk = result["host_key"] if not result["host_key_pinned"]: tested_host_keys.pop(pid, None) tested_host_keys[pid] = hk slots = tagcodec.parse_record(result["record"], settings.get().id_offset) return {"ok": True, "host_key": {"fingerprint": hk["fingerprint"], "type": hk["type"]}, "host_key_pinned": result["host_key_pinned"], "slots": slots} @app.post("/api/printers/{pid}/ssh/hostkey") async def ssh_hostkey(pid: str, request: Request): p = printer_or_404(pid) body = await _json(request) hk = tested_host_keys.get(pid) if not hk or hk["fingerprint"] != body.get("fingerprint"): raise HTTPException(400, "Bitte zuerst „Verbindung testen“ und dann diesen Fingerprint bestätigen") key = paramiko.PKey.from_type_string(hk["type"], base64.b64decode(hk["key"])) keys.pin_host_key(pid, p.host, key) tested_host_keys.pop(pid, None) bridge.log.add("info", f"{p.name}: Host-Key bestätigt ({hk['fingerprint']})") bridge.wake.set() return {"host_key": keys.host_key(pid)} @app.delete("/api/printers/{pid}/ssh/hostkey") def ssh_hostkey_delete(pid: str): p = printer_or_404(pid) keys.forget_host_key(pid) close_readers() bridge.log.add("info", f"{p.name}: gespeicherter Host-Key gelöscht") return {"host_key": None} # ---------- tag generator ---------- @app.get("/api/materials") def get_materials(): return { "materials": {k: {"temps": v["temps"], "code": f"{v['code']:08X}"} for k, v in materials.MATERIALS.items()}, "subtypes": [{"code": c, "name": n, "hidden": h, "material": materials.material_of_subtype(c)} for c, n, h in materials.SUBTYPES], "palette": tagcodec.PALETTE, } @app.get("/api/spoolman/spools") def spoolman_spools(archived: bool = False): s = settings.get() client = bridge.spoolman_factory(s) try: spools = client.spools(archived=archived) except SpoolmanError as e: raise HTTPException(502, str(e)) from None finally: client.close() out = [] for sp in spools: info = summary(sp) sug = materials.suggest(info["material"], info["name"]) info["suggest"] = { "material": sug["material"], "subtype": sug["subtype"], "temp_min": sug["temps"][0], "temp_max": sug["temps"][1], "color": tagcodec.nearest_palette_color(info["color"]), } info["spoolman_url"] = f"{s.spoolman_url}/spool/show/{info['id']}" out.append(info) out.sort(key=lambda x: x["id"]) return {"spools": out, "id_offset": s.id_offset, "max_id": tagcodec.MAX_WEIGHT - s.id_offset} @app.get("/api/spool/{smid}") def spool_info(smid: int): s = settings.get() client = bridge.spoolman_factory(s) try: sm = client.spool(smid) except SpoolmanNotFound: raise HTTPException(404, f"Spoolman-Spule {smid} gibt es nicht") from None except SpoolmanError as e: raise HTTPException(502, str(e)) from None finally: client.close() info = summary(sm) name = info["label"] return { "spool": info, "filament_name": (sm.get("filament") or {}).get("name") or "", "spoolman_url": f"{s.spoolman_url}/spool/show/{smid}", **bridge.spool_whereabouts(smid), "storage": s.location_storage, "log": [e for e in bridge.log.recent(300) if name in e["msg"]][:8], "dry_run": s.dry_run, } @app.get("/api/spoolman/catalog") def spoolman_catalog(): client = bridge.spoolman_factory(settings.get()) try: return newspool.catalog(client) except SpoolmanError as e: raise HTTPException(502, str(e)) from None finally: client.close() @app.post("/api/spoolman/create") async def spoolman_create(request: Request): body = await _json(request) s = settings.get() try: p = newspool.plan(body) except (newspool.NewSpoolError, TypeError, ValueError) as e: raise HTTPException(400, str(e)) from None if s.dry_run: bridge.log.add("dry", f"würde in Spoolman {p['count']} neue Spule(n) anlegen") raise HTTPException(409, "Dry-Run aktiv: in Spoolman wird nichts angelegt") client = bridge.spoolman_factory(s) try: spools, what = newspool.create(client, p, s.location_storage) except newspool.NewSpoolError as e: raise HTTPException(400, str(e)) from None except SpoolmanError as e: raise HTTPException(502, str(e)) from None finally: client.close() bridge.log.add("info", f"erledigt: in Spoolman angelegt: {what}") return {"spools": spools} @app.post("/api/tag") async def tag(request: Request): body = await _json(request) s = settings.get() try: data = tagcodec.TagData( spool_id=int(body["spool_id"]), color=str(body["color"]), material=str(body["material"]), subtype=int(body["subtype"]), temp_min=int(body["temp_min"]), temp_max=int(body["temp_max"]), diameter=float(body.get("diameter") or 1.75), finish=int(body.get("finish", 0xFF)), id_offset=s.id_offset, url=str(body.get("url") or tagcodec.DEFAULT_URL), ) commands = tagcodec.encode_commands(data) except (KeyError, TypeError, ValueError) as e: raise HTTPException(400, str(e) if isinstance(e, tagcodec.TagError) else f"Ungültige Eingabe ({e})") from None decoded = tagcodec.decode_pages(tagcodec.parse_commands(commands), s.id_offset) qr = segno.make(commands, error="l").svg_inline(scale=3, border=2, dark="#000", light="#fff") return {"commands": commands, "decoded": decoded, "qr_svg": qr} @app.post("/api/tag/decode") async def tag_decode(request: Request): body = await _json(request) try: pages = tagcodec.parse_commands(str(body.get("commands", ""))) except tagcodec.TagError as e: raise HTTPException(400, str(e)) from None return tagcodec.decode_pages(pages, settings.get().id_offset) return app async def _json(request: Request) -> dict: try: body = await request.json() except ValueError: raise HTTPException(400, "Ungültiges JSON") from None if not isinstance(body, dict): raise HTTPException(400, "Ungültiges JSON") return body def main() -> None: import uvicorn logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO"), format="%(asctime)s %(levelname)s %(name)s: %(message)s") logging.getLogger("paramiko").setLevel(logging.WARNING) logging.getLogger("httpx").setLevel(logging.WARNING) uvicorn.run(create_app(), host="0.0.0.0", port=int(os.environ.get("PORT", "5066")), log_level="warning", proxy_headers=False) if __name__ == "__main__": main()