Disable TLS certificate verification in crawler allows MITM/SSRF

.env.example

@@ -29,6 +29,7 @@ SECUSCAN_DOCKER_NETWORK=restricted
 SECUSCAN_SAFE_MODE_DEFAULT=true
 SECUSCAN_REQUIRE_CONSENT=true
 SECUSCAN_ALLOW_LOOPBACK_SCANS=true
+# SECUSCAN_TLS_VERIFY=true
 # SECUSCAN_ALLOWED_NETWORKS=127.0.0.1,192.168.*.*,10.*.*.*,172.16.*.*
 # SECUSCAN_CORS_ALLOWED_ORIGINS=http://127.0.0.1:5173,http://localhost:5173
 

backend/secuscan/config.py

@@ -40,6 +40,7 @@ class Settings(BaseSettings):
     # Security
     safe_mode_default: bool = True
     verify_ssl: bool = True
+    tls_verify: bool = True
     dns_resolution_timeout_seconds: float = 1.5
     dns_cache_ttl_seconds: int = 60
     dns_rebind_check: bool = True

backend/secuscan/crawler.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import logging
 from html.parser import HTMLParser
 import re
 from typing import Any, Dict, List
@@ -11,6 +12,8 @@
 
 from .config import settings
 
+logger = logging.getLogger(__name__)
+
 
 class _SurfaceParser(HTMLParser):
     def __init__(self) -> None:
@@ -85,36 +88,86 @@ async def crawl_target(
 ) -> Dict[str, Any]:
     """Fetch a target and normalize discovered links/forms/scripts/API hints."""
     headers = _build_headers(extra_headers)
+    tls_verify = getattr(settings, "tls_verify", True) and getattr(settings, "verify_ssl", True)
+
+    seed_parsed = urlparse(url)
+    seed_hostname = (seed_parsed.hostname or "").lower()
+
+    current_url = url
+    history: List[httpx.Response] = []
+    redirect_count = 0
+
     async with httpx.AsyncClient(
-        follow_redirects=True,
-        max_redirects=max_redirects,
+        follow_redirects=False,
         timeout=timeout,
         headers=headers,
         cookies=cookies or {},
-        verify=settings.verify_ssl,
+        verify=tls_verify,
     ) as client:
-        async with client.stream("GET", url) as response:
-            # Check Content-Length header if present
-            content_length = response.headers.get("content-length")
-            if content_length:
-                try:
-                    cl_val = int(content_length)
-                except ValueError:
-                    cl_val = 0
-                if cl_val > max_size:
-                    raise ValueError(f"Response size exceeds limit of {max_size} bytes")
-
-            # Read response in chunks to enforce size limit
-            body_chunks = []
-            bytes_read = 0
-            async for chunk in response.aiter_bytes():
-                bytes_read += len(chunk)
-                if bytes_read > max_size:
-                    raise ValueError(f"Response size exceeds limit of {max_size} bytes")
-                body_chunks.append(chunk)
-
-            body_bytes = b"".join(body_chunks)
-            body = body_bytes.decode("utf-8", errors="replace")
+        while True:
+            async with client.stream("GET", current_url) as response:
+                # Check Content-Length header if present
+                content_length = response.headers.get("content-length")
+                if content_length:
+                    try:
+                        cl_val = int(content_length)
+                    except ValueError:
+                        cl_val = 0
+                    if cl_val > max_size:
+                        raise ValueError(f"Response size exceeds limit of {max_size} bytes")
+
+                # Read response in chunks to enforce size limit
+                body_chunks = []
+                bytes_read = 0
+                async for chunk in response.aiter_bytes():
+                    bytes_read += len(chunk)
+                    if bytes_read > max_size:
+                        raise ValueError(f"Response size exceeds limit of {max_size} bytes")
+                    body_chunks.append(chunk)
+
+                body_bytes = b"".join(body_chunks)
+                body = body_bytes.decode("utf-8", errors="replace")
+
+            # Check if redirect and process same-host policy
+            if response.is_redirect and "location" in response.headers:
+                if redirect_count >= max_redirects:
+                    raise httpx.TooManyRedirects(
+                        f"Too many redirects (max_redirects={max_redirects})",
+                        request=httpx.Request("GET", current_url),
+                    )
+
+                location = response.headers["location"]
+                next_url = urljoin(current_url, location)
+                next_parsed = urlparse(next_url)
+                next_scheme = (next_parsed.scheme or "").lower()
+                next_hostname = (next_parsed.hostname or "").lower()
+
+                if next_scheme not in {"http", "https"} or not next_hostname:
+                    logger.warning(
+                        "Crawler redirect blocked due to unsupported/invalid Location: %s -> %s",
+                        current_url,
+                        next_url,
+                    )
+                    break
+
+                if seed_hostname and next_hostname != seed_hostname:
+                    logger.warning(
+                        "Crawler redirect blocked due to host mismatch: %s -> %s (seed host: %s, redirect host: %s)",
+                        current_url,
+                        next_url,
+                        seed_hostname,
+                        next_hostname,
+                    )
+                    break
+
+                history.append(response)
+                current_url = next_url
+                redirect_count += 1
+                continue
+
+            break
+
+    response.history = history
     parser = _SurfaceParser()
     parser.feed(body)
 

testing/backend/unit/test_crawler_security.py

@@ -0,0 +1,152 @@
+"""
+Unit tests for crawler TLS verification and redirect security policy (Issue #1747).
+"""
+
+from __future__ import annotations
+
+from unittest.mock import patch, MagicMock
+import pytest
+
+from backend.secuscan.config import settings
+from backend.secuscan.crawler import crawl_target
+
+
+class TestCrawlerTLSVerification:
+    def test_tls_verify_setting_exists(self):
+        assert hasattr(settings, "tls_verify")
+        assert settings.tls_verify is True
+
+    @pytest.mark.asyncio
+    async def test_crawl_target_uses_tls_verify_setting(self):
+        """crawl_target must construct AsyncClient with verify=tls_verify."""
+        mock_response = MagicMock()
+        mock_response.headers = {}
+        mock_response.is_redirect = False
+        mock_response.url = "http://example.com"
+        mock_response.status_code = 200
+
+        async def fake_aiter_bytes():
+            yield b"<html><head><title>Test</title></head><body>OK</body></html>"
+
+        mock_response.aiter_bytes = fake_aiter_bytes
+
+        class MockStreamContext:
+            async def __aenter__(self):
+                return mock_response
+            async def __aexit__(self, exc_type, exc_val, exc_tb):
+                pass
+
+        mock_client_instance = MagicMock()
+        mock_client_instance.stream.return_value = MockStreamContext()
+
+        class MockAsyncClient:
+            def __init__(self, **kwargs):
+                self.kwargs = kwargs
+            async def __aenter__(self):
+                return mock_client_instance
+            async def __aexit__(self, exc_type, exc_val, exc_tb):
+                pass
+
+        with patch("backend.secuscan.crawler.httpx.AsyncClient", side_effect=MockAsyncClient) as mock_client_cls:
+            res = await crawl_target("http://example.com")
+            assert mock_client_cls.call_count == 1
+            client_kwargs = mock_client_cls.call_args.kwargs
+            assert client_kwargs.get("verify") is True
+            assert client_kwargs.get("follow_redirects") is False
+
+
+class TestCrawlerRedirectPolicy:
+    @pytest.mark.asyncio
+    async def test_same_host_redirect_allowed(self):
+        """Redirects to the same hostname must be followed."""
+        resp1 = MagicMock()
+        resp1.headers = {"location": "/login"}
+        resp1.is_redirect = True
+        resp1.status_code = 302
+        resp1.url = "http://example.com/start"
+
+        async def fake_aiter_bytes1():
+            yield b"Redirecting..."
+        resp1.aiter_bytes = fake_aiter_bytes1
+
+        resp2 = MagicMock()
+        resp2.headers = {}
+        resp2.is_redirect = False
+        resp2.status_code = 200
+        resp2.url = "http://example.com/login"
+
+        async def fake_aiter_bytes2():
+            yield b"<html><head><title>Login Page</title></head><body>Form</body></html>"
+        resp2.aiter_bytes = fake_aiter_bytes2
+
+        responses = [resp1, resp2]
+        call_index = 0
+
+        class MockStreamContext:
+            def __init__(self, url):
+                nonlocal call_index
+                self.resp = responses[call_index]
+                call_index += 1
+            async def __aenter__(self):
+                return self.resp
+            async def __aexit__(self, exc_type, exc_val, exc_tb):
+                pass
+
+        mock_client_instance = MagicMock()
+        mock_client_instance.stream.side_effect = lambda method, url: MockStreamContext(url)
+
+        class MockAsyncClient:
+            def __init__(self, **kwargs):
+                pass
+            async def __aenter__(self):
+                return mock_client_instance
+            async def __aexit__(self, exc_type, exc_val, exc_tb):
+                pass
+
+        with patch("backend.secuscan.crawler.httpx.AsyncClient", side_effect=MockAsyncClient):
+            result = await crawl_target("http://example.com/start")
+            assert result["final_url"] == "http://example.com/login"
+            assert result["status_code"] == 200
+            assert len(result["redirect_chain"]) == 1
+            assert result["redirect_chain"][0]["url"] == "http://example.com/start"
+            assert result["redirect_chain"][0]["location"] == "/login"
+
+    @pytest.mark.asyncio
+    async def test_cross_host_redirect_blocked(self):
+        """Redirects to a different hostname (MITM / SSRF attempt) must be blocked."""
+        resp1 = MagicMock()
+        resp1.headers = {"location": "http://evil.com/phish"}
+        resp1.is_redirect = True
+        resp1.status_code = 302
+        resp1.url = "http://example.com/start"
+
+        async def fake_aiter_bytes1():
+            yield b"Redirecting..."
+        resp1.aiter_bytes = fake_aiter_bytes1
+
+        class MockStreamContext:
+            async def __aenter__(self):
+                return resp1
+            async def __aexit__(self, exc_type, exc_val, exc_tb):
+                pass
+
+        mock_client_instance = MagicMock()
+        mock_client_instance.stream.return_value = MockStreamContext()
+
+        class MockAsyncClient:
+            def __init__(self, **kwargs):
+                pass
+            async def __aenter__(self):
+                return mock_client_instance
+            async def __aexit__(self, exc_type, exc_val, exc_tb):
+                pass
+
+        with patch("backend.secuscan.crawler.httpx.AsyncClient", side_effect=MockAsyncClient):
+            with patch("backend.secuscan.crawler.logger.warning") as mock_warning:
+                result = await crawl_target("http://example.com/start")
+                # Cross-host redirect should be stopped immediately
+                assert result["final_url"] == "http://example.com/start"
+                assert result["status_code"] == 302
+                assert len(result["redirect_chain"]) == 0
+                mock_warning.assert_called_once()
+                assert "blocked due to host mismatch" in mock_warning.call_args[0][0]