From 72d835ad86e2798ca65adc29512c5d118acbe121 Mon Sep 17 00:00:00 2001 From: Naman Singh Date: Tue, 7 Jul 2026 09:44:56 +0530 Subject: [PATCH 1/2] fix: enable TLS certificate verification across all scanners and crawler SECUSCAN-001 (Critical): TLS certificate verification was globally disabled with verify=False in 4 locations (crawler, XSS scanner, API scanner, network vulnerability scanner), leaving all outbound HTTPS connections vulnerable to MITM attacks. Added verify_ssl: bool = True to Settings (configurable via SECUSCAN_VERIFY_SSL env var) and updated all httpx.AsyncClient calls to reference it. --- backend/secuscan/config.py | 1 + backend/secuscan/crawler.py | 4 +++- backend/secuscan/scanners/api_scanner.py | 3 ++- backend/secuscan/scanners/network_vulnerability_scanner.py | 3 ++- backend/secuscan/scanners/xss_validation_scanner.py | 3 ++- 5 files changed, 10 insertions(+), 4 deletions(-) diff --git a/backend/secuscan/config.py b/backend/secuscan/config.py index 78e733fcf..0e0471aec 100644 --- a/backend/secuscan/config.py +++ b/backend/secuscan/config.py @@ -38,6 +38,7 @@ class Settings(BaseSettings): # Security safe_mode_default: bool = True + verify_ssl: bool = True dns_resolution_timeout_seconds: float = 1.5 dns_cache_ttl_seconds: int = 60 dns_rebind_check: bool = True diff --git a/backend/secuscan/crawler.py b/backend/secuscan/crawler.py index 183a50d30..7013184af 100644 --- a/backend/secuscan/crawler.py +++ b/backend/secuscan/crawler.py @@ -9,6 +9,8 @@ import httpx +from .config import settings + class _SurfaceParser(HTMLParser): def __init__(self) -> None: @@ -86,7 +88,7 @@ async def crawl_target( timeout=timeout, headers=headers, cookies=cookies or {}, - verify=False, + verify=settings.verify_ssl, ) as client: response = await client.get(url) diff --git a/backend/secuscan/scanners/api_scanner.py b/backend/secuscan/scanners/api_scanner.py index d3ed5a6d6..1cef1d076 100644 --- a/backend/secuscan/scanners/api_scanner.py +++ b/backend/secuscan/scanners/api_scanner.py @@ -6,6 +6,7 @@ import httpx +from ..config import settings from .base import BaseScanner from ..crawler import crawl_target @@ -52,7 +53,7 @@ async def run(self, target: str, inputs: Dict[str, Any]) -> Dict[str, Any]: timeout=timeout, headers={str(k): str(v) for k, v in extra_headers.items()}, cookies={str(k): str(v) for k, v in cookies.items()}, - verify=False, + verify=settings.verify_ssl, ) as client: for path in self.COMMON_SPEC_PATHS: url = urljoin(target.rstrip("/") + "/", path.lstrip("/")) diff --git a/backend/secuscan/scanners/network_vulnerability_scanner.py b/backend/secuscan/scanners/network_vulnerability_scanner.py index cc0371d5f..b4a9cac41 100644 --- a/backend/secuscan/scanners/network_vulnerability_scanner.py +++ b/backend/secuscan/scanners/network_vulnerability_scanner.py @@ -8,6 +8,7 @@ import httpx +from ..config import settings from .base import BaseScanner from ..knowledgebase import KnowledgeBase from ..plugins import get_plugin_manager @@ -186,7 +187,7 @@ async def _probe_http(self, host: str, port: int, *, tls: bool) -> Optional[Dict scheme = "https" if tls else "http" url = f"{scheme}://{host}:{port}/" try: - async with httpx.AsyncClient(follow_redirects=True, timeout=5, verify=False) as client: + async with httpx.AsyncClient(follow_redirects=True, timeout=5, verify=settings.verify_ssl) as client: response = await client.get(url) except Exception: return None diff --git a/backend/secuscan/scanners/xss_validation_scanner.py b/backend/secuscan/scanners/xss_validation_scanner.py index 7e806f3ec..a66ae0308 100644 --- a/backend/secuscan/scanners/xss_validation_scanner.py +++ b/backend/secuscan/scanners/xss_validation_scanner.py @@ -6,6 +6,7 @@ import httpx +from ..config import settings from .base import BaseScanner @@ -28,7 +29,7 @@ async def run(self, target: str, inputs: Dict[str, Any]) -> Dict[str, Any]: findings: List[Dict[str, Any]] = [] self.update_progress(0.25) - async with httpx.AsyncClient(follow_redirects=True, timeout=int(inputs.get("timeout") or 10), verify=False) as client: + async with httpx.AsyncClient(follow_redirects=True, timeout=int(inputs.get("timeout") or 10), verify=settings.verify_ssl) as client: response = await client.get(probe_url) body = response.text From 8fd336b71fe91122bf46fccc4c53449c443888c7 Mon Sep 17 00:00:00 2001 From: Naman Singh Date: Fri, 10 Jul 2026 13:39:43 +0530 Subject: [PATCH 2/2] test: add focused TLS verification tests Covers Settings.verify_ssl default and env override, verifies each scanner (crawler, API, XSS, network vuln) passes the setting to httpx.AsyncClient, and asserts no hardcoded verify=False remains in the patched files. --- testing/backend/unit/test_tls_verification.py | 304 ++++++++++++++++++ 1 file changed, 304 insertions(+) create mode 100644 testing/backend/unit/test_tls_verification.py diff --git a/testing/backend/unit/test_tls_verification.py b/testing/backend/unit/test_tls_verification.py new file mode 100644 index 000000000..be8281d83 --- /dev/null +++ b/testing/backend/unit/test_tls_verification.py @@ -0,0 +1,304 @@ +""" +Tests for TLS certificate verification behaviour. + +Verifies that: + - Settings.verify_ssl defaults to True + - SECUSCAN_VERIFY_SSL env var overrides the default + - crawler.crawl_target passes settings.verify_ssl to httpx.AsyncClient + - APIScanner passes settings.verify_ssl to httpx.AsyncClient + - XSSValidationScanner passes settings.verify_ssl to httpx.AsyncClient + - NetworkVulnerabilityScanner._probe_http passes settings.verify_ssl to httpx.AsyncClient +""" + +from __future__ import annotations + +import importlib +import os +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + + +# --------------------------------------------------------------------------- +# Settings defaults +# --------------------------------------------------------------------------- + + +class TestSettingsVerifySslDefault: + def test_default_is_true(self): + from backend.secuscan.config import Settings + + s = Settings() + assert s.verify_ssl is True + + def test_env_override_false(self, monkeypatch): + monkeypatch.setenv("SECUSCAN_VERIFY_SSL", "false") + from backend.secuscan.config import Settings + + s = Settings() + assert s.verify_ssl is False + + def test_env_override_true(self, monkeypatch): + monkeypatch.setenv("SECUSCAN_VERIFY_SSL", "true") + from backend.secuscan.config import Settings + + s = Settings() + assert s.verify_ssl is True + + +# --------------------------------------------------------------------------- +# Crawler +# --------------------------------------------------------------------------- + + +class TestCrawlerVerifySsl: + @pytest.mark.asyncio + async def test_crawl_target_passes_verify_ssl(self): + mock_response = MagicMock() + mock_response.text = "" + mock_response.url = "https://example.com/" + mock_response.status_code = 200 + mock_response.headers = {} + mock_response.history = [] + + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_response) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + + with patch("backend.secuscan.crawler.httpx.AsyncClient", return_value=mock_client) as mock_cls: + with patch("backend.secuscan.crawler.settings") as mock_settings: + mock_settings.verify_ssl = True + from backend.secuscan.crawler import crawl_target + + await crawl_target("https://example.com") + + mock_cls.assert_called_once() + _, kwargs = mock_cls.call_args + assert kwargs["verify"] is True + + @pytest.mark.asyncio + async def test_crawl_target_verify_ssl_false_when_disabled(self): + mock_response = MagicMock() + mock_response.text = "" + mock_response.url = "https://example.com/" + mock_response.status_code = 200 + mock_response.headers = {} + mock_response.history = [] + + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_response) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + + with patch("backend.secuscan.crawler.httpx.AsyncClient", return_value=mock_client) as mock_cls: + with patch("backend.secuscan.crawler.settings") as mock_settings: + mock_settings.verify_ssl = False + from backend.secuscan.crawler import crawl_target + + await crawl_target("https://example.com") + + _, kwargs = mock_cls.call_args + assert kwargs["verify"] is False + + +# --------------------------------------------------------------------------- +# API Scanner +# --------------------------------------------------------------------------- + + +class TestAPIScannerVerifySsl: + @pytest.mark.asyncio + async def test_api_scanner_passes_verify_ssl(self): + from backend.secuscan.scanners.api_scanner import APIScanner + + scanner = APIScanner("task-verify", None) + + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=MagicMock(status_code=404, text="", headers={})) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + + with patch("backend.secuscan.scanners.api_scanner.httpx.AsyncClient", return_value=mock_client) as mock_cls: + with patch("backend.secuscan.scanners.api_scanner.settings") as mock_settings: + mock_settings.verify_ssl = True + with patch.object(scanner, "_fetch_spec", return_value=None): + with patch.object(scanner, "_probe_graphql", return_value=([], [])): + await scanner.run("https://example.com", {}) + + mock_cls.assert_called() + _, kwargs = mock_cls.call_args + assert kwargs["verify"] is True + + @pytest.mark.asyncio + async def test_api_scanner_verify_ssl_false_when_disabled(self): + from backend.secuscan.scanners.api_scanner import APIScanner + + scanner = APIScanner("task-no-verify", None) + + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=MagicMock(status_code=404, text="", headers={})) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + + with patch("backend.secuscan.scanners.api_scanner.httpx.AsyncClient", return_value=mock_client) as mock_cls: + with patch("backend.secuscan.scanners.api_scanner.settings") as mock_settings: + mock_settings.verify_ssl = False + with patch.object(scanner, "_fetch_spec", return_value=None): + with patch.object(scanner, "_probe_graphql", return_value=([], [])): + await scanner.run("https://example.com", {}) + + _, kwargs = mock_cls.call_args + assert kwargs["verify"] is False + + +# --------------------------------------------------------------------------- +# XSS Validation Scanner +# --------------------------------------------------------------------------- + + +class TestXSSScannerVerifySsl: + @pytest.mark.asyncio + async def test_xss_scanner_passes_verify_ssl(self): + from backend.secuscan.scanners.xss_validation_scanner import XSSValidationScanner + + scanner = XSSValidationScanner("task-xss", None) + + mock_response = MagicMock() + mock_response.text = "" + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_response) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + + with patch("backend.secuscan.scanners.xss_validation_scanner.httpx.AsyncClient", return_value=mock_client) as mock_cls: + with patch("backend.secuscan.scanners.xss_validation_scanner.settings") as mock_settings: + mock_settings.verify_ssl = True + await scanner.run("https://example.com/?q=test", {"timeout": 5}) + + mock_cls.assert_called_once() + _, kwargs = mock_cls.call_args + assert kwargs["verify"] is True + + @pytest.mark.asyncio + async def test_xss_scanner_verify_ssl_false_when_disabled(self): + from backend.secuscan.scanners.xss_validation_scanner import XSSValidationScanner + + scanner = XSSValidationScanner("task-xss-noverify", None) + + mock_response = MagicMock() + mock_response.text = "" + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_response) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + + with patch("backend.secuscan.scanners.xss_validation_scanner.httpx.AsyncClient", return_value=mock_client) as mock_cls: + with patch("backend.secuscan.scanners.xss_validation_scanner.settings") as mock_settings: + mock_settings.verify_ssl = False + await scanner.run("https://example.com/?q=test", {"timeout": 5}) + + _, kwargs = mock_cls.call_args + assert kwargs["verify"] is False + + +# --------------------------------------------------------------------------- +# Network Vulnerability Scanner +# --------------------------------------------------------------------------- + + +class TestNetworkVulnScannerVerifySsl: + @pytest.mark.asyncio + async def test_probe_http_passes_verify_ssl(self): + from backend.secuscan.scanners.network_vulnerability_scanner import NetworkVulnerabilityScanner + + scanner = NetworkVulnerabilityScanner("task-net", None) + + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.text = "Test" + mock_response.headers = {"server": "nginx"} + mock_response.url = "https://192.168.1.1:443/" + + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_response) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + + with patch("backend.secuscan.scanners.network_vulnerability_scanner.httpx.AsyncClient", return_value=mock_client) as mock_cls: + with patch("backend.secuscan.scanners.network_vulnerability_scanner.settings") as mock_settings: + mock_settings.verify_ssl = True + result = await scanner._probe_http("192.168.1.1", 443, tls=True) + + mock_cls.assert_called_once() + _, kwargs = mock_cls.call_args + assert kwargs["verify"] is True + assert result is not None + + @pytest.mark.asyncio + async def test_probe_http_verify_ssl_false_when_disabled(self): + from backend.secuscan.scanners.network_vulnerability_scanner import NetworkVulnerabilityScanner + + scanner = NetworkVulnerabilityScanner("task-net-nv", None) + + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.text = "Test" + mock_response.headers = {"server": "nginx"} + mock_response.url = "https://192.168.1.1:443/" + + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_response) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + + with patch("backend.secuscan.scanners.network_vulnerability_scanner.httpx.AsyncClient", return_value=mock_client) as mock_cls: + with patch("backend.secuscan.scanners.network_vulnerability_scanner.settings") as mock_settings: + mock_settings.verify_ssl = False + result = await scanner._probe_http("192.168.1.1", 443, tls=True) + + _, kwargs = mock_cls.call_args + assert kwargs["verify"] is False + assert result is not None + + +# --------------------------------------------------------------------------- +# No hardcoded verify=False remains +# --------------------------------------------------------------------------- + + +class TestNoHardcodedVerifyFalse: + """Ensure no httpx.AsyncClient call in the patched files still uses verify=False.""" + + def _scan_file(self, path: str) -> list[str]: + violations = [] + with open(path) as f: + for i, line in enumerate(f, 1): + stripped = line.strip() + if "AsyncClient" in stripped and "verify=False" in stripped: + violations.append(f"{path}:{i}: {stripped}") + return violations + + def test_crawler_no_hardcoded_verify_false(self): + from backend.secuscan import crawler + + violations = self._scan_file(crawler.__file__) + assert violations == [], f"Hardcoded verify=False found: {violations}" + + def test_api_scanner_no_hardcoded_verify_false(self): + from backend.secuscan.scanners import api_scanner + + violations = self._scan_file(api_scanner.__file__) + assert violations == [], f"Hardcoded verify=False found: {violations}" + + def test_xss_scanner_no_hardcoded_verify_false(self): + from backend.secuscan.scanners import xss_validation_scanner + + violations = self._scan_file(xss_validation_scanner.__file__) + assert violations == [], f"Hardcoded verify=False found: {violations}" + + def test_network_vuln_scanner_no_hardcoded_verify_false(self): + from backend.secuscan.scanners import network_vulnerability_scanner + + violations = self._scan_file(network_vulnerability_scanner.__file__) + assert violations == [], f"Hardcoded verify=False found: {violations}"