ken/muxplex
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: update serve() to accept and pass TLS SSL params to uvicorn

Browse Source 
- 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 <path> 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
This commit is contained in:
Brian Krabach
2026-04-03 21:47:35 -07:00
parent 89f39b2366
commit 4ea28d7e1c
2 changed files with 225 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
muxplex/cli.py
+37 -4
View File
@@ -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,
)
muxplex/tests/test_cli.py
+188 -8
View File
@@ -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}"
)