test_spoolbuddy_schema_validation.py 8.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207
  1. """Unit tests for SpoolBuddy schema validation (security fixes H2, M2, M4).
  2. Tests Pydantic model validation without requiring a running server or DB.
  3. """
  4. import pytest
  5. from pydantic import ValidationError
  6. from backend.app.schemas.spoolbuddy import (
  7. DeviceRegisterRequest,
  8. HeartbeatRequest,
  9. ScaleReadingRequest,
  10. UpdateStatusRequest,
  11. WriteTagResultRequest,
  12. )
  13. # ---------------------------------------------------------------------------
  14. # H2 — UpdateStatusRequest: only valid Literal values accepted
  15. # ---------------------------------------------------------------------------
  16. class TestUpdateStatusRequestValidation:
  17. def test_valid_status_updating(self):
  18. req = UpdateStatusRequest(status="updating")
  19. assert req.status == "updating"
  20. def test_valid_status_complete(self):
  21. req = UpdateStatusRequest(status="complete")
  22. assert req.status == "complete"
  23. def test_valid_status_error(self):
  24. req = UpdateStatusRequest(status="error")
  25. assert req.status == "error"
  26. def test_invalid_status_rejected(self):
  27. """Arbitrary status strings must be rejected (H2: prevents unbounded WS injection)."""
  28. with pytest.raises(ValidationError):
  29. UpdateStatusRequest(status="hacked")
  30. def test_empty_status_rejected(self):
  31. with pytest.raises(ValidationError):
  32. UpdateStatusRequest(status="")
  33. def test_message_max_length_enforced(self):
  34. """message field must not exceed 255 chars."""
  35. with pytest.raises(ValidationError):
  36. UpdateStatusRequest(status="updating", message="x" * 256)
  37. def test_message_at_max_length_accepted(self):
  38. req = UpdateStatusRequest(status="complete", message="x" * 255)
  39. assert len(req.message) == 255
  40. # ---------------------------------------------------------------------------
  41. # M2 — HeartbeatRequest: system_stats size limit (4096 bytes)
  42. # ---------------------------------------------------------------------------
  43. class TestHeartbeatSystemStatsValidation:
  44. def test_none_accepted(self):
  45. req = HeartbeatRequest(system_stats=None)
  46. assert req.system_stats is None
  47. def test_small_dict_accepted(self):
  48. req = HeartbeatRequest(system_stats={"cpu": 12.5, "mem": 60.0})
  49. assert req.system_stats["cpu"] == 12.5
  50. def test_oversized_dict_rejected(self):
  51. """system_stats exceeding 4096 bytes JSON-encoded must be rejected (M2)."""
  52. huge = {"data": "x" * 5000}
  53. with pytest.raises(ValidationError, match="4096"):
  54. HeartbeatRequest(system_stats=huge)
  55. def test_exactly_4096_bytes_accepted(self):
  56. """A dict whose JSON is exactly 4096 bytes must pass."""
  57. import json
  58. # Build a dict whose JSON is exactly 4096 bytes
  59. filler = "x" * (4096 - len('{"k": ""}'))
  60. d = {"k": filler}
  61. assert len(json.dumps(d)) == 4096
  62. req = HeartbeatRequest(system_stats=d)
  63. assert req.system_stats is not None
  64. def test_one_byte_over_limit_rejected(self):
  65. import json
  66. filler = "x" * (4097 - len('{"k": ""}'))
  67. d = {"k": filler}
  68. assert len(json.dumps(d)) == 4097
  69. with pytest.raises(ValidationError):
  70. HeartbeatRequest(system_stats=d)
  71. # ---------------------------------------------------------------------------
  72. # M4 — DeviceRegisterRequest: max_length on device-sourced string fields
  73. # ---------------------------------------------------------------------------
  74. class TestDeviceRegisterRequestValidation:
  75. VALID_BASE = {"device_id": "dev1", "hostname": "spoolbuddy.local", "ip_address": "192.168.1.50"}
  76. def test_valid_minimal_accepted(self):
  77. req = DeviceRegisterRequest(**self.VALID_BASE)
  78. assert req.device_id == "dev1"
  79. def test_firmware_version_too_long_rejected(self):
  80. with pytest.raises(ValidationError):
  81. DeviceRegisterRequest(**self.VALID_BASE, firmware_version="x" * 21)
  82. def test_firmware_version_at_max_accepted(self):
  83. req = DeviceRegisterRequest(**self.VALID_BASE, firmware_version="x" * 20)
  84. assert req.firmware_version == "x" * 20
  85. def test_nfc_reader_type_too_long_rejected(self):
  86. with pytest.raises(ValidationError):
  87. DeviceRegisterRequest(**self.VALID_BASE, nfc_reader_type="x" * 21)
  88. def test_nfc_connection_too_long_rejected(self):
  89. with pytest.raises(ValidationError):
  90. DeviceRegisterRequest(**self.VALID_BASE, nfc_connection="x" * 21)
  91. def test_backend_url_too_long_rejected(self):
  92. with pytest.raises(ValidationError):
  93. DeviceRegisterRequest(**self.VALID_BASE, backend_url="http://" + "x" * 249)
  94. def test_backend_url_at_max_accepted(self):
  95. url = "http://" + "x" * (255 - len("http://"))
  96. req = DeviceRegisterRequest(**self.VALID_BASE, backend_url=url)
  97. assert req.backend_url == url
  98. def test_device_id_too_long_rejected(self):
  99. with pytest.raises(ValidationError):
  100. DeviceRegisterRequest(device_id="x" * 51, hostname="h", ip_address="1.2.3.4")
  101. # ---------------------------------------------------------------------------
  102. # M4 — WriteTagResultRequest: device_id max_length
  103. # ---------------------------------------------------------------------------
  104. class TestWriteTagResultRequestValidation:
  105. def test_device_id_too_long_rejected(self):
  106. with pytest.raises(ValidationError):
  107. WriteTagResultRequest(device_id="x" * 51, spool_id=1, tag_uid="AABBCCDD", success=True)
  108. def test_device_id_at_max_accepted(self):
  109. req = WriteTagResultRequest(device_id="x" * 50, spool_id=1, tag_uid="AABBCCDD", success=True)
  110. assert len(req.device_id) == 50
  111. def test_tag_uid_hex_pattern_accepted(self):
  112. req = WriteTagResultRequest(device_id="dev1", spool_id=1, tag_uid="AABBCCDD", success=True)
  113. assert req.tag_uid == "AABBCCDD"
  114. def test_tag_uid_non_hex_rejected(self):
  115. """Non-hex characters in tag_uid must be rejected (prevents injection via NFC write-back)."""
  116. with pytest.raises(ValidationError):
  117. WriteTagResultRequest(device_id="dev1", spool_id=1, tag_uid="AABB; DROP", success=True)
  118. def test_tag_uid_too_short_rejected(self):
  119. with pytest.raises(ValidationError):
  120. WriteTagResultRequest(device_id="dev1", spool_id=1, tag_uid="AABB", success=True)
  121. def test_tag_uid_max_length_accepted(self):
  122. req = WriteTagResultRequest(device_id="dev1", spool_id=1, tag_uid="A" * 30, success=True)
  123. assert len(req.tag_uid) == 30
  124. def test_tag_uid_over_max_length_rejected(self):
  125. with pytest.raises(ValidationError):
  126. WriteTagResultRequest(device_id="dev1", spool_id=1, tag_uid="A" * 31, success=True)
  127. # ---------------------------------------------------------------------------
  128. # M4 — ScaleReadingRequest: weight_grams accepts any float (raw uncalibrated ADC)
  129. # ---------------------------------------------------------------------------
  130. class TestScaleReadingRequestValidation:
  131. def test_valid_weight_accepted(self):
  132. req = ScaleReadingRequest(device_id="sb1", weight_grams=250.0)
  133. assert req.weight_grams == 250.0
  134. def test_zero_weight_accepted(self):
  135. req = ScaleReadingRequest(device_id="sb1", weight_grams=0.0)
  136. assert req.weight_grams == 0.0
  137. def test_large_raw_adc_weight_accepted(self):
  138. # Uncalibrated scale with factor=1.0 produces raw ADC values in the millions
  139. req = ScaleReadingRequest(device_id="sb1", weight_grams=5_000_000.0)
  140. assert req.weight_grams == 5_000_000.0
  141. def test_negative_weight_accepted(self):
  142. # Scale can legitimately read negative values when tare is not calibrated
  143. req = ScaleReadingRequest(device_id="sb1", weight_grams=-50_000.0)
  144. assert req.weight_grams == -50_000.0
  145. def test_nan_weight_rejected(self):
  146. import math
  147. with pytest.raises(ValidationError):
  148. ScaleReadingRequest(device_id="sb1", weight_grams=math.nan)
  149. def test_inf_weight_rejected(self):
  150. import math
  151. with pytest.raises(ValidationError):
  152. ScaleReadingRequest(device_id="sb1", weight_grams=math.inf)