Files
bambuddy/backend/tests/unit/test_vp_certificate_rotation.py
maziggy 597762685c fix(virtual-printer): #1558 Send pre-flight + slicer-surface audit bundle
#1558: cached-as-base push_status only forced gcode_state=IDLE while letting
  the real printer's live-progress fields (mc_percent, stg_cur, layer_num, ...)
  leak through. Bambu Studio's Send pre-flight read them as busy and refused.
  The cached branch now overrides the activity-field set the same way it
  already overrode storage indicators (#1228) and protocol fields.

  Same bundle ships a multi-round VP audit that found adjacent bugs in the
  same family:

  - #1558: cached branch zeroes mc_print_stage / mc_percent / mc_remaining_time / stg / stg_cur / layer_num / total_layer_num / print_error
  - MQTT auth: per-IP rate-limit (5/60s lockout), hmac.compare_digest, access_code redacted in DEBUG log
  - FTP cmd_STOR streams chunks to disk + 4 GiB cap (was buffering whole upload)
  - Sticky-keys allowlist extended with upgrade_state / xcam / hw_switch_state / nozzle_diameter / nozzle_type / online / ams_status
  - _pending_files cleanup in finally for archive / queue / dispatch handlers
  - _add_to_print_queue position uses MAX+1 (was hardcoded 1)
  - DELETE VP removes orphan PendingUpload rows + upload_dir from disk
  - Per-VP cert regenerates on shared-CA rotation (real signature verification, not DN match)
  - DHCP target-IP refresh + queue_force_color_match toggle now restart proxy VPs
  - Per-slicer bridge-response routing (multi-slicer cross-leak fix via sequence_id map)
  - Child-service readiness barrier (FTP / MQTT / Bind / SSDP) — no false is_running before sockets bind
  - H2D Pro O1E / O2D model codes added (experimental, needs field confirmation)
  - FTP passive port range widened 50000-51000; docker-compose + wiki updated
  - VP refresh_loop crash now unbinds raw_message_handler; tailscale catches asyncio.TimeoutError; SlicerProxyManager lifecycle hardening
2026-05-30 13:34:10 +02:00

77 lines
3.7 KiB
Python

"""Tests for CertificateService.ensure_certificates' CA-rotation guard.
When the shared CA is regenerated (e.g. its expiry crossed
``CA_EXPIRY_THRESHOLD_DAYS``), any per-VP printer certificate that was
signed by the OLD CA becomes orphaned: it still exists on disk and the
old fallback ``cert_path.exists()`` check would happily reuse it. A
slicer that imported the NEW CA then fails the TLS handshake because
the printer cert's issuer doesn't match anything in its trust store.
``_cert_matches_current_ca`` is the guard. It compares the on-disk
printer cert's issuer against the on-disk CA cert's subject; on
mismatch ``ensure_certificates`` regenerates the per-VP cert under the
current CA.
"""
from backend.app.services.virtual_printer.certificate import CertificateService
def test_ensure_certificates_reuses_cert_when_issuer_matches_ca(tmp_path):
"""Happy path: a freshly-generated CA + per-VP cert pair shares
issuer/subject. ``ensure_certificates`` reads them back without
regenerating."""
svc = CertificateService(cert_dir=tmp_path, serial="01P00A391800001")
# First call: generates the CA + per-VP cert from scratch.
first_cert, first_key = svc.ensure_certificates()
first_cert_bytes = first_cert.read_bytes()
# Second call: cert + CA exist and the issuer matches. Should reuse.
second_cert, _ = svc.ensure_certificates()
assert second_cert.read_bytes() == first_cert_bytes
def test_ensure_certificates_regenerates_when_ca_rotated(tmp_path):
"""CA rotation scenario: the CA file is replaced with a different one
(e.g. previous expired and was regenerated). The per-VP cert on disk
was signed by the old CA, so its issuer no longer matches the new CA's
subject. ``ensure_certificates`` must regenerate the per-VP cert."""
# Build the first pair.
svc1 = CertificateService(cert_dir=tmp_path, serial="01P00A391800001")
orig_cert_bytes = svc1.ensure_certificates()[0].read_bytes()
orig_ca_bytes = svc1.ca_cert_path.read_bytes()
# Simulate CA rotation: build a SECOND CA in a different dir, then
# swap that CA's files into the original CA path. The per-VP cert
# still on disk was signed by the original CA — issuer mismatch now.
rotated_dir = tmp_path / "rotated"
rotated_dir.mkdir()
svc_rotated = CertificateService(cert_dir=rotated_dir, serial="01P00A391800002")
svc_rotated.ensure_certificates()
# Overwrite the original CA on disk with the rotated one.
svc1.ca_cert_path.write_bytes(svc_rotated.ca_cert_path.read_bytes())
svc1.ca_key_path.write_bytes(svc_rotated.ca_key_path.read_bytes())
assert svc1.ca_cert_path.read_bytes() != orig_ca_bytes # confirm rotation
# Build a fresh service against the rotated CA, then ensure_certificates
# should detect the mismatch and regenerate the per-VP cert.
svc2 = CertificateService(cert_dir=tmp_path, serial="01P00A391800001")
new_cert, _ = svc2.ensure_certificates()
new_cert_bytes = new_cert.read_bytes()
# New per-VP cert must differ from the old (signed by a different CA now).
assert new_cert_bytes != orig_cert_bytes
def test_cert_matches_current_ca_returns_false_when_no_ca(tmp_path):
"""If the CA file is missing entirely, the match check must return
False so ``ensure_certificates`` falls through to ``generate_certificates``
instead of returning a per-VP cert that nothing can validate."""
svc = CertificateService(cert_dir=tmp_path, serial="01P00A391800001")
# Write a per-VP cert without a CA.
svc.cert_path.write_bytes(b"fake cert content")
svc.key_path.write_bytes(b"fake key content")
# No bbl_ca.crt on disk → match fails safely.
assert svc._cert_matches_current_ca() is False