""" tests/backend/test_security.py Tests for all security controls in api.py: - Email validation and header injection prevention - File size limit - File type validation - Parameter bounds validation - Rate limiting - Dataset name sanitization - Path traversal prevention - Job not found (404) - API docs disabled """ import json import io import pytest # ── Helpers ─────────────────────────────────────────────────────────────────── def _submit(client, csv_content: str, params: dict, email: str = "", content_type: str = "text/csv"): """Helper to POST to /api/jobs.""" return client.post( "/api/jobs", data={ "params": json.dumps(params), "email": email, }, files={"file": ("test.csv", io.BytesIO(csv_content.encode()), content_type)}, ) VALID_CSV = ( "score,sample_assignments\n" + "1.5,0\n" * 10 + "-1.5,1\n" * 10 + "0.1,3\n" * 5 ) VALID_PARAMS = { "components": [2, 3], "nBootstraps": 10, "fitsPerBootstrap": 5, "benignMethod": "avg", "conservativeMonotonicity": True, } # ── Email validation ────────────────────────────────────────────────────────── class TestEmailValidation: def test_valid_email_accepted(self, client): res = _submit(client, VALID_CSV, VALID_PARAMS, email="user@example.com") assert res.status_code == 200 assert "job_id" in res.json() def test_empty_email_accepted(self, client): res = _submit(client, VALID_CSV, VALID_PARAMS, email="") assert res.status_code == 200 def test_invalid_email_rejected(self, client): res = _submit(client, VALID_CSV, VALID_PARAMS, email="notanemail") assert res.status_code == 400 def test_email_header_injection_rejected(self, client): """Newline injection attempt must be rejected.""" malicious = "victim@example.com\nBcc: spam@evil.com" res = _submit(client, VALID_CSV, VALID_PARAMS, email=malicious) assert res.status_code == 400 def test_email_cr_injection_rejected(self, client): malicious = "victim@example.com\r\nBcc: spam@evil.com" res = _submit(client, VALID_CSV, VALID_PARAMS, email=malicious) assert res.status_code == 400 def test_email_with_plus_accepted(self, client): res = _submit(client, VALID_CSV, VALID_PARAMS, email="user+tag@example.com") assert res.status_code == 200 # ── File size limit ─────────────────────────────────────────────────────────── class TestFileSizeLimit: def test_oversized_file_rejected(self, client): # 101 MB of data big_content = b"score,sample_assignments\n" + b"1.5,0\n" * (101 * 1024 * 1024 // 6) res = client.post( "/api/jobs", data={"params": json.dumps(VALID_PARAMS), "email": ""}, files={"file": ("big.csv", io.BytesIO(big_content), "text/csv")}, ) assert res.status_code == 413 def test_normal_size_accepted(self, client): res = _submit(client, VALID_CSV, VALID_PARAMS) assert res.status_code == 200 # ── File type validation ────────────────────────────────────────────────────── class TestFileTypeValidation: def test_csv_content_type_accepted(self, client): res = _submit(client, VALID_CSV, VALID_PARAMS, content_type="text/csv") assert res.status_code == 200 def test_text_plain_accepted(self, client): res = _submit(client, VALID_CSV, VALID_PARAMS, content_type="text/plain") assert res.status_code == 200 def test_json_content_type_rejected(self, client): res = _submit(client, VALID_CSV, VALID_PARAMS, content_type="application/json") assert res.status_code == 400 def test_executable_content_type_rejected(self, client): res = _submit(client, VALID_CSV, VALID_PARAMS, content_type="application/x-executable") assert res.status_code == 400 def test_pdf_content_type_rejected(self, client): res = _submit(client, VALID_CSV, VALID_PARAMS, content_type="application/pdf") assert res.status_code == 400 # ── Parameter bounds ────────────────────────────────────────────────────────── class TestParameterBounds: def test_valid_params_accepted(self, client): res = _submit(client, VALID_CSV, VALID_PARAMS) assert res.status_code == 200 def test_invalid_component_rejected(self, client): p = {**VALID_PARAMS, "components": [5]} res = _submit(client, VALID_CSV, p) assert res.status_code == 400 def test_empty_components_rejected(self, client): p = {**VALID_PARAMS, "components": []} res = _submit(client, VALID_CSV, p) assert res.status_code == 400 def test_zero_bootstraps_rejected(self, client): p = {**VALID_PARAMS, "nBootstraps": 0} res = _submit(client, VALID_CSV, p) assert res.status_code == 400 def test_excess_bootstraps_rejected(self, client): p = {**VALID_PARAMS, "nBootstraps": 1001} res = _submit(client, VALID_CSV, p) assert res.status_code == 400 def test_zero_fits_rejected(self, client): p = {**VALID_PARAMS, "fitsPerBootstrap": 0} res = _submit(client, VALID_CSV, p) assert res.status_code == 400 def test_excess_fits_rejected(self, client): p = {**VALID_PARAMS, "fitsPerBootstrap": 101} res = _submit(client, VALID_CSV, p) assert res.status_code == 400 def test_invalid_benign_method_rejected(self, client): p = {**VALID_PARAMS, "benignMethod": "invalid"} res = _submit(client, VALID_CSV, p) assert res.status_code == 400 def test_non_boolean_monotonicity_rejected(self, client): p = {**VALID_PARAMS, "conservativeMonotonicity": "yes"} res = _submit(client, VALID_CSV, p) assert res.status_code == 400 def test_missing_required_param_rejected(self, client): p = {k: v for k, v in VALID_PARAMS.items() if k != "nBootstraps"} res = _submit(client, VALID_CSV, p) assert res.status_code == 400 def test_invalid_json_params_rejected(self, client): res = client.post( "/api/jobs", data={"params": "not json", "email": ""}, files={"file": ("test.csv", io.BytesIO(VALID_CSV.encode()), "text/csv")}, ) assert res.status_code == 400 # ── Rate limiting ───────────────────────────────────────────────────────────── class TestRateLimiting: def test_rate_limit_enforced(self, client): """11th submission from the same IP within an hour should be blocked.""" # Submit up to the limit for i in range(10): res = _submit(client, VALID_CSV, VALID_PARAMS) assert res.status_code == 200, f"Submission {i+1} should succeed" # 11th should be rate limited res = _submit(client, VALID_CSV, VALID_PARAMS) assert res.status_code == 429 def test_rate_limit_message(self, client): for _ in range(10): _submit(client, VALID_CSV, VALID_PARAMS) res = _submit(client, VALID_CSV, VALID_PARAMS) assert "Too many jobs" in res.json()["detail"] # ── Path traversal ──────────────────────────────────────────────────────────── class TestPathTraversal: def _make_complete_job(self, client): """Submit a job and manually mark it complete for file endpoint testing.""" res = _submit(client, VALID_CSV, VALID_PARAMS) assert res.status_code == 200 return res.json()["job_id"] def test_path_traversal_rejected(self, client): job_id = self._make_complete_job(client) res = client.get(f"/api/jobs/{job_id}/files/../../../etc/passwd") assert res.status_code in (400, 404) def test_double_dot_filename_rejected(self, client): job_id = self._make_complete_job(client) res = client.get(f"/api/jobs/{job_id}/files/..%2F..%2Fetc%2Fpasswd") assert res.status_code in (400, 404) def test_valid_filename_allowed(self, client): """A well-formed filename should not be blocked by the path check.""" job_id = self._make_complete_job(client) # File won't exist since pipeline didn't run, but should get 404 not 400 res = client.get(f"/api/jobs/{job_id}/files/output.json") assert res.status_code == 404 # ── Job not found ───────────────────────────────────────────────────────────── class TestJobNotFound: def test_status_404_for_unknown_job(self, client): res = client.get("/api/jobs/00000000-0000-0000-0000-000000000000/status") assert res.status_code == 404 def test_results_404_for_unknown_job(self, client): res = client.get("/api/jobs/00000000-0000-0000-0000-000000000000/results") assert res.status_code == 404 def test_fits_404_for_unknown_job(self, client): res = client.get("/api/jobs/00000000-0000-0000-0000-000000000000/fits") assert res.status_code == 404 def test_csv_404_for_unknown_job(self, client): res = client.get("/api/jobs/00000000-0000-0000-0000-000000000000/csv") assert res.status_code == 404 def test_meta_404_for_unknown_job(self, client): res = client.get("/api/jobs/00000000-0000-0000-0000-000000000000/meta") assert res.status_code == 404 # ── API docs disabled ───────────────────────────────────────────────────────── class TestApiDocsDisabled: def test_swagger_ui_disabled(self, client): res = client.get("/docs") assert res.status_code == 404 def test_redoc_disabled(self, client): res = client.get("/redoc") assert res.status_code == 404 def test_openapi_json_disabled(self, client): res = client.get("/openapi.json") assert res.status_code == 404 # ── Error sanitization ──────────────────────────────────────────────────────── class TestErrorSanitization: def test_failed_job_returns_generic_error(self, client): """A failed job should not expose internal paths or stderr in the response.""" res = _submit(client, VALID_CSV, VALID_PARAMS) job_id = res.json()["job_id"] # Manually mark job as failed with internal error detail from backend.api import jobs jobs[job_id]["status"] = "failed" jobs[job_id]["error"] = "Pipeline encountered an error processing your dataset." jobs[job_id]["error_internal"] = "/var/www/html/ExCALIBR/pipeline/secret_path.py line 42" res = client.get(f"/api/jobs/{job_id}/status") assert res.status_code == 200 body = res.json() assert body["status"] == "failed" # Generic message should be present assert "error" in body # Internal details must NOT be exposed assert "secret_path" not in body["error"] assert "/var/www" not in body["error"]