Add --insecure option to the CLI.

Fix #1675..

Author: aaugustin

docs/project/changelog.rst

@@ -56,6 +56,9 @@ New features
 
 * Validated compatibility with Python 3.15.
 
+* Added the ``--insecure`` option to the ``websockets`` CLI to disable TLS
+  certificate validation.
+
 .. _16.1:
 
 16.1
@@ -75,7 +78,7 @@ Improvements
 * Stripped ``Authorization``, ``Cookie``, and ``Proxy-Authorization`` headers
   when clients follow cross-origin redirects, a security hardening measure.
 
-* Escaped control characters in incoming messages in ``python -m websockets``.
+* Escaped control characters in incoming messages in the ``websockets`` CLI.
 
 * Blocked non-ASCII values in :class:`~datastructures.Headers`.
 
@@ -135,6 +138,11 @@ Bug fixes
 
 *March 5, 2025*
 
+Improvements
+............
+
+* Added an entry point for running the CLI as ``websockets``.
+
 Bug fixes
 .........
 

src/websockets/cli.py

@@ -4,8 +4,9 @@
 import asyncio
 import itertools
 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
@@ -123,9 +124,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)
@@ -175,6 +176,11 @@ def main(argv: list[str] | None = None) -> None:
         action="store_true",
         help="show usage and exit",
     )
+    parser.add_argument(
+        "--insecure",
+        action="store_true",
+        help="disable TLS certificate verification",
+    )
     parser.add_argument(
         "--version",
         action="store_true",
@@ -209,7 +215,13 @@ def main(argv: list[str] | None = None) -> None:
     except ImportError:  # readline isn't available on all platforms
         pass
 
+    kwargs = {}
+    if args.insecure and args.uri.startswith("wss://"):
+        # This isn't a public API but it's mentioned in the changelog:
+        # https://docs.python.org/3/whatsnew/3.4.html#changed-in-3-4-3
+        kwargs["ssl"] = ssl._create_unverified_context()
+
     try:
-        asyncio.run(interactive_client(args.uri))
+        asyncio.run(interactive_client(args.uri, **kwargs))
     except KeyboardInterrupt:  # pragma: no cover
         pass

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")
@@ -24,6 +25,10 @@ def add_connection_messages(output, server_uri):
     return f"Connected to {server_uri}.\n{output}Connection closed: 1000 (OK).\n"
 
 
+def echo_handler(websocket):
+    websocket.send(websocket.recv())
+
+
 class CLITests(unittest.TestCase):
     def run_main(self, argv, inputs="", close_input=False, expected_exit_code=None):
         # Replace sys.stdin with a file-like object backed by a file descriptor
@@ -93,9 +98,6 @@ def handler(websocket):
         )
 
     def test_send_message(self):
-        def echo_handler(websocket):
-            websocket.send(websocket.recv())
-
         with run_server(echo_handler) as server:
             server_uri = get_uri(server)
             output = self.run_main([server_uri], "hello\n")
@@ -117,6 +119,21 @@ def wait_handler(websocket):
             add_connection_messages("", server_uri),
         )
 
+    def test_insecure_connection_rejected(self):
+        with run_server(echo_handler, ssl=SERVER_CONTEXT) as server:
+            server_uri = get_uri(server, secure=True)
+            output = self.run_main([server_uri], "hello\n", expected_exit_code=1)
+        self.assertTrue(output.startswith(f"Failed to connect to {server_uri}: "))
+
+    def test_insecure_connection_accepted(self):
+        with run_server(echo_handler, ssl=SERVER_CONTEXT) as server:
+            server_uri = get_uri(server, secure=True)
+            output = self.run_main([server_uri, "--insecure"], "hello\n")
+        self.assertEqual(
+            remove_commands_and_prompts(output),
+            add_connection_messages("\n< hello\n", server_uri),
+        )
+
     def test_connection_failure(self):
         output = self.run_main(["ws://localhost:54321"], expected_exit_code=1)
         self.assertTrue(