From 46374d60813d1d9669082c9e34dfd9917023ef47 Mon Sep 17 00:00:00 2001
From: majiayu000 <1835304752@qq.com>
Date: Thu, 11 Dec 2025 14:01:29 +0800
Subject: [PATCH] Add --insecure flag to CLI for wss:// connections

Add an --insecure option to the CLI that allows connecting to
wss:// endpoints with self-signed or invalid SSL certificates.
This is useful for testing and development purposes.

When --insecure is specified, the CLI creates an SSL context that
disables certificate verification and hostname checking.

Fixes #1675
---
 src/websockets/cli.py | 21 +++++++++++++++++----
 tests/test_cli.py     | 26 +++++++++++++++++++++++++-
 2 files changed, 42 insertions(+), 5 deletions(-)

diff --git a/src/websockets/cli.py b/src/websockets/cli.py
index e084b62a..a83db277 100644
--- a/src/websockets/cli.py
+++ b/src/websockets/cli.py
@@ -3,8 +3,9 @@
 import argparse
 import asyncio
 import os
+import ssl
 import sys
-from typing import Generator
+from typing import Any, Generator
 
 from .asyncio.client import ClientConnection, connect
 from .asyncio.messages import SimpleQueue
@@ -101,9 +102,9 @@ async def send_outgoing_messages(
             break
 
 
-async def interactive_client(uri: str) -> None:
+async def interactive_client(uri: str, **kwargs: Any) -> None:
     try:
-        websocket = await connect(uri)
+        websocket = await connect(uri, **kwargs)
     except Exception as exc:
         print(f"Failed to connect to {uri}: {exc}.")
         sys.exit(1)
@@ -151,6 +152,11 @@ def main(argv: list[str] | None = None) -> None:
     group = parser.add_mutually_exclusive_group()
     group.add_argument("--version", action="store_true")
     group.add_argument("uri", metavar="<uri>", nargs="?")
+    parser.add_argument(
+        "--insecure",
+        action="store_true",
+        help="Disable SSL certificate verification for wss:// connections.",
+    )
     args = parser.parse_args(argv)
 
     if args.version:
@@ -171,8 +177,15 @@ def main(argv: list[str] | None = None) -> None:
     except ImportError:  # readline isn't available on all platforms
         pass
 
+    kwargs: dict[str, Any] = {}
+    if args.insecure:
+        ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        ssl_context.check_hostname = False
+        ssl_context.verify_mode = ssl.CERT_NONE
+        kwargs["ssl"] = ssl_context
+
     # Remove the try/except block when dropping Python < 3.11.
     try:
-        asyncio.run(interactive_client(args.uri))
+        asyncio.run(interactive_client(args.uri, **kwargs))
     except KeyboardInterrupt:  # pragma: no cover
         pass
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 391f7580..896c4288 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -11,6 +11,7 @@
 # Run a test server in a thread. This is easier than running an asyncio server
 # because we would have to run main() in a thread, due to using asyncio.run().
 from .sync.server import get_uri, run_server
+from .utils import SERVER_CONTEXT
 
 
 vt100_commands = re.compile(r"\x1b\[[A-Z]|\x1b[78]|\r")
@@ -109,4 +110,27 @@ def test_connection_failure(self):
 
     def test_no_args(self):
         output = self.run_main([], expected_exit_code=2)
-        self.assertEqual(output, "usage: websockets [--version | <uri>]\n")
+        self.assertEqual(output, "usage: websockets [--version] [--insecure] [<uri>]\n")
+
+    def test_insecure_connection(self):
+        def text_handler(websocket):
+            websocket.send("secure hello")
+
+        with run_server(text_handler, ssl=SERVER_CONTEXT) as server:
+            server_uri = get_uri(server)
+            output = self.run_main([server_uri, "--insecure"], "")
+        self.assertEqual(
+            remove_commands_and_prompts(output),
+            add_connection_messages("\n< secure hello\n", server_uri),
+        )
+
+    def test_insecure_connection_fails_without_flag(self):
+        def text_handler(websocket):
+            websocket.send("secure hello")
+
+        with run_server(text_handler, ssl=SERVER_CONTEXT) as server:
+            server_uri = get_uri(server)
+            output = self.run_main([server_uri], expected_exit_code=1)
+        self.assertTrue(
+            output.startswith(f"Failed to connect to {server_uri}: ")
+        )