Add --insecure flag to skip SSL cert verification. (#133)

* Add --insecure flag to skip SSL cert verification.

* Fix the insecure arg passed to upload in run_scan_inspect

* fix tests

Author: hemang-snyk

src/mcp_scan/MCPScanner.py

@@ -76,6 +76,7 @@ def __init__(
         verbose: bool = False,
         additional_headers: dict | None = None,
         control_servers: list | None = None,
+        insecure: bool = False,
         **kwargs: Any,
     ):
         logger.info("Initializing MCPScanner")
@@ -94,6 +95,7 @@ def __init__(
         self.include_built_in = include_built_in
         self.control_servers = control_servers
         self.verbose = verbose
+        self.insecure = insecure
         logger.debug(
             "MCPScanner initialized with timeout: %d, checks_per_server: %d", server_timeout, checks_per_server
         )
@@ -304,6 +306,7 @@ async def scan(self) -> list[ScanPathResult]:
             opt_out_of_identity=self.opt_out_of_identity,
             skip_pushing=bool(self.control_servers),
             verbose=self.verbose,
+            insecure=self.insecure,
         )
         logger.debug("Result verified: %s", result_verified)
         logger.debug("Saving storage file")

src/mcp_scan/cli.py

@@ -212,6 +212,12 @@ def add_common_arguments(parser):
         default=False,
         help="Output results in JSON format instead of rich text",
     )
+    parser.add_argument(
+        "--insecure",
+        default=False,
+        action="store_true",
+        help="Disable SSL certificate verification",
+    )
 
 
 def add_server_arguments(parser):
@@ -787,8 +793,9 @@ async def run_scan_inspect(mode="scan", args=None):
                 server_config["url"],
                 server_config["identifier"],
                 server_config["opt_out"],
-                verbose=hasattr(args, "verbose") and args.verbose,
+                verbose=getattr(args, "verbose", False),
                 additional_headers=parse_headers(server_config["headers"]),
+                insecure=getattr(args, "insecure", False),
             )
     return result
 

src/mcp_scan/upload.py

@@ -61,6 +61,7 @@ async def upload(
     verbose: bool = False,
     additional_headers: dict | None = None,
     max_retries: int = 3,
+    insecure: bool = False,
 ) -> None:
     """
     Upload the scan results to the control server with retry logic.
@@ -73,6 +74,7 @@ async def upload(
         verbose: Whether to enable verbose logging
         additional_headers: Additional HTTP headers to send
         max_retries: Maximum number of retry attempts (default: 3)
+        insecure: Whether to disable SSL certificate verification (default: False)
     """
     if not results:
         logger.info("No scan results to upload")
@@ -100,7 +102,10 @@ async def upload(
 
     for attempt in range(max_retries):
         try:
-            async with aiohttp.ClientSession(trace_configs=trace_configs, connector=setup_tcp_connector()) as session:
+            async with aiohttp.ClientSession(
+                trace_configs=trace_configs,
+                connector=setup_tcp_connector(insecure=insecure),
+            ) as session:
                 headers = {"Content-Type": "application/json"}
                 headers.update(additional_headers)
 

src/mcp_scan/verify_api.py

@@ -103,10 +103,17 @@ async def on_request_redirect(session, trace_config_ctx, params):
     return [trace_config]
 
 
-def setup_tcp_connector() -> aiohttp.TCPConnector:
+def setup_tcp_connector(insecure: bool = False) -> aiohttp.TCPConnector:
     """
-    Setup a TCP connector with a default SSL context and cleanup enabled.
+    Setup a TCP connector with SSL settings.
+
+    When insecure is True, disable SSL verification and hostname checking.
+    Otherwise, use a secure default SSL context with certifi CA and TLSv1.2+.
     """
+    if insecure:
+        # Disable SSL verification at the connector level
+        return aiohttp.TCPConnector(ssl=False, enable_cleanup_closed=True)
+
     ssl_context = ssl.create_default_context(cafile=certifi.where())
     ssl_context.minimum_version = ssl.TLSVersion.TLSv1_2
     connector = aiohttp.TCPConnector(ssl=ssl_context, enable_cleanup_closed=True)
@@ -145,6 +152,7 @@ async def analyze_machine(
     verbose: bool = False,
     skip_pushing: bool = False,
     max_retries: int = 3,
+    insecure: bool = False,
 ) -> list[ScanPathResult]:
     """
     Analyze the scan paths with the analysis server.
@@ -158,6 +166,7 @@ async def analyze_machine(
         verbose: Whether to enable verbose logging
         skip_pushing: Whether to skip pushing the scan to the platform
         max_retries: Maximum number of retry attempts
+        insecure: Whether to skip SSL verification
     """
     logger.debug(f"Analyzing scan path with URL: {analysis_url}")
     user_info = get_user_info(identifier=identifier, opt_out=opt_out_of_identity)
@@ -179,7 +188,10 @@ async def analyze_machine(
 
     for attempt in range(max_retries):
         try:
-            async with aiohttp.ClientSession(trace_configs=trace_configs, connector=setup_tcp_connector()) as session:
+            async with aiohttp.ClientSession(
+                trace_configs=trace_configs,
+                connector=setup_tcp_connector(insecure=insecure),
+            ) as session:
                 async with session.post(
                     analysis_url,
                     data=payload.model_dump_json(),

tests/unit/test_cli_parsing.py

@@ -407,3 +407,51 @@ async def test_no_upload_when_no_control_servers(self):
 
             # Verify upload was not called
             mock_upload.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_upload_with_insecure(self):
+        """Test that upload is called with insecure option."""
+        from argparse import Namespace
+
+        from mcp_scan.cli import run_scan_inspect
+
+        mock_result = ScanPathResult(path="/test/path")
+
+        with patch("mcp_scan.cli.MCPScanner") as MockScanner, patch("mcp_scan.cli.upload") as mock_upload:
+            # Setup scanner mock
+            mock_scanner_instance = AsyncMock()
+            mock_scanner_instance.scan = AsyncMock(return_value=[mock_result])
+            mock_scanner_instance.__aenter__ = AsyncMock(return_value=mock_scanner_instance)
+            mock_scanner_instance.__aexit__ = AsyncMock(return_value=None)
+            MockScanner.return_value = mock_scanner_instance
+
+            # Setup upload mock
+            mock_upload.return_value = None
+
+            # Create args with a control server and without the insecure option
+            args_without_insecure = Namespace(
+                verification_H=None,
+                control_servers=[{"url": "https://server1.com", "headers": [], "identifier": None, "opt_out": False}],
+            )
+
+            # Run the scan
+            await run_scan_inspect(mode="scan", args=args_without_insecure)
+
+            # Verify upload was called and insecure was not propagated
+            _, kwargs = mock_upload.call_args
+            assert kwargs.get("insecure") is False
+
+            # Create args with a control server and insecure option
+            args_with_insecure = Namespace(
+                verification_H=None,
+                control_servers=[{"url": "https://server1.com", "headers": [], "identifier": None, "opt_out": False}],
+                insecure=True,
+            )
+
+            # Run the scan
+            await run_scan_inspect(mode="scan", args=args_with_insecure)
+
+            # Verify upload was called and insecure was propagated
+            assert mock_upload.call_count == 2
+            _, kwargs = mock_upload.call_args
+            assert kwargs.get("insecure") is True