test_t3_security.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Security tier tests — T3 dataset approval boundary and injection defenses.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import unittest |
| 6 | |
| 7 | from scooling_lab_helpers import valid_payload |
| 8 | |
| 9 | from scooling_lab.dataset_review import ( |
| 10 | DatasetStore, |
| 11 | RejectionReasonCode, |
| 12 | default_dataset_shape, |
| 13 | require_dataset_id, |
| 14 | validate_review_request, |
| 15 | ) |
| 16 | from scooling_lab.errors import ApiError, ErrorCode |
| 17 | from scooling_lab.service import TrainingApiService |
| 18 | from scooling_lab.store import TrainingJobStore |
| 19 | |
| 20 | |
| 21 | class T3SecurityDatasetBoundaryTests(unittest.TestCase): |
| 22 | """Security tests for dataset approval boundaries and injection rejection.""" |
| 23 | |
| 24 | def test_security_t3_unknown_dataset_id_refused_at_job_creation(self) -> None: |
| 25 | """A dataset id that was never registered is refused with DATASET_NOT_APPROVED.""" |
| 26 | |
| 27 | ds_store = DatasetStore() |
| 28 | service = TrainingApiService(TrainingJobStore(), dataset_store=ds_store) |
| 29 | with self.assertRaises(ApiError) as raised: |
| 30 | service.create_training_job( |
| 31 | { |
| 32 | "idempotencyKey": "sec-unknown-ds", |
| 33 | "datasetId": "totally-unknown-ds-id", |
| 34 | "modelId": "fixture-tiny-llm", |
| 35 | "requestedBy": "security-test", |
| 36 | } |
| 37 | ) |
| 38 | self.assertEqual(raised.exception.code, ErrorCode.DATASET_NOT_APPROVED) |
| 39 | |
| 40 | def test_security_t3_rejected_dataset_id_refused_at_job_creation(self) -> None: |
| 41 | """A rejected dataset cannot slip into job creation.""" |
| 42 | |
| 43 | ds_store = DatasetStore() |
| 44 | ds_store.register_shape( |
| 45 | "sec-rejected-ds", |
| 46 | default_dataset_shape(RejectionReasonCode.POLICY_VIOLATION), |
| 47 | ) |
| 48 | ds_store.submit_for_review("sec-rejected-ds") |
| 49 | ds_store.reject("sec-rejected-ds", RejectionReasonCode.POLICY_VIOLATION) |
| 50 | service = TrainingApiService(TrainingJobStore(), dataset_store=ds_store) |
| 51 | with self.assertRaises(ApiError) as raised: |
| 52 | service.create_training_job( |
| 53 | { |
| 54 | "idempotencyKey": "sec-rejected", |
| 55 | "datasetId": "sec-rejected-ds", |
| 56 | "modelId": "fixture-tiny-llm", |
| 57 | "requestedBy": "security-test", |
| 58 | } |
| 59 | ) |
| 60 | self.assertEqual(raised.exception.code, ErrorCode.DATASET_NOT_APPROVED) |
| 61 | |
| 62 | def test_security_t3_injection_shaped_dataset_ids_rejected(self) -> None: |
| 63 | """Injection-shaped dataset ids are refused by the validator.""" |
| 64 | |
| 65 | injection_ids = ( |
| 66 | "../private/dataset", |
| 67 | "https://attacker.invalid/ds", |
| 68 | "ds;rm${IFS}-rf", |
| 69 | "`whoami`", |
| 70 | "<script>x</script>", |
| 71 | "d", # too short |
| 72 | "a" * 97, # too long |
| 73 | ) |
| 74 | for bad in injection_ids: |
| 75 | with self.subTest(bad=bad): |
| 76 | with self.assertRaises(ApiError): |
| 77 | require_dataset_id(bad) |
| 78 | |
| 79 | def test_security_t3_injection_shaped_reason_strings_rejected(self) -> None: |
| 80 | """Injection-shaped strings are not accepted as rejection reason codes.""" |
| 81 | |
| 82 | injection_reasons = ( |
| 83 | "https://attacker.invalid/reason", |
| 84 | "../etc/passwd", |
| 85 | "POLICY_VIOLATION; rm -rf /", |
| 86 | "because I said so", |
| 87 | "", |
| 88 | ) |
| 89 | from scooling_lab.dataset_review import require_rejection_reason |
| 90 | |
| 91 | for bad in injection_reasons: |
| 92 | with self.subTest(bad=bad): |
| 93 | with self.assertRaises(ApiError): |
| 94 | require_rejection_reason(bad) |
| 95 | |
| 96 | def test_security_t3_no_free_text_reflected_in_dataset_rejection(self) -> None: |
| 97 | """Rejection records never echo caller-supplied text.""" |
| 98 | |
| 99 | ds_store = DatasetStore() |
| 100 | ds_store.register_shape( |
| 101 | "sec-reflect-check", |
| 102 | default_dataset_shape(RejectionReasonCode.SCHEMA_MISMATCH), |
| 103 | ) |
| 104 | ds_store.submit_for_review("sec-reflect-check") |
| 105 | ds_store.reject("sec-reflect-check", RejectionReasonCode.SCHEMA_MISMATCH) |
| 106 | public_str = str(ds_store.get("sec-reflect-check").to_public_dict()) |
| 107 | self.assertNotIn("caller supplied reason text", public_str) |
| 108 | self.assertNotIn("free-text", public_str) |
| 109 | |
| 110 | def test_security_t3_review_request_rejects_unknown_keys(self) -> None: |
| 111 | """Unknown keys in a review payload are rejected, including injections.""" |
| 112 | |
| 113 | injection_payloads = ( |
| 114 | {"action": "approve", "workerUrl": "http://attacker.invalid"}, |
| 115 | {"action": "reject", "reasonCode": "POLICY_VIOLATION", "callbackUrl": "x"}, |
| 116 | {"action": "approve", "reasonCode": "POLICY_VIOLATION"}, |
| 117 | ) |
| 118 | for payload in injection_payloads: |
| 119 | with self.subTest(payload=payload): |
| 120 | with self.assertRaises(ApiError): |
| 121 | validate_review_request(payload) |
| 122 | |
| 123 | def test_security_t3_dataset_not_approved_error_does_not_echo_id(self) -> None: |
| 124 | """The DATASET_NOT_APPROVED error message does not echo the dataset id.""" |
| 125 | |
| 126 | ds_store = DatasetStore() |
| 127 | service = TrainingApiService(TrainingJobStore(), dataset_store=ds_store) |
| 128 | try: |
| 129 | service.create_training_job( |
| 130 | { |
| 131 | "idempotencyKey": "sec-no-echo", |
| 132 | "datasetId": "fixture:synthetic-tiny-v1", |
| 133 | "modelId": "fixture-tiny-llm", |
| 134 | "requestedBy": "security-test", |
| 135 | } |
| 136 | ) |
| 137 | except ApiError as exc: |
| 138 | # Override store to simulate unapproved state; verify message is generic. |
| 139 | msg = exc.message |
| 140 | self.assertNotIn("fixture:synthetic-tiny-v1", msg) |
| 141 | self.assertNotIn("attacker", msg) |
| 142 | |
| 143 | # Force an unapproved scenario with a clearly synthetic id. |
| 144 | ds_store2 = DatasetStore() |
| 145 | service2 = TrainingApiService(TrainingJobStore(), dataset_store=ds_store2) |
| 146 | ds_store2.register("echo-test-ds") |
| 147 | with self.assertRaises(ApiError) as raised: |
| 148 | service2.create_training_job( |
| 149 | { |
| 150 | "idempotencyKey": "sec-no-echo2", |
| 151 | "datasetId": "echo-test-ds", |
| 152 | "modelId": "fixture-tiny-llm", |
| 153 | "requestedBy": "security-test", |
| 154 | } |
| 155 | ) |
| 156 | # Error message must not echo the dataset id. |
| 157 | self.assertNotIn("echo-test-ds", raised.exception.message) |
| 158 | self.assertEqual(raised.exception.code, ErrorCode.DATASET_NOT_APPROVED) |
| 159 | |
| 160 | def test_security_t3_registered_dataset_id_refused_before_submission(self) -> None: |
| 161 | """A registered dataset that has not been submitted blocks job creation.""" |
| 162 | |
| 163 | ds_store = DatasetStore() |
| 164 | ds_store.register("pending-review-ds") |
| 165 | service = TrainingApiService(TrainingJobStore(), dataset_store=ds_store) |
| 166 | with self.assertRaises(ApiError) as raised: |
| 167 | service.create_training_job( |
| 168 | { |
| 169 | "idempotencyKey": "sec-pending", |
| 170 | "datasetId": "pending-review-ds", |
| 171 | "modelId": "fixture-tiny-llm", |
| 172 | "requestedBy": "security-test", |
| 173 | } |
| 174 | ) |
| 175 | self.assertEqual(raised.exception.code, ErrorCode.DATASET_NOT_APPROVED) |
| 176 | |
| 177 | |
| 178 | if __name__ == "__main__": |
| 179 | unittest.main() |