Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions backend/secuscan/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,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
Expand Down
4 changes: 3 additions & 1 deletion backend/secuscan/crawler.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@

import httpx

from .config import settings


class _SurfaceParser(HTMLParser):
def __init__(self) -> None:
Expand Down Expand Up @@ -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)

Expand Down
3 changes: 2 additions & 1 deletion backend/secuscan/scanners/api_scanner.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@

import httpx

from ..config import settings
from .base import BaseScanner
from ..crawler import crawl_target

Expand Down Expand Up @@ -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("/"))
Expand Down
3 changes: 2 additions & 1 deletion backend/secuscan/scanners/network_vulnerability_scanner.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@

import httpx

from ..config import settings
from .base import BaseScanner
from ..knowledgebase import KnowledgeBase
from ..plugins import get_plugin_manager
Expand Down Expand Up @@ -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
Expand Down
3 changes: 2 additions & 1 deletion backend/secuscan/scanners/xss_validation_scanner.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@

import httpx

from ..config import settings
from .base import BaseScanner


Expand All @@ -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
Expand Down
304 changes: 304 additions & 0 deletions testing/backend/unit/test_tls_verification.py
Original file line number Diff line number Diff line change
@@ -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 = "<html></html>"
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 = "<html></html>"
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 = "<html><title>Test</title></html>"
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 = "<html><title>Test</title></html>"
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}"
Loading