fix(api): adressiere Copilot+Gemini Review-Findings auf PR #92

1. batch.py: check_printer_access(auth, printer.id) nach Resolve.
   Pattern aus printers.py (Z.243, 392, 418). Schliesst ACL-Bypass:
   api-key mit allowed_printer_ids=[A] kann jetzt nicht POST /api/print/B/batch.
   (Copilot + Gemini: security-critical)

2. batch.py: validiere printer.id == app.state.printer_id.
   Hub ist single-printer at startup (main.py:329 wired PrintService
   auf einen UUID). Ohne Check wuerde unsere Route silently die falsche
   Hardware ansprechen. Bei Mismatch: 404 printer_not_active.
   Future-proof fuer Multi-Printer durch klare Error-Message.
   (Copilot: correctness-critical)

3. test_job_state_fragment.py: Jinja2 autoescape=True.
   CodeQL py/jinja2-autoescape. Kein funktionaler Unterschied fuer
   UUID/String-Substitutionen. (CodeQL)

4. test_batch_endpoint_auth.py: nutzlose Variablen entfernt.
   RUF059 unused unpacks weil test sowieso skipped.

Lokal: 831 passed, 6 skipped, ruff/format/mypy alle grün.

Refs strausmann/hangar#78

Author: strausmann

backend/app/api/routes/batch.py

@@ -8,7 +8,7 @@
 from fastapi import APIRouter, Depends, HTTPException, Path, Request, status
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.auth.dependencies import AuthContext
+from app.auth.dependencies import AuthContext, check_printer_access
 from app.auth.scope_deps import require_print
 from app.db.session import get_session
 from app.models.print_batch import PrintBatch
@@ -55,12 +55,34 @@ async def create_batch(
     session: SessionDep,
     auth: Annotated[AuthContext, Depends(require_print)],
 ) -> BatchResponse:
-    # 1. Resolve printer
+    # 1. Resolve printer (404 if unknown slug/uuid)
     printer = await printers_repo.resolve_by_slug_or_uuid(session, printer_key)
     if printer is None:
         raise HTTPException(404, detail={"error_code": "printer_not_found"})
 
-    # 2. Best-effort dispatch
+    # 2. ACL: api-key may be restricted to a subset of printer_ids
+    check_printer_access(auth, printer.id)
+
+    # 3. Verify the resolved printer matches the singleton wired into
+    # app.state.print_service. The Hub is currently single-printer at
+    # startup (main.py wires PrintService to app.state.printer_id).
+    # If the URL slug points to a different printer row, the dispatch
+    # would silently route to the wrong device. Reject explicitly.
+    seeded_printer_id = getattr(http.app.state, "printer_id", None)
+    if seeded_printer_id is not None and printer.id != seeded_printer_id:
+        raise HTTPException(
+            404,
+            detail={
+                "error_code": "printer_not_active",
+                "error_message": (
+                    "Resolved printer is not the currently-seeded device. "
+                    "Hub is single-printer at startup; multi-printer routing "
+                    "is a future enhancement."
+                ),
+            },
+        )
+
+    # 4. Best-effort dispatch
     service = http.app.state.print_service
     try:
         job_ids, errors = await dispatch_batch(service, body.items)
@@ -73,7 +95,7 @@ async def create_batch(
             },
         ) from exc
 
-    # 3. Persist tracking row
+    # 5. Persist tracking row
     # auth.subject_id does NOT exist — use api_key_id or source
     created_by = str(auth.api_key_id) if auth.api_key_id else auth.source
     batch_row = PrintBatch(

backend/tests/integration/test_batch_endpoint_auth.py

@@ -73,10 +73,6 @@ async def auth_db_session():
 @pytest.mark.asyncio
 async def test_batch_requires_auth(auth_client, auth_db_session):
     """Genuine 401 requires unauthenticated client; covered by Phase 7c auth tests."""
-    client, inner_app = auth_client
-    p = Printer(name="X", slug="x", model="X", backend="mock")
-    await printers_repo.create(auth_db_session, p)
-
     pytest.skip("401 requires unauthenticated client; covered by Phase 7c auth tests")
 
 

backend/tests/integration/test_batch_endpoint_happy.py

@@ -87,9 +87,7 @@ async def test_batch_happy_path(batch_client, batch_db_session, batch_auth_heade
             for i in range(3)
         ]
     }
-    resp = await client.post(
-        f"/api/print/{p.slug}/batch", json=body, headers=batch_auth_headers
-    )
+    resp = await client.post(f"/api/print/{p.slug}/batch", json=body, headers=batch_auth_headers)
     assert resp.status_code == 202, resp.text
     data = resp.json()
     assert "batch_id" in data

backend/tests/integration/test_batch_endpoint_partial_failure.py

@@ -60,9 +60,7 @@ def partial_auth_headers() -> dict:
 
 
 @pytest.mark.asyncio
-async def test_batch_partial_failure(
-    partial_client, partial_db_session, partial_auth_headers
-):
+async def test_batch_partial_failure(partial_client, partial_db_session, partial_auth_headers):
     client, inner_app = partial_client
     p = Printer(name="Brother PT-P750W", slug="brother-p750w", model="PT-P750W", backend="mock")
     await printers_repo.create(partial_db_session, p)
@@ -86,9 +84,7 @@ async def test_batch_partial_failure(
             },
         ]
     }
-    resp = await client.post(
-        f"/api/print/{p.slug}/batch", json=body, headers=partial_auth_headers
-    )
+    resp = await client.post(f"/api/print/{p.slug}/batch", json=body, headers=partial_auth_headers)
     assert resp.status_code == 202, resp.text
     data = resp.json()
     assert len(data["job_ids"]) == 2

backend/tests/integration/test_batch_endpoint_printer_offline.py

@@ -87,8 +87,6 @@ async def _raise(self, req):
             }
         ]
     }
-    resp = await client.post(
-        f"/api/print/{p.slug}/batch", json=body, headers=offline_auth_headers
-    )
+    resp = await client.post(f"/api/print/{p.slug}/batch", json=body, headers=offline_auth_headers)
     assert resp.status_code == 409, resp.text
     assert resp.json()["detail"]["error_code"] == "printer_offline"

backend/tests/integration/test_batch_endpoint_tape_mismatch.py

@@ -101,9 +101,7 @@ async def _maybe_raise(self, req):
             },
         ]
     }
-    resp = await client.post(
-        f"/api/print/{p.slug}/batch", json=body, headers=tape_auth_headers
-    )
+    resp = await client.post(f"/api/print/{p.slug}/batch", json=body, headers=tape_auth_headers)
     assert resp.status_code == 202, resp.text
     data = resp.json()
     assert len(data["job_ids"]) == 2

backend/tests/unit/test_job_state_fragment.py

@@ -13,7 +13,12 @@
 def jinja_env():
     # Hub-Templates liegen unter backend/app/templates/
     template_dir = Path(__file__).parent.parent.parent / "app" / "templates"
-    return Environment(loader=FileSystemLoader(str(template_dir)))
+    # autoescape=True via select_autoescape default — no functional change
+    # for our UUID/string substitutions, satisfies CodeQL py/jinja2-autoescape.
+    return Environment(
+        loader=FileSystemLoader(str(template_dir)),
+        autoescape=True,
+    )
 
 
 def test_job_state_fragment_has_data_job_id_and_state(jinja_env):