Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions backend/app/api/routes/jobs.py
Original file line number Diff line number Diff line change
Expand Up @@ -207,6 +207,7 @@ async def cancel_job(
async def pause_job(
job_id: UUID,
session: SessionDep,
_auth: PrintAuthDep,
) -> ProblemDetail:
"""Return 501 — pause is not yet implemented."""
# Verify the job exists so we return 404 rather than 501 for unknown jobs
Expand Down Expand Up @@ -241,6 +242,7 @@ async def pause_job(
async def resume_job(
job_id: UUID,
session: SessionDep,
_auth: PrintAuthDep,
) -> ProblemDetail:
"""Return 501 — resume is not yet implemented."""
# Verify the job exists so we return 404 rather than 501 for unknown jobs
Expand Down Expand Up @@ -276,6 +278,7 @@ async def resume_job(
async def retry_job(
job_id: UUID,
session: SessionDep,
_auth: PrintAuthDep,
) -> JobRead:
"""Clone a terminal job into a fresh QUEUED job."""
job = await _get_job_or_404(session, job_id)
Expand Down
3 changes: 3 additions & 0 deletions backend/app/api/routes/printers.py
Original file line number Diff line number Diff line change
Expand Up @@ -143,9 +143,12 @@ async def list_printers(
async def get_printer(
printer_id: UUID,
session: SessionDep,
_auth: ReadAuthDep,
) -> PrinterRead:
"""Return full printer metadata for a single printer."""
printer = await _get_printer_or_404(session, printer_id)
if _auth is not None:
check_printer_access(_auth, printer_id)
state = await printer_state_repo.get(session, printer_id)
return PrinterRead(
id=printer.id,
Expand Down
122 changes: 122 additions & 0 deletions backend/tests/integration/api/test_auth_wiring.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,13 @@

Phase 1k.1a (Task 25): Removed TemplateLoader and /api/templates tests.
Templates are deleted in Phase 1k.1a.

2026-08-14 security audit: added coverage for GET /api/printers/{id} and
POST /api/jobs/{id}/retry, which shipped with NO auth dependency at all
while their sibling endpoints in the same router file all required one
(see tests/unit/api/test_route_auth_coverage_guardrail.py for the
structural guardrail that now catches this class of gap for every route,
not just these two).
"""

from __future__ import annotations
Expand Down Expand Up @@ -119,3 +126,118 @@ async def test_pangolin_sso_header_grants_read(api_client_with_seed):
headers={"X-Pangolin-User": "testuser@example.com"},
)
assert resp.status_code == 200, f"SSO should grant read: {resp.status_code}"


# ---------------------------------------------------------------------------
# 2026-08-14 security audit regressions:
# GET /api/printers/{id} and POST /api/jobs/{id}/retry had NO auth
# dependency at all — any caller that could reach the backend got full
# printer connection metadata / could trigger a real reprint with zero
# credentials. See app/api/routes/printers.py::get_printer and
# app/api/routes/jobs.py::retry_job.
# ---------------------------------------------------------------------------


async def _seed_printer(factory):
from app.models.printer import Printer

async with factory() as s:
printer = Printer(
name="audit-seed-printer",
slug="audit-seed-printer",
model="PT-P750W",
backend="ptouch",
connection={"host": "192.0.2.10", "port": 9100},
)
s.add(printer)
await s.commit()
await s.refresh(printer)
return printer


async def _seed_failed_job(factory, printer_id):
from app.models.job import Job, JobState

async with factory() as s:
job = Job(
printer_id=printer_id,
template_key="audit-seed-template",
state=JobState.FAILED.value,
payload={"text": "audit"},
)
s.add(job)
await s.commit()
await s.refresh(job)
return job


@pytest.mark.asyncio
async def test_get_printer_detail_without_auth_returns_401(api_client_with_seed):
"""GET /api/printers/{id} must reject unauthenticated requests.

Regression test: this endpoint previously had no auth dependency at
all, unlike GET /api/printers (list) and every other single-printer
endpoint (status/tape/queue/pause/resume) in the same file.
"""
import app.db.engine as _engine_module

printer = await _seed_printer(_engine_module.async_session)

resp = await api_client_with_seed.get(f"/api/printers/{printer.id}")
assert resp.status_code == 401, f"Expected 401, got {resp.status_code}: {resp.text}"


@pytest.mark.asyncio
async def test_get_printer_detail_with_read_key_returns_200(api_client_with_seed):
"""GET /api/printers/{id} succeeds for a caller with a valid read-scope key."""
import app.db.engine as _engine_module

factory = _engine_module.async_session
printer = await _seed_printer(factory)
read_key = await _make_read_key(factory)

resp = await api_client_with_seed.get(
f"/api/printers/{printer.id}",
headers={"X-Label-Hub-Key": read_key},
)
assert resp.status_code == 200, f"Expected 200, got {resp.status_code}: {resp.text}"
assert resp.json()["id"] == str(printer.id)


@pytest.mark.asyncio
async def test_retry_job_without_auth_returns_401(api_client_with_seed):
"""POST /api/jobs/{id}/retry must reject unauthenticated requests.

Regression test: this endpoint previously had no auth dependency at
all, unlike list_jobs/get_job (read scope) and cancel_job (print scope)
in the same file — despite retry_job creating a new QUEUED job that the
worker actually dispatches to the physical printer.
"""
import app.db.engine as _engine_module

factory = _engine_module.async_session
printer = await _seed_printer(factory)
job = await _seed_failed_job(factory, printer.id)

resp = await api_client_with_seed.post(f"/api/jobs/{job.id}/retry")
assert resp.status_code == 401, f"Expected 401, got {resp.status_code}: {resp.text}"


@pytest.mark.asyncio
async def test_retry_job_with_print_key_returns_201(api_client_with_seed):
"""POST /api/jobs/{id}/retry succeeds for a caller with a valid print-scope key."""
import app.db.engine as _engine_module

factory = _engine_module.async_session
printer = await _seed_printer(factory)
job = await _seed_failed_job(factory, printer.id)
print_key = await _make_print_key(factory)

resp = await api_client_with_seed.post(
f"/api/jobs/{job.id}/retry",
headers={"X-Label-Hub-Key": print_key},
)
assert resp.status_code == 201, f"Expected 201, got {resp.status_code}: {resp.text}"
body = resp.json()
assert body["id"] != str(job.id)
assert body["state"] == "queued"
51 changes: 51 additions & 0 deletions backend/tests/integration/api/test_printer_acl.py
Original file line number Diff line number Diff line change
Expand Up @@ -117,3 +117,54 @@ async def test_key_restricted_to_printer_a_allowed_on_printer_a(api_client_with_
assert resp.status_code in (204, 404), (
f"Expected 204 or 404, got {resp.status_code}: {resp.text}"
)


@pytest.mark.asyncio
async def test_get_printer_detail_restricted_to_printer_a_blocked_on_printer_b(
api_client_with_seed,
):
"""2026-08-14 follow-up: GET /api/printers/{id} (single-printer detail)
now requires auth but was missing the per-printer ACL check that every
sibling endpoint (get_printer_status/pause_printer/resume_printer/
clear_printer_queue) enforces via check_printer_access(). A read-key
scoped to printer A must not be able to read printer B's metadata
(including its connection dict — host/port/SNMP config).

Analogous to test_key_restricted_to_printer_a_blocked_on_printer_b above,
but seeds its own printer B directly instead of relying on
api_client_with_seed's (frequently empty, skip-on-empty) default seed
data — this test must actually run, not skip, to be a real regression
guard.
"""
import app.db.engine as _engine_module
from app.models.printer import Printer

factory = _engine_module.async_session

async with factory() as s:
printer_b = Printer(
name="acl-test-printer-b",
slug="acl-test-printer-b",
model="PT-P750W",
backend="ptouch",
connection={"host": "192.0.2.20", "port": 9100},
)
s.add(printer_b)
await s.commit()
await s.refresh(printer_b)

printer_a_id = str(uuid4()) # some other printer the key IS allowed for

plaintext = await _insert_restricted_key(
factory,
allowed_printer_ids=[printer_a_id],
scopes=["read"],
)

resp = await api_client_with_seed.get(
f"/api/printers/{printer_b.id}",
headers={"X-Label-Hub-Key": plaintext},
)
assert resp.status_code == 403, (
f"Expected 403 for restricted key on wrong printer, got {resp.status_code}: {resp.text}"
)
3 changes: 2 additions & 1 deletion backend/tests/unit/api/test_jobs_routes.py
Original file line number Diff line number Diff line change
Expand Up @@ -545,7 +545,8 @@ async def test_retry_job_direct_returns_new_queued_job(session) -> None:
printer = await _make_printer(session)
original = await _make_job(session, printer.id, state=JobState.FAILED.value)

result = await retry_job(job_id=original.id, session=session)
fake_auth = AuthContext(source="api-key", scope="admin", api_key_id=_uuid4(), ip="127.0.0.1")
result = await retry_job(job_id=original.id, session=session, _auth=fake_auth)

assert str(result.id) != str(original.id)
assert result.state == JobState.QUEUED.value
Expand Down
Loading
Loading