Disable mandatory TLS verification with --insecure

README.md

@@ -1206,6 +1206,8 @@ cached file instead of plain text.
 Now use CA flags with other
 [plugin examples](#plugin-examples) to see them work with `https` traffic.
 
+To intercept TLS traffic from a server using a self-signed certificate add the `--insecure` flag to disable mandatory TLS certificate validation.
+
 ## TLS Interception With Docker
 
 Important notes about TLS Interception with Docker container:
@@ -2362,7 +2364,7 @@ usage: -m [-h] [--tunnel-hostname TUNNEL_HOSTNAME] [--tunnel-port TUNNEL_PORT]
  [--server-recvbuf-size SERVER_RECVBUF_SIZE]
  [--max-sendbuf-size MAX_SENDBUF_SIZE] [--timeout TIMEOUT]
  [--disable-http-proxy] [--disable-headers DISABLE_HEADERS]
- [--ca-key-file CA_KEY_FILE] [--ca-cert-dir CA_CERT_DIR]
+ [--ca-key-file CA_KEY_FILE] [--insecure] [--ca-cert-dir CA_CERT_DIR]
  [--ca-cert-file CA_CERT_FILE] [--ca-file CA_FILE]
  [--ca-signing-key-file CA_SIGNING_KEY_FILE]
  [--auth-plugin AUTH_PLUGIN] [--cache-requests]
@@ -2497,6 +2499,7 @@ options:
  Default: None. CA key to use for signing dynamically
  generated HTTPS certificates. If used, must also pass
  --ca-cert-file and --ca-signing-key-file
+ --insecure Default: False. Disables certificate verification
  --ca-cert-dir CA_CERT_DIR
  Default: ~/.proxy/certificates. Directory to store
  dynamically generated certificates. Also see --ca-key-
@@ -2523,9 +2526,8 @@ options:
  from responses. Extracted content type is written to
  the cache directory e.g. video.mp4.
  --cache-dir CACHE_DIR
- Default: /Users/abhinavsingh/.proxy/cache. Flag only
- applicable when cache plugin is used with on-disk
- storage.
+ Default: /home/kali/.proxy/cache. Flag only applicable
+ when cache plugin is used with on-disk storage.
  --proxy-pool PROXY_POOL
  List of upstream proxies to use in the pool
  --enable-web-server Default: False. Whether to enable

proxy/common/constants.py

@@ -90,6 +90,7 @@ def _env_threadless_compliant() -> bool:
 DEFAULT_MAX_SEND_SIZE = 64 * 1024
 DEFAULT_BUFFER_SIZE = 128 * 1024
 DEFAULT_CA_CERT_DIR = None
+DEFAULT_INSECURE = False
 DEFAULT_CA_CERT_FILE = None
 DEFAULT_CA_KEY_FILE = None
 DEFAULT_CA_SIGNING_KEY_FILE = None

proxy/common/utils.py

@@ -17,12 +17,15 @@
 import socket
 import logging
 import argparse
+import tempfile
 import functools
 import ipaddress
 import contextlib
 from types import TracebackType
 from typing import Any, Dict, List, Type, Tuple, Callable, Optional
 
+import _ssl # noqa: WPS436
+
 from .types import HostPort
 from .constants import (
  CRLF, COLON, HTTP_1_1, IS_WINDOWS, WHITESPACE, DEFAULT_TIMEOUT,
@@ -36,6 +39,18 @@
 logger = logging.getLogger(__name__)
 
 
+def cert_der_to_dict(certificate_der: Optional[bytes]) -> dict[str, Any]:
+ """Parse a DER formatted certificate to a python dict"""
+ if not certificate_der:
+ return {}
+ with tempfile.NamedTemporaryFile() as cert_file:
+ certificate_pem = ssl.DER_cert_to_PEM_cert(certificate_der)
+ cert_file.write(certificate_pem.encode())
+ cert_file.seek(0)
+ certificate = _ssl._test_decode_cert(cert_file.name)
+ return certificate or {}
+
+
 def tls_interception_enabled(flags: argparse.Namespace) -> bool:
  return flags.ca_key_file is not None and \
  flags.ca_cert_dir is not None and \

proxy/core/connection/server.py

@@ -56,7 +56,10 @@ def wrap(
  cafile=ca_file,
  )
  ctx.options |= ssl.OP_NO_SSLv2 | ssl.OP_NO_SSLv3 | ssl.OP_NO_TLSv1 | ssl.OP_NO_TLSv1_1
- ctx.check_hostname = hostname is not None
+ if verify_mode == ssl.VerifyMode.CERT_NONE:
+ ctx.check_hostname = False
+ else:
+ ctx.check_hostname = hostname is not None
  ctx.verify_mode = verify_mode
  self.connection.setblocking(True)
  self._conn = ctx.wrap_socket(

proxy/http/proxy/server.py

@@ -21,7 +21,7 @@
 import logging
 import threading
 import subprocess
-from typing import Any, Dict, List, Union, Optional, cast
+from typing import Any, Dict, List, Union, Optional
 
 from .plugin import HttpProxyBasePlugin
 from ..parser import HttpParser, httpParserTypes, httpParserStates
@@ -35,15 +35,16 @@
 from ...core.event import eventNames
 from ...common.flag import flags
 from ...common.types import Readables, Writables, Descriptors
-from ...common.utils import text_
+from ...common.utils import text_, cert_der_to_dict
 from ...core.connection import (
  TcpServerConnection, TcpConnectionUninitializedException,
 )
 from ...common.constants import (
- COMMA, DEFAULT_CA_FILE, PLUGIN_PROXY_AUTH, DEFAULT_CA_CERT_DIR,
- DEFAULT_CA_KEY_FILE, DEFAULT_CA_CERT_FILE, DEFAULT_DISABLE_HEADERS,
- PROXY_AGENT_HEADER_VALUE, DEFAULT_DISABLE_HTTP_PROXY,
- DEFAULT_CA_SIGNING_KEY_FILE, DEFAULT_HTTP_PROXY_ACCESS_LOG_FORMAT,
+ COMMA, DEFAULT_CA_FILE, DEFAULT_INSECURE, PLUGIN_PROXY_AUTH,
+ DEFAULT_CA_CERT_DIR, DEFAULT_CA_KEY_FILE, DEFAULT_CA_CERT_FILE,
+ DEFAULT_DISABLE_HEADERS, PROXY_AGENT_HEADER_VALUE,
+ DEFAULT_DISABLE_HTTP_PROXY, DEFAULT_CA_SIGNING_KEY_FILE,
+ DEFAULT_HTTP_PROXY_ACCESS_LOG_FORMAT,
  DEFAULT_HTTPS_PROXY_ACCESS_LOG_FORMAT,
 )
 
@@ -74,6 +75,13 @@
  'HTTPS certificates. If used, must also pass --ca-cert-file and --ca-signing-key-file',
 )
 
+flags.add_argument(
+ '--insecure',
+ action='store_true',
+ default=DEFAULT_INSECURE,
+ help='Default: False. Disables certificate verification',
+)
+
 flags.add_argument(
  '--ca-cert-dir',
  type=str,
@@ -760,10 +768,12 @@ def wrap_server(self) -> bool:
  assert isinstance(self.upstream.connection, socket.socket)
  do_close = False
  try:
+ verify_mode = ssl.VerifyMode.CERT_NONE if self.flags.insecure else ssl.VerifyMode.CERT_REQUIRED
  self.upstream.wrap(
  text_(self.request.host),
  self.flags.ca_file,
  as_non_blocking=True,
+ verify_mode=verify_mode,
  )
  except ssl.SSLCertVerificationError: # Server raised certificate verification error
  # When --disable-interception-on-ssl-cert-verification-error flag is on,
@@ -802,7 +812,7 @@ def wrap_client(self) -> bool:
  try:
  # TODO: Perform async certificate generation
  generated_cert = self.generate_upstream_certificate(
- cast(Dict[str, Any], self.upstream.connection.getpeercert()),
+ cert_der_to_dict(self.upstream.connection.getpeercert(True)),
  )
  self.client.wrap(self.flags.ca_signing_key_file, generated_cert)
  except subprocess.TimeoutExpired as e: # Popen communicate timeout

tests/certificates/test_cert_data.py

@@ -0,0 +1,13 @@
+# -*- coding: utf-8 -*-
+"""
+ proxy.py
+ ~~~~~~~~
+ ⚡⚡⚡ Fast, Lightweight, Pluggable, TLS interception capable proxy server focused on
+ Network monitoring, controls & Application development, testing, debugging.
+
+ :copyright: (c) 2013-present by Abhinav Singh and contributors.
+ :license: BSD, see LICENSE for more details.
+"""
+# flake8: noqa
+test_cert_bytes = b'0\x82\x03\xa30\x82\x02\x8b\xa0\x03\x02\x01\x02\x02\x14PE\x01\x8c\xa6\xea\xd8#\xcf\x90\xb0D\xc7\x04\xde\x9b9Y\xf3 0\r\x06\t*\x86H\x86\xf7\r\x01\x01\x0b\x05\x000a1\x0b0\t\x06\x03U\x04\x06\x13\x02as1\x0b0\t\x06\x03U\x04\x08\x0c\x02as1\x0b0\t\x06\x03U\x04\x07\x0c\x02as1\x0b0\t\x06\x03U\x04\n\x0c\x02as1\x0b0\t\x06\x03U\x04\x0b\x0c\x02as1\x0b0\t\x06\x03U\x04\x03\x0c\x02as1\x110\x0f\x06\t*\x86H\x86\xf7\r\x01\t\x01\x16\x02as0\x1e\x17\r240429125057Z\x17\r250429125057Z0a1\x0b0\t\x06\x03U\x04\x06\x13\x02as1\x0b0\t\x06\x03U\x04\x08\x0c\x02as1\x0b0\t\x06\x03U\x04\x07\x0c\x02as1\x0b0\t\x06\x03U\x04\n\x0c\x02as1\x0b0\t\x06\x03U\x04\x0b\x0c\x02as1\x0b0\t\x06\x03U\x04\x03\x0c\x02as1\x110\x0f\x06\t*\x86H\x86\xf7\r\x01\t\x01\x16\x02as0\x82\x01"0\r\x06\t*\x86H\x86\xf7\r\x01\x01\x01\x05\x00\x03\x82\x01\x0f\x000\x82\x01\n\x02\x82\x01\x01\x00\xee\xcbU\xe3\xc4]\x83\xb9\x9d\xb1(v0\x18\x18\xc3\x00\x96\xc0\x0f\xc29\x84\xe7/W\xc7\x0b\xec\xdf\x9d-\xec\xd9\x876\xe5m\xda\x96\xea\xb0\xc6\x00\x7f\xb6\x93;\xd6\x1bK`\xd4Hc<\xa0g\xe5Q[\xe3\xe1\xd1DD5\x9b\x12\xdf\xd0\xd0\xc6X\xc9\x98\xc9\xb1\x81\xf5\xa2\x12\xaa\xc1\xb0\x80\xe8)R\xa7\xed\xe3P6\x82\x05\xbcA4\x91\xbcs?\xc2\xf2\xfd-\xe65\'};\xa7E\xb2yN\x0fiO7\x82-`CX\xdb\xe0\x9c\xd7\x8e\x00N\nAu\xac/\xb3o\xcaG;\xa4\x8d\xca\x92\xe3F\x96\xe5\xbd\x1dq\xf6\xa5\x9f\xc5@I=\xfc\x1cl\x81\xb3y\x93FaPa^\x08\x0f\x80t\xb8J\xfd\xb8]\xd52\xf5\x9bE\xe8J:\x08\x8c\x98m0\xba\x85\x1b\xb6\x97\xe5\xba4\xe3nU\xa5\xc7\xeb\xde_z\x1a(j\xa7\xeb\x8a\xb4\xe1\'?\x91\x80MhG=y\xc7\xf1|\xcaJ@\xae\xc4\'\xd6\xd6}L\xf4\x91NV`\x98\x80\xef%\xa2hq\x05s\x02\x03\x01\x00\x01\xa3S0Q0\x1d\x06\x03U\x1d\x0e\x04\x16\x04\x14\xc6\xa4,\xe5\xe3\x15j\x18\x15@Xw!\xdd\xbf\xc6\xe5\xf0vG0\x1f\x06\x03U\x1d#\x04\x180\x16\x80\x14\xc6\xa4,\xe5\xe3\x15j\x18\x15@Xw!\xdd\xbf\xc6\xe5\xf0vG0\x0f\x06\x03U\x1d\x13\x01\x01\xff\x04\x050\x03\x01\x01\xff0\r\x06\t*\x86H\x86\xf7\r\x01\x01\x0b\x05\x00\x03\x82\x01\x01\x00\xacx\xeb\x02\x8a\xd3\x966\xb73\xfb\n\x1eb\ng\xda\x84\x18\x97P\xb4\x7f\x8a\xbd\x82\xf3\x1b\xe8k%\xcc\x0f\xbd\x7fB\xb9\x1df|-k\x01\xf3\x89\x08r\xb9\x93\xf5?Z\x16\xff\x0f\x97\x91b#\xef$I\x11\x9e\x16\xb2J\x97\xd1\x0e\xd6\xabD\xca@\xe7\xb3\xbe\x84S\x1e\xdb;\x9b\xc4\xf4\x18\xf4\x9a\x1b\xcej\xe0qmx\xe4N?K\n.p\xa8\xa6\xfa\xb0\xf7y\xe8\x0f\xbd\x0c216\xb0\xa1d\x1f\x7f3\xa1l?\xbe\x9a\x06\xed]\x1a\x00\xab\xb4e\x13:\x17\x1b\x88\x8e\xcaqp"\x8f\xa6\xf7\x06J?`\xe0\xf7\xce\xf8K\x08\x15\x18\xa1\xc4\xb5\xd9hB\xb0\xc6\\\xae?\xa9\x83FL\x8cm\xd1\xad^\xf0\xa5:\x8e\x97\x07\xd2\xd0l\x0e\x9d\x01\xa00c)\xae\xd0@\xefr\xe7,\xb7[\xd3H\xfe1\xfb\xa9|\xd0\xac\xc6i\x98\xe5\xd5\xd1\xf2\x97<\xf9\xe1?=\x93\xfaM\x86\xa2\x9dy\xdeZj\x93&\xa6\x84d\x07a\xbf\xd6\xdde\xaa)\t\xd6\x0e\x99\x85K'
+cert_dict = {'subject': ((('countryName', 'as'),), (('stateOrProvinceName', 'as'),), (('localityName', 'as'),), (('organizationName', 'as'),), (('organizationalUnitName', 'as'),), (('commonName', 'as'),), (('emailAddress', 'as'),)), 'issuer': ((('countryName', 'as'),), (('stateOrProvinceName', 'as'),), (('localityName', 'as'),), (('organizationName', 'as'),), (('organizationalUnitName', 'as'),), (('commonName', 'as'),), (('emailAddress', 'as'),)), 'version': 3, 'serialNumber': '5045018CA6EAD823CF90B044C704DE9B3959F320', 'notBefore': 'Apr 29 12:50:57 2024 GMT', 'notAfter': 'Apr 29 12:50:57 2025 GMT'}

tests/http/proxy/test_http_proxy_tls_interception.py

@@ -30,6 +30,7 @@
 from proxy.core.connection import TcpServerConnection
 from proxy.common.constants import DEFAULT_CA_FILE
 from ...test_assertions import Assertions
+from ...certificates.test_cert_data import test_cert_bytes
 
 
 class TestHttpProxyTlsInterception(Assertions):
@@ -56,9 +57,13 @@ async def test_e2e(self, mocker: MockerFixture) -> None:
  self.fileno = 10
  self.mock_socket_dup.side_effect = lambda fd: fd
 
+ def mock_cert(a: Any) -> Any:
+ return test_cert_bytes
+
  # Used for server side wrapping
  self.mock_ssl_context = mocker.patch('ssl.create_default_context')
  upstream_tls_sock = mock.MagicMock(spec=ssl.SSLSocket)
+ upstream_tls_sock.getpeercert = mock_cert
  self.mock_ssl_context.return_value.wrap_socket.return_value = upstream_tls_sock
 
  # Used for client wrapping
@@ -75,8 +80,8 @@ def mock_connection() -> Any:
 
  # Do not mock the original wrap method
  self.mock_server_conn.return_value.wrap.side_effect = \
- lambda x, y, as_non_blocking: TcpServerConnection.wrap(
- self.mock_server_conn.return_value, x, y, as_non_blocking=as_non_blocking,
+ lambda x, y, as_non_blocking, verify_mode: TcpServerConnection.wrap(
+ self.mock_server_conn.return_value, x, y, as_non_blocking=as_non_blocking, verify_mode=verify_mode,
  )
 
  type(self.mock_server_conn.return_value).connection = \

tests/plugin/test_http_proxy_plugins_with_tls_interception.py

@@ -29,6 +29,7 @@
 from proxy.core.connection import TcpServerConnection
 from .utils import get_plugin_by_test_name
 from ..test_assertions import Assertions
+from ..certificates.test_cert_data import test_cert_bytes
 
 
 class TestHttpProxyPluginExamplesWithTlsInterception(Assertions):
@@ -77,9 +78,12 @@ def _setUp(self, request: Any, mocker: MockerFixture) -> None:
  )
  self.protocol_handler.initialize()
 
- self.server = self.mock_server_conn.return_value
+ def mock_cert(a: Any) -> Any:
+ return test_cert_bytes
 
+ self.server = self.mock_server_conn.return_value
  self.server_ssl_connection = mocker.MagicMock(spec=ssl.SSLSocket)
+ self.server_ssl_connection.getpeercert = mock_cert
  self.mock_ssl_context.return_value.wrap_socket.return_value = self.server_ssl_connection
  self.client_ssl_connection = mocker.MagicMock(spec=ssl.SSLSocket)
  self.mock_ssl_wrap.return_value = self.client_ssl_connection
@@ -97,8 +101,8 @@ def mock_connection() -> Any:
 
  # Do not mock the original wrap method
  self.server.wrap.side_effect = \
- lambda x, y, as_non_blocking: TcpServerConnection.wrap(
- self.server, x, y, as_non_blocking=as_non_blocking,
+ lambda x, y, as_non_blocking, verify_mode: TcpServerConnection.wrap(
+ self.server, x, y, as_non_blocking=as_non_blocking, verify_mode=verify_mode,
  )
 
  self.server.has_buffer.side_effect = has_buffer