diff --git a/frontend/src/test/security/api-security.test.ts b/frontend/src/test/security/api-security.test.ts index 0fd5376dd..c486c1b16 100644 --- a/frontend/src/test/security/api-security.test.ts +++ b/frontend/src/test/security/api-security.test.ts @@ -2,9 +2,9 @@ * TC_PERF_001 - TC_PERF_002: Performance Tests * TC_SEC_001 - TC_SEC_002: Security Tests * TC_INT_001: Integration Resilience Tests - * + * * Tests for load handling, stress conditions, and security vulnerabilities. - * + * * Compliance: OWASP Top 10 */ @@ -68,21 +68,34 @@ describe('API Security Tests', () => { '', '', 'javascript:alert(1)', + 'data:text/html,', + 'vbscript:msgbox(1)', '', '">', + 'oononerror=alert(1)', ]; it.each(xssPayloads)( 'should sanitize XSS payload: %s', async (payload) => { const sanitizeHTML = (input: string): string => { - return input + // Escape markup first so attribute injection cannot reopen tags. + let out = input + .replace(/&/g, '&') .replace(//g, '>') .replace(/"/g, '"') - .replace(/'/g, ''') - .replace(/javascript:/gi, '') - .replace(/on\w+=/gi, ''); + .replace(/'/g, '''); + // Neutralize dangerous URL schemes (javascript, data, vbscript). + out = out.replace(/(?:javascript|data|vbscript):/gi, ''); + // Strip event-handler assignments; repeat until stable so + // overlapping prefixes like "oononerror=" cannot survive. + let prev = ''; + while (out !== prev) { + prev = out; + out = out.replace(/on\w+\s*=/gi, ''); + } + return out; }; const sanitized = sanitizeHTML(payload); @@ -90,6 +103,8 @@ describe('API Security Tests', () => { expect(sanitized).not.toContain(' bool: hosts = _effective_db_hosts(database_url) return bool(hosts) and all(h in LOCAL_HOSTS for h in hosts) + # Matches the AUTH_TEST_BYPASS default tenant and the legacy phase-2b seed # tenant, so a key minted here composes with existing local fixtures. TEST_TENANT_ID = "11111111-1111-1111-1111-111111111111" @@ -227,10 +229,7 @@ def _seed_billing_status(tenant_id: str, allow_production: bool = False) -> str: """ redis_url = os.getenv("REDIS_URL", "redis://localhost:6379/0") if not _is_local_url(redis_url) and not allow_production: - return ( - "SKIPPED (REDIS_URL is not a localhost URL; not writing billing " - "state without --allow-production)." - ) + return "SKIPPED (REDIS_URL is not a localhost URL; not writing billing state without --allow-production)." try: import redis as redis_lib diff --git a/services/ingestion/tests/test_alerts_1342.py b/services/ingestion/tests/test_alerts_1342.py index 2e5fb08d1..6469ed475 100644 --- a/services/ingestion/tests/test_alerts_1342.py +++ b/services/ingestion/tests/test_alerts_1342.py @@ -9,6 +9,7 @@ from datetime import datetime, timezone from typing import Any, Optional +from urllib.parse import urlparse import pytest from app import alerts as alerts_mod @@ -294,7 +295,9 @@ def test_full_row_maps_correctly(self): assert a.severity == "critical" assert a.metadata["recall_number"] == "F-123-2026" assert a.metadata["match_tier"] == "lot_code" - assert a.metadata["fda_url"].startswith("https://www.accessdata.fda.gov") + fda = urlparse(a.metadata["fda_url"]) + assert fda.scheme == "https" + assert fda.hostname == "www.accessdata.fda.gov" assert a.acknowledged is False def test_supplier_match_tier(self): diff --git a/services/ingestion/tests/test_models_ssrf_1342.py b/services/ingestion/tests/test_models_ssrf_1342.py index e98d9f7b4..4a5a39137 100644 --- a/services/ingestion/tests/test_models_ssrf_1342.py +++ b/services/ingestion/tests/test_models_ssrf_1342.py @@ -31,6 +31,7 @@ import socket import sys from pathlib import Path +from urllib.parse import urlparse import pytest from pydantic import ValidationError @@ -293,7 +294,9 @@ class TestIngestRequestValidator: def test_public_url_accepted(self, monkeypatch: pytest.MonkeyPatch) -> None: _patch_resolver(monkeypatch, ["93.184.216.34"]) # example.com req = IngestRequest(url="https://example.com/api", source_system="fda") - assert str(req.url).startswith("https://example.com") + parsed = urlparse(str(req.url)) + assert parsed.scheme == "https" + assert parsed.hostname == "example.com" assert req.source_system == "fda" def test_private_url_rejected_via_validator(self, monkeypatch: pytest.MonkeyPatch) -> None: diff --git a/services/ingestion/tests/test_regengine_sources_rest_qa.py b/services/ingestion/tests/test_regengine_sources_rest_qa.py index 237d797fe..bc77cc793 100644 --- a/services/ingestion/tests/test_regengine_sources_rest_qa.py +++ b/services/ingestion/tests/test_regengine_sources_rest_qa.py @@ -82,7 +82,7 @@ def test_first_call_does_not_sleep(self): finally: time.sleep = original assert slept == [] - assert "example.com" in limiter._last_request + assert limiter._last_request.get("example.com") is not None def test_second_immediate_call_sleeps(self): slept = [] @@ -207,8 +207,7 @@ def test_helpers_record_expected_actions(self, tmp_path): logger.log_store("doc-1", "success", "documents/v/doc-1.bin") logger.log_skip("doc-2", "duplicate") records = [ - json.loads(line) - for line in (tmp_path / "job_job-2.jsonl").read_text(encoding="utf-8").strip().split("\n") + json.loads(line) for line in (tmp_path / "job_job-2.jsonl").read_text(encoding="utf-8").strip().split("\n") ] assert [r["action"] for r in records] == ["fetch", "parse", "store", "skip"] assert records[0]["details"] == {"http_status": 200} @@ -436,7 +435,9 @@ def test_run_job_stores_and_dedups(self, tmp_path, monkeypatch): from regengine_ingestion.models import SourceMetadata meta = SourceMetadata( - source_url="https://x", fetch_timestamp=datetime.now(timezone.utc), http_headers={"Content-Type": "text/plain"} + source_url="https://x", + fetch_timestamp=datetime.now(timezone.utc), + http_headers={"Content-Type": "text/plain"}, ) items = [ (b"doc-one", meta, {"title": "One"}), diff --git a/services/ingestion/tests/test_routes_ingest_internal_urls.py b/services/ingestion/tests/test_routes_ingest_internal_urls.py index d0dfef619..eec49fd97 100644 --- a/services/ingestion/tests/test_routes_ingest_internal_urls.py +++ b/services/ingestion/tests/test_routes_ingest_internal_urls.py @@ -17,6 +17,7 @@ from datetime import datetime, timezone from pathlib import Path from types import SimpleNamespace +from urllib.parse import urlparse import pytest from fastapi import HTTPException @@ -68,17 +69,13 @@ def _event_kwargs(**overrides): class TestStoredSourceMarkers: @pytest.mark.parametrize("marker", [MANUAL_MARKER, UPLOAD_MARKER]) - def test_document_accepts_markers_without_dns( - self, monkeypatch: pytest.MonkeyPatch, marker: str - ) -> None: + def test_document_accepts_markers_without_dns(self, monkeypatch: pytest.MonkeyPatch, marker: str) -> None: monkeypatch.setattr("app.models.socket.getaddrinfo", _boom_resolver) doc = NormalizedDocument(**_doc_kwargs(source_url=marker)) assert doc.source_url == marker @pytest.mark.parametrize("marker", [MANUAL_MARKER, UPLOAD_MARKER]) - def test_event_accepts_markers_without_dns( - self, monkeypatch: pytest.MonkeyPatch, marker: str - ) -> None: + def test_event_accepts_markers_without_dns(self, monkeypatch: pytest.MonkeyPatch, marker: str) -> None: monkeypatch.setattr("app.models.socket.getaddrinfo", _boom_resolver) event = NormalizedEvent(**_event_kwargs(source_url=marker)) assert event.source_url == marker @@ -113,7 +110,9 @@ def _fake(host, port, *args, **kwargs): monkeypatch.setattr("app.models.socket.getaddrinfo", _fake) doc = NormalizedDocument(**_doc_kwargs(source_url="https://example.com/reg/1")) - assert doc.source_url.startswith("https://example.com") + parsed = urlparse(doc.source_url) + assert parsed.scheme == "https" + assert parsed.hostname == "example.com" class TestFetchBoundaryStillStrict: @@ -149,9 +148,7 @@ def _fail(*_a, **_kw): class TestProcessAndEmitWithMarkers: @pytest.mark.parametrize("marker", [MANUAL_MARKER, UPLOAD_MARKER]) - def test_emit_succeeds_for_internal_markers( - self, monkeypatch: pytest.MonkeyPatch, marker: str - ) -> None: + def test_emit_succeeds_for_internal_markers(self, monkeypatch: pytest.MonkeyPatch, marker: str) -> None: import app.routes_ingest as routes_ingest monkeypatch.setattr(routes_ingest, "put_bytes", lambda *a, **k: "s3://raw/x") @@ -190,9 +187,7 @@ def _boom(*_a, **_k): ) assert exc.value.status_code == 422 - async def test_direct_ingest_succeeds_with_default_marker( - self, monkeypatch: pytest.MonkeyPatch - ) -> None: + async def test_direct_ingest_succeeds_with_default_marker(self, monkeypatch: pytest.MonkeyPatch) -> None: import app.routes_ingest as routes_ingest from app.models import DirectIngestRequest diff --git a/services/ingestion/tests/test_stripe_helpers_1342.py b/services/ingestion/tests/test_stripe_helpers_1342.py index 4bdc4d7f4..6864ed7ac 100644 --- a/services/ingestion/tests/test_stripe_helpers_1342.py +++ b/services/ingestion/tests/test_stripe_helpers_1342.py @@ -109,8 +109,8 @@ def test_extra_exact_hosts_appended(self, monkeypatch): exact, _ = _allowed_redirect_hosts() # Lowercased and trimmed; empty entries from double-commas # skipped (covers the ``if host:`` guard). - assert "staging.example.com" in exact - assert "other.example.com" in exact + assert any(host == "staging.example.com" for host in exact) + assert any(host == "other.example.com" for host in exact) def test_undotted_suffixes_are_rejected(self, monkeypatch, caplog): # ``regengine.co`` as a suffix (no leading dot) would match @@ -123,8 +123,8 @@ def test_undotted_suffixes_are_rejected(self, monkeypatch, caplog): ".ok.example.com,undotted.bad, ,.also-ok.example.com", ) _, suffixes = _allowed_redirect_hosts() - assert ".ok.example.com" in suffixes - assert ".also-ok.example.com" in suffixes + assert any(suffix == ".ok.example.com" for suffix in suffixes) + assert any(suffix == ".also-ok.example.com" for suffix in suffixes) # Un-dotted and empty entries skipped. assert "undotted.bad" not in suffixes assert "" not in suffixes diff --git a/services/scheduler/tests/test_notifications.py b/services/scheduler/tests/test_notifications.py index 88808c1c3..3d2e62cd1 100644 --- a/services/scheduler/tests/test_notifications.py +++ b/services/scheduler/tests/test_notifications.py @@ -15,6 +15,7 @@ from datetime import datetime, timezone from pathlib import Path from unittest.mock import MagicMock, patch +from urllib.parse import urlparse os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:") os.environ.setdefault("ENVIRONMENT", "test") @@ -368,6 +369,7 @@ def test_delivers_to_multiple_urls(self): results = n.notify([_make_item()]) assert len(results) == 2 urls = {r.url for r in results} - assert "https://hook1.example.com" in urls - assert "https://hook2.example.com" in urls + hosts = {urlparse(u).hostname for u in urls} + assert hosts == {"hook1.example.com", "hook2.example.com"} + assert all(urlparse(u).scheme == "https" for u in urls) n.close()