fix(vp): auto-derive access code from target printer in non-proxy modes

Non-proxy VPs (Archive / Review / Queue) with a target printer set up
  a live-mirror bridge that forwards the slicer's MQTT and RTSPS auth
  bytes through to the real printer. The slicer holds one code in its
  profile (the one it bound the VP with), and that code has to satisfy
  both the VP listener and the real printer at the far end of the
  bridge. If the codes diverge the bridge silently fails at the second
  hop — slicer reaches .49:8883, FINs before sending a ClientHello,
  retries identically. The wiki framed the code-match requirement as a
  camera-only concern; it isn't, all bridged protocols inherit.

  Fix removes the foot-gun instead of re-documenting it. When a target
  is selected on a non-proxy VP the access-code field switches to a
  read-only display showing the target's code with an Eye-toggle
  reveal; the backend auto-inherits on every create / update (any
  explicit access_code submitted alongside a target is silently
  overridden as belt-and-braces for non-UI clients). The required-when-
  enabling check now treats target-set as satisfying the access-code
  requirement. Standalone (no-target) non-proxy VPs still get the
  editable input + Save button.

  One-shot startup migration corrects any pre-existing mismatched
  rows: SELECTs diverged VPs and logs one INFO line per row for the
  audit trail, then UPDATEs via correlated subquery. Idempotent and
  portable between SQLite and Postgres.
This commit is contained in:
maziggy
2026-06-05 17:19:16 +02:00
parent 12d17bfbe7
commit 19073f5c84
21 changed files with 788 additions and 146 deletions
@@ -452,3 +452,150 @@ class TestVirtualPrinterDiagnosticAPI:
by_id = {c["id"]: c["status"] for c in result["checks"]}
assert by_id["enabled"] == "fail"
assert by_id["running"] == "skip"
class TestVirtualPrinterAccessCodeInheritance:
"""Non-proxy VPs with a target printer must inherit the target's access
code at write time.
The live-mirror bridge forwards the slicer's MQTT/RTSPS auth bytes to
the real printer — if the codes diverge the slicer binds the VP but the
bridge fails at the second hop. The route layer force-derives the code
on every create / update so a non-UI client can't introduce a divergence
either.
"""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_with_target_ignores_submitted_access_code(
self, async_client: AsyncClient, printer_factory, db_session
):
from sqlalchemy import select
from backend.app.models.virtual_printer import VirtualPrinter
target = await printer_factory(name="Real X1C", access_code="REALCODE")
response = await async_client.post(
"/api/v1/virtual-printers",
json={
"name": "QueueVP",
"mode": "queue",
"access_code": "WRONGAAA",
"target_printer_id": target.id,
},
)
assert response.status_code == 200
vp_id = response.json()["id"]
assert response.json()["access_code_set"] is True
vp = (await db_session.execute(select(VirtualPrinter).where(VirtualPrinter.id == vp_id))).scalar_one()
assert vp.access_code == "REALCODE"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_with_target_and_no_access_code_still_enables(
self, async_client: AsyncClient, printer_factory
):
"""A non-proxy VP with a target set can be enabled without supplying
access_code separately — the inheritance makes the explicit field
redundant, and the validator now knows this."""
target = await printer_factory(name="Real X1C", access_code="REALCODE")
response = await async_client.post(
"/api/v1/virtual-printers",
json={
"name": "QueueVP",
"mode": "queue",
"target_printer_id": target.id,
"bind_ip": "192.168.1.50",
"enabled": True,
},
)
assert response.status_code == 200
assert response.json()["access_code_set"] is True
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_without_target_still_requires_access_code_on_enable(self, async_client: AsyncClient):
"""The relaxation only kicks in when a target is set. A standalone
non-proxy VP still needs its own access code."""
response = await async_client.post(
"/api/v1/virtual-printers",
json={
"name": "StandaloneVP",
"mode": "archive",
"bind_ip": "192.168.1.51",
"enabled": True,
},
)
assert response.status_code == 400
assert "access code" in response.json()["detail"].lower()
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_target_resyncs_access_code(self, async_client: AsyncClient, printer_factory, db_session):
from sqlalchemy import select
from backend.app.models.virtual_printer import VirtualPrinter
first = await printer_factory(name="Printer A", access_code="AAAAAAAA")
second = await printer_factory(name="Printer B", access_code="BBBBBBBB")
create_resp = await async_client.post(
"/api/v1/virtual-printers",
json={
"name": "MovingTarget",
"mode": "queue",
"target_printer_id": first.id,
},
)
assert create_resp.status_code == 200
vp_id = create_resp.json()["id"]
vp = (await db_session.execute(select(VirtualPrinter).where(VirtualPrinter.id == vp_id))).scalar_one()
assert vp.access_code == "AAAAAAAA"
# Repoint to the second printer — access code should follow.
update_resp = await async_client.put(
f"/api/v1/virtual-printers/{vp_id}",
json={"target_printer_id": second.id},
)
assert update_resp.status_code == 200
await db_session.refresh(vp)
assert vp.access_code == "BBBBBBBB"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_explicit_access_code_with_target_is_overridden(
self, async_client: AsyncClient, printer_factory, db_session
):
"""An update that submits both an explicit access_code AND keeps a
target_printer_id silently uses the target's code — belt-and-braces
for non-UI clients that might try to set a divergent value."""
from sqlalchemy import select
from backend.app.models.virtual_printer import VirtualPrinter
target = await printer_factory(name="Real X1C", access_code="REALCODE")
create_resp = await async_client.post(
"/api/v1/virtual-printers",
json={
"name": "BeltBraces",
"mode": "queue",
"target_printer_id": target.id,
},
)
assert create_resp.status_code == 200
vp_id = create_resp.json()["id"]
update_resp = await async_client.put(
f"/api/v1/virtual-printers/{vp_id}",
json={"access_code": "FORGEDCD"},
)
assert update_resp.status_code == 200
vp = (await db_session.execute(select(VirtualPrinter).where(VirtualPrinter.id == vp_id))).scalar_one()
assert vp.access_code == "REALCODE"