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