aaronrene / scooling-lab public
test_security.py python
190 lines 8.2 KB
Raw
sha256:ca1d0e687bff8686b37126eb8e4fb6d38b40e189352a4920d66ae89c7274e340 Add T4 job cancellation retry and validation Human minor ⚠ breaking 41 days ago
1 """Security tier tests for Scooling Lab request and dependency boundaries."""
2
3 from __future__ import annotations
4
5 import unittest
6
7 from scooling_lab_helpers import PROJECT_ROOT, valid_payload
8
9 from scooling_lab.bom import audit_repository_paths
10 from scooling_lab.contracts import TrainingJobRequest
11 from scooling_lab.fake_worker import fixture_dataset_bytes
12 from scooling_lab.errors import ApiError, ErrorCode
13 from scooling_lab.license_policy import BomEntry, LicensePolicyError, validate_entry
14 from scooling_lab.service import TrainingApiService
15 from scooling_lab.store import TrainingJobStore
16
17
18 class ScoolingLabSecurityTests(unittest.TestCase):
19 """Security tests for injection rejection and AGPL boundary enforcement."""
20
21 def test_security_rejects_path_traversal_command_and_url_injection(self) -> None:
22 """Untrusted paths, commands, URLs, callbacks, and worker fields fail closed."""
23
24 attacks: list[dict[str, object]] = []
25 path_payload = valid_payload("path")
26 path_payload["datasetId"] = "../private"
27 attacks.append(path_payload)
28
29 command_payload = valid_payload("command")
30 command_payload["trainingParameters"] = {"epochs": 1, "command": "rm -rf /"}
31 attacks.append(command_payload)
32
33 callback_payload = valid_payload("callback")
34 callback_payload["callbackUrl"] = "https://attacker.invalid/callback"
35 attacks.append(callback_payload)
36
37 worker_payload = valid_payload("worker")
38 worker_payload["workerUrl"] = "http://127.0.0.1:9999"
39 attacks.append(worker_payload)
40
41 for payload in attacks:
42 with self.subTest(payload=payload):
43 with self.assertRaises(ApiError):
44 TrainingJobRequest.from_mapping(payload)
45
46 def test_security_secret_scan_and_bom_audit_are_wired(self) -> None:
47 """CI contains gitleaks and the repository path audit passes locally."""
48
49 workflow = (PROJECT_ROOT / ".github/workflows/ci.yml").read_text(
50 encoding="utf-8"
51 )
52 self.assertIn("gitleaks detect", workflow)
53 self.assertIn("python -m scooling_lab.bom", workflow)
54 audit_repository_paths(PROJECT_ROOT)
55
56 def test_security_agpl_package_and_blocked_paths_are_rejected(self) -> None:
57 """AGPL license ids and Studio/CLI paths cannot enter the BOM."""
58
59 with self.assertRaises(LicensePolicyError):
60 validate_entry(
61 BomEntry(
62 name="studio",
63 version="1.0.0",
64 license="AGPL-3.0-only",
65 source_path="studio/backend/run.py",
66 evidence="fixture",
67 )
68 )
69
70 def test_security_provenance_excludes_synthetic_fixture_text_markers(self) -> None:
71 """Provenance output never contains text from the fixture dataset."""
72
73 marker = "synthetic learner practices"
74 self.assertIn(marker, fixture_dataset_bytes().decode("utf-8"))
75 service = TrainingApiService(TrainingJobStore())
76 created = service.create_training_job(valid_payload("marker-absence"))
77 provenance_text = str(service.get_provenance(str(created["id"])))
78
79 self.assertNotIn(marker, provenance_text)
80 self.assertNotIn("study habits", provenance_text)
81 self.assertNotIn("astronomy facts", provenance_text)
82
83 def test_security_forged_artifact_id_cannot_delete_another_job(self) -> None:
84 """A valid artifact id from one job cannot delete a different job."""
85
86 service = TrainingApiService(TrainingJobStore())
87 first = service.create_training_job(valid_payload("forged-a"))
88 second = service.create_training_job(valid_payload("forged-b"))
89 first_job_id = str(first["id"])
90 second_job_id = str(second["id"])
91 second_artifact_id = str(service.list_artifacts(second_job_id)["artifacts"][0]["id"])
92
93 with self.assertRaises(ApiError):
94 service.delete_artifact(first_job_id, second_artifact_id)
95
96 self.assertEqual(service.get_training_job(first_job_id)["status"], "succeeded")
97 self.assertEqual(service.get_training_job(second_job_id)["status"], "succeeded")
98
99 def test_security_path_traversal_and_injection_ids_are_rejected(self) -> None:
100 """Job and artifact ids reject path traversal and command characters."""
101
102 service = TrainingApiService(TrainingJobStore())
103 created = service.create_training_job(valid_payload("id-injection"))
104 job_id = str(created["id"])
105 artifact_id = str(service.list_artifacts(job_id)["artifacts"][0]["id"])
106
107 attacks = (
108 ("../private", artifact_id),
109 (job_id, "../artifact"),
110 (job_id, f"{artifact_id};rm-rf"),
111 ("https://attacker.invalid/job", artifact_id),
112 )
113 for attack_job_id, attack_artifact_id in attacks:
114 with self.subTest(job_id=attack_job_id, artifact_id=attack_artifact_id):
115 with self.assertRaises(ApiError):
116 service.delete_artifact(attack_job_id, attack_artifact_id)
117 with self.assertRaises(LicensePolicyError):
118 validate_entry(
119 BomEntry(
120 name="unsloth-cli",
121 version="1.0.0",
122 license="Apache-2.0",
123 source_path="vendor/unsloth_cli/app.py",
124 evidence="fixture",
125 )
126 )
127
128 def test_security_t4_dataset_shape_rejections_are_enum_only(self) -> None:
129 """Unknown and forbidden dataset metadata reject without reflecting fields."""
130
131 service = TrainingApiService(TrainingJobStore())
132 unknown = service.register_dataset(
133 {
134 "datasetId": "security-unknown-shape",
135 "rowCount": 3,
136 "declaredSchema": {
137 "exampleId": "string",
138 "inputTokenCount": "integer",
139 "outputTokenCount": "integer",
140 "split": "string",
141 },
142 "extraField": "not-returned",
143 }
144 )
145 forbidden = service.register_dataset(
146 {
147 "datasetId": "security-forbidden-shape",
148 "rowCount": 3,
149 "declaredSchema": {
150 "exampleId": "string",
151 "inputTokenCount": "integer",
152 "outputTokenCount": "integer",
153 "promptPayload": "string",
154 "split": "string",
155 },
156 }
157 )
158
159 unknown_decision = service.submit_dataset_for_review(str(unknown["datasetId"]))
160 forbidden_decision = service.submit_dataset_for_review(str(forbidden["datasetId"]))
161 combined = f"{unknown_decision} {forbidden_decision}"
162
163 self.assertEqual(unknown_decision["status"], "rejected")
164 self.assertEqual(unknown_decision["rejectionReasonCode"], "FORMAT_INVALID")
165 self.assertEqual(forbidden_decision["status"], "rejected")
166 self.assertEqual(forbidden_decision["rejectionReasonCode"], "POLICY_VIOLATION")
167 self.assertNotIn("extraField", combined)
168 self.assertNotIn("not-returned", combined)
169 self.assertNotIn("promptPayload", combined)
170
171 def test_security_t4_cancel_and_retry_reject_injection_shaped_ids(self) -> None:
172 """Cancel and retry validate job ids before any store lookup."""
173
174 service = TrainingApiService(TrainingJobStore(), auto_run_worker=False)
175 created = service.create_training_job(valid_payload("security-t4-id"))
176 cancelled = service.cancel_training_job(str(created["id"]))
177 self.assertEqual(cancelled["status"], "cancelled")
178
179 for unsafe_job_id in ("../job-id", "job_" + "a" * 24 + ";x"):
180 with self.subTest(unsafe_job_id=unsafe_job_id):
181 with self.assertRaises(ApiError) as cancel_error:
182 service.cancel_training_job(unsafe_job_id)
183 self.assertEqual(cancel_error.exception.code, ErrorCode.VALIDATION_ERROR)
184 with self.assertRaises(ApiError) as retry_error:
185 service.retry_training_job(unsafe_job_id)
186 self.assertEqual(retry_error.exception.code, ErrorCode.VALIDATION_ERROR)
187
188
189 if __name__ == "__main__":
190 unittest.main()
File History 1 commit
sha256:ca1d0e687bff8686b37126eb8e4fb6d38b40e189352a4920d66ae89c7274e340 Add T4 job cancellation retry and validation Human minor ⚠ 41 days ago