From 4ea28d7e1c8d7168e5cabc8bdbd0c8f6331f2d0f Mon Sep 17 00:00:00 2001 From: Brian Krabach Date: Fri, 3 Apr 2026 21:47:35 -0700 Subject: [PATCH] feat: update serve() to accept and pass TLS SSL params to uvicorn - Add tls_cert and tls_key keyword arguments to serve() function - Resolve TLS params via CLI flag > settings.json > empty string default - Build ssl_kwargs dict conditionally based on file existence check - Pass ssl_certfile and ssl_keyfile to uvicorn.run() when TLS active - Print warning 'TLS not found, falling back to HTTP' when files missing - Print 'https://' URL when TLS active, 'http://' when not - Add --tls-cert and --tls-key flags to _add_serve_flags() and propagate to serve() - Add 5 new TLS tests: ssl params passed, no ssl default, fallback warning, https URL when active, http URL when inactive - Update existing serve()-call-signature tests to include tls_cert=None, tls_key=None Task: task-2-serve-ssl --- muxplex/cli.py | 41 +++++++- muxplex/tests/test_cli.py | 196 ++++++++++++++++++++++++++++++++++++-- 2 files changed, 225 insertions(+), 12 deletions(-) diff --git a/muxplex/cli.py b/muxplex/cli.py index b02f344..bc5733c 100644 --- a/muxplex/cli.py +++ b/muxplex/cli.py @@ -204,6 +204,8 @@ def serve( port: int | None = None, auth: str | None = None, session_ttl: int | None = None, + tls_cert: str | None = None, + tls_key: str | None = None, ) -> None: """Start the muxplex server. @@ -220,6 +222,8 @@ def serve( session_ttl = ( session_ttl if session_ttl is not None else settings.get("session_ttl", 604800) ) + tls_cert = tls_cert if tls_cert is not None else settings.get("tls_cert", "") + tls_key = tls_key if tls_key is not None else settings.get("tls_key", "") os.environ["MUXPLEX_PORT"] = str(port) os.environ["MUXPLEX_AUTH"] = auth @@ -230,8 +234,20 @@ def serve( from muxplex.main import app # noqa: PLC0415 - print(f" muxplex → http://{host}:{port}") - uvicorn.run(app, host=host, port=port, log_level="info") + # Resolve SSL configuration + ssl_kwargs: dict = {} + if tls_cert and tls_key: + cert_path = Path(tls_cert) + key_path = Path(tls_key) + missing = [str(p) for p in (cert_path, key_path) if not p.exists()] + if missing: + print(f" TLS {', '.join(missing)} not found, falling back to HTTP") + else: + ssl_kwargs = {"ssl_certfile": tls_cert, "ssl_keyfile": tls_key} + + scheme = "https" if ssl_kwargs else "http" + print(f" muxplex → {scheme}://{host}:{port}") + uvicorn.run(app, host=host, port=port, log_level="info", **ssl_kwargs) def doctor() -> None: @@ -666,7 +682,7 @@ def config_reset(key: str | None = None) -> None: def _add_serve_flags(parser: argparse.ArgumentParser) -> None: - """Add --host, --port, --auth, --session-ttl flags to a parser. + """Add --host, --port, --auth, --session-ttl, --tls-cert, --tls-key flags to a parser. All default to None so serve() can distinguish 'not passed' from 'passed the default value'. @@ -695,6 +711,18 @@ def _add_serve_flags(parser: argparse.ArgumentParser) -> None: dest="session_ttl", help="Session TTL in seconds (default: from settings.json, then 604800; 0 = browser session)", ) + parser.add_argument( + "--tls-cert", + default=None, + dest="tls_cert", + help="Path to TLS certificate file (default: from settings.json)", + ) + parser.add_argument( + "--tls-key", + default=None, + dest="tls_key", + help="Path to TLS private key file (default: from settings.json)", + ) def main() -> None: @@ -813,5 +841,10 @@ def main() -> None: else: _check_dependencies() serve( - host=args.host, port=args.port, auth=args.auth, session_ttl=args.session_ttl + host=args.host, + port=args.port, + auth=args.auth, + session_ttl=args.session_ttl, + tls_cert=args.tls_cert, + tls_key=args.tls_key, ) diff --git a/muxplex/tests/test_cli.py b/muxplex/tests/test_cli.py index 6b7b50d..739d116 100644 --- a/muxplex/tests/test_cli.py +++ b/muxplex/tests/test_cli.py @@ -21,7 +21,12 @@ def test_main_calls_serve_by_default(): with patch("sys.argv", ["muxplex"]): main() mock_serve.assert_called_once_with( - host=None, port=None, auth=None, session_ttl=None + host=None, + port=None, + auth=None, + session_ttl=None, + tls_cert=None, + tls_key=None, ) @@ -33,7 +38,12 @@ def test_main_passes_custom_host_and_port(): with patch("sys.argv", ["muxplex", "--host", "192.168.1.1", "--port", "9000"]): main() mock_serve.assert_called_once_with( - host="192.168.1.1", port=9000, auth=None, session_ttl=None + host="192.168.1.1", + port=9000, + auth=None, + session_ttl=None, + tls_cert=None, + tls_key=None, ) @@ -56,7 +66,12 @@ def test_main_passes_auth_flag(): with patch("sys.argv", ["muxplex", "--auth", "password"]): main() mock_serve.assert_called_once_with( - host=None, port=None, auth="password", session_ttl=None + host=None, + port=None, + auth="password", + session_ttl=None, + tls_cert=None, + tls_key=None, ) @@ -68,7 +83,12 @@ def test_main_passes_session_ttl_flag(): with patch("sys.argv", ["muxplex", "--session-ttl", "3600"]): main() mock_serve.assert_called_once_with( - host=None, port=None, auth=None, session_ttl=3600 + host=None, + port=None, + auth=None, + session_ttl=3600, + tls_cert=None, + tls_key=None, ) @@ -685,14 +705,19 @@ def test_serve_session_ttl_zero_is_valid(tmp_path, monkeypatch): def test_main_passes_none_for_unset_flags(): - """main() with no flags passes None for host/port/auth/session_ttl to serve().""" + """main() with no flags passes None for host/port/auth/session_ttl/tls_cert/tls_key to serve().""" from muxplex.cli import main with patch("muxplex.cli.serve") as mock_serve: with patch("sys.argv", ["muxplex"]): main() mock_serve.assert_called_once_with( - host=None, port=None, auth=None, session_ttl=None + host=None, + port=None, + auth=None, + session_ttl=None, + tls_cert=None, + tls_key=None, ) @@ -704,7 +729,12 @@ def test_main_passes_explicit_host_only(): with patch("sys.argv", ["muxplex", "--host", "10.0.0.1"]): main() mock_serve.assert_called_once_with( - host="10.0.0.1", port=None, auth=None, session_ttl=None + host="10.0.0.1", + port=None, + auth=None, + session_ttl=None, + tls_cert=None, + tls_key=None, ) @@ -718,7 +748,12 @@ def test_main_serve_subcommand_accepts_flags(): ): main() mock_serve.assert_called_once_with( - host="10.0.0.1", port=9000, auth=None, session_ttl=None + host="10.0.0.1", + port=9000, + auth=None, + session_ttl=None, + tls_cert=None, + tls_key=None, ) @@ -1227,3 +1262,148 @@ def test_upgrade_uses_service_module_install(monkeypatch, capsys): assert len(service_install_calls) > 0, ( "upgrade() must call muxplex.service.service_install() to regenerate the service file" ) + + +# --------------------------------------------------------------------------- +# task-2-serve-ssl: serve() TLS / SSL parameter tests +# --------------------------------------------------------------------------- + + +def test_serve_passes_ssl_params_to_uvicorn(tmp_path, monkeypatch): + """serve() with valid tls_cert and tls_key paths must pass ssl_certfile/ssl_keyfile to uvicorn.""" + import muxplex.cli as cli_mod + + # Create real cert/key files + cert_file = tmp_path / "server.crt" + key_file = tmp_path / "server.key" + cert_file.write_text("fake cert content") + key_file.write_text("fake key content") + + settings_file = tmp_path / "settings.json" + monkeypatch.setattr("muxplex.settings.SETTINGS_PATH", settings_file) + + uvicorn_calls = [] + + def fake_run(*args, **kwargs): + uvicorn_calls.append(kwargs) + + with patch("uvicorn.run", fake_run): + with patch.dict("sys.modules", {"muxplex.main": MagicMock()}): + cli_mod.serve(tls_cert=str(cert_file), tls_key=str(key_file)) + + assert len(uvicorn_calls) == 1 + kwargs = uvicorn_calls[0] + assert "ssl_certfile" in kwargs, ( + "uvicorn.run must receive ssl_certfile when TLS paths are set" + ) + assert "ssl_keyfile" in kwargs, ( + "uvicorn.run must receive ssl_keyfile when TLS paths are set" + ) + assert kwargs["ssl_certfile"] == str(cert_file) + assert kwargs["ssl_keyfile"] == str(key_file) + + +def test_serve_no_ssl_when_tls_paths_empty(tmp_path, monkeypatch): + """serve() with no TLS paths (default) must NOT pass ssl_certfile/ssl_keyfile to uvicorn.""" + import muxplex.cli as cli_mod + + settings_file = tmp_path / "settings.json" + monkeypatch.setattr("muxplex.settings.SETTINGS_PATH", settings_file) + + uvicorn_calls = [] + + def fake_run(*args, **kwargs): + uvicorn_calls.append(kwargs) + + with patch("uvicorn.run", fake_run): + with patch.dict("sys.modules", {"muxplex.main": MagicMock()}): + cli_mod.serve() # Default: tls_cert=None, tls_key=None + + assert len(uvicorn_calls) == 1 + kwargs = uvicorn_calls[0] + assert "ssl_certfile" not in kwargs, ( + "uvicorn.run must NOT receive ssl_certfile when no TLS" + ) + assert "ssl_keyfile" not in kwargs, ( + "uvicorn.run must NOT receive ssl_keyfile when no TLS" + ) + + +def test_serve_falls_back_to_http_when_cert_file_missing(tmp_path, monkeypatch, capsys): + """serve() prints a warning and skips SSL when tls_cert/tls_key paths don't exist on disk.""" + import muxplex.cli as cli_mod + + # Paths are set but the files do NOT exist + cert_file = tmp_path / "nonexistent.crt" + key_file = tmp_path / "nonexistent.key" + + settings_file = tmp_path / "settings.json" + monkeypatch.setattr("muxplex.settings.SETTINGS_PATH", settings_file) + + uvicorn_calls = [] + + def fake_run(*args, **kwargs): + uvicorn_calls.append(kwargs) + + with patch("uvicorn.run", fake_run): + with patch.dict("sys.modules", {"muxplex.main": MagicMock()}): + cli_mod.serve(tls_cert=str(cert_file), tls_key=str(key_file)) + + # Warning must be printed + captured = capsys.readouterr() + out_lower = captured.out.lower() + assert "not found" in out_lower or "falling back" in out_lower, ( + f"Must print warning about missing TLS files, got: {captured.out!r}" + ) + + # SSL must NOT be passed to uvicorn + assert len(uvicorn_calls) == 1 + kwargs = uvicorn_calls[0] + assert "ssl_certfile" not in kwargs, ( + "Must not pass ssl_certfile when cert file missing" + ) + assert "ssl_keyfile" not in kwargs, ( + "Must not pass ssl_keyfile when cert file missing" + ) + + +def test_serve_prints_https_url_when_tls_active(tmp_path, monkeypatch, capsys): + """serve() must print 'https://' URL when TLS is active.""" + import muxplex.cli as cli_mod + + cert_file = tmp_path / "server.crt" + key_file = tmp_path / "server.key" + cert_file.write_text("fake cert") + key_file.write_text("fake key") + + settings_file = tmp_path / "settings.json" + monkeypatch.setattr("muxplex.settings.SETTINGS_PATH", settings_file) + + with patch("uvicorn.run"): + with patch.dict("sys.modules", {"muxplex.main": MagicMock()}): + cli_mod.serve(tls_cert=str(cert_file), tls_key=str(key_file)) + + captured = capsys.readouterr() + assert "https://" in captured.out, ( + f"Must print 'https://' when TLS is active, got: {captured.out!r}" + ) + + +def test_serve_prints_http_url_when_no_tls(tmp_path, monkeypatch, capsys): + """serve() must print 'http://' URL when TLS is not configured.""" + import muxplex.cli as cli_mod + + settings_file = tmp_path / "settings.json" + monkeypatch.setattr("muxplex.settings.SETTINGS_PATH", settings_file) + + with patch("uvicorn.run"): + with patch.dict("sys.modules", {"muxplex.main": MagicMock()}): + cli_mod.serve() # No TLS + + captured = capsys.readouterr() + assert "http://" in captured.out, ( + f"Must print 'http://' when no TLS, got: {captured.out!r}" + ) + assert "https://" not in captured.out, ( + f"Must NOT print 'https://' when no TLS, got: {captured.out!r}" + )