From 6d0a3e179fc9f5349b859c5cc17f1f0783791b8f Mon Sep 17 00:00:00 2001 From: hampuslinden <62944155+hampuslinden@users.noreply.github.com> Date: Sun, 14 Jun 2026 15:58:27 +0100 Subject: [PATCH] ci: add more pytest files and update cli to download root --- README.md | 12 + ssltui/__main__.py | 48 ++- tests/{test_smoke.py => test_1_smoke.py} | 0 .../{test_tui_init.py => test_2_tui_init.py} | 0 ...ecycle.py => test_3_tui_cert_lifecycle.py} | 0 tests/test_4_api_serve.py | 334 ++++++++++++++++++ tests/test_5_getroot.py | 57 +++ 7 files changed, 450 insertions(+), 1 deletion(-) rename tests/{test_smoke.py => test_1_smoke.py} (100%) rename tests/{test_tui_init.py => test_2_tui_init.py} (100%) rename tests/{test_tui_cert_lifecycle.py => test_3_tui_cert_lifecycle.py} (100%) create mode 100644 tests/test_4_api_serve.py create mode 100644 tests/test_5_getroot.py diff --git a/README.md b/README.md index b3e1c49..f467298 100644 --- a/README.md +++ b/README.md @@ -158,6 +158,18 @@ uv run ssltui renew --cert myapp.local uv run ssltui issue --cn api.local --san www.api.local --san 10.0.0.1 --days 365 ``` +### Get the root CA certificate + +Print the root CA certificate (PEM) to stdout — handy for piping into a trust +store. A short summary is written to stderr when run interactively, so the +redirected output stays a clean PEM: + +```bash +uv run ssltui getroot > local-ca.crt +# or write it directly +uv run ssltui getroot --out local-ca.crt +``` + ### Cron entry Install via the TUI's **Cron Schedule** option, or add manually: diff --git a/ssltui/__main__.py b/ssltui/__main__.py index 73b0001..8206044 100644 --- a/ssltui/__main__.py +++ b/ssltui/__main__.py @@ -91,10 +91,20 @@ def _build_parser() -> argparse.ArgumentParser: help="Write to FILE instead of stdout (recommended for keys)", ) + # getroot + getroot = sub.add_parser( + "getroot", help="Print the root CA certificate (PEM) to stdout" + ) + getroot.add_argument( + "--out", + metavar="FILE", + help="Write the CA cert to FILE instead of stdout", + ) + return p -_SUBCMDS = frozenset({"renew", "status", "issue", "serve", "get"}) +_SUBCMDS = frozenset({"renew", "status", "issue", "serve", "get", "getroot"}) def main(argv: list[str] | None = None) -> None: @@ -123,6 +133,8 @@ def main(argv: list[str] | None = None) -> None: _cmd_serve(args) elif args.cmd == "get": _cmd_get(args) + elif args.cmd == "getroot": + _cmd_getroot(args) else: # Default: interactive TUI from ssltui.tui import run_tui @@ -359,5 +371,39 @@ def _cmd_get(args) -> None: sys.stdout.buffer.write(data) +def _cmd_getroot(args) -> None: + from ssltui import config + from ssltui.ca import CAError, ca_expiry, ca_fingerprint, ca_subject + + root = config.data_dir() + ca_path = config.ca_cert_path(root) + if not ca_path.exists(): + print(f"ERROR: CA not initialised at {root}.", file=sys.stderr) + sys.exit(1) + + data = ca_path.read_bytes() + + if args.out: + out = Path(args.out) + out.write_bytes(data) + out.chmod(0o644) + print(f"Written to {out}") + return + + # When writing to a terminal, print a short summary to stderr so stdout + # stays a clean PEM that can be piped or redirected. + if sys.stdout.isatty(): + try: + print( + f"Root CA: {ca_subject(root)}\n" + f"Expires: {ca_expiry(root)}\n" + f"SHA256: {ca_fingerprint(root)}", + file=sys.stderr, + ) + except CAError: + pass + sys.stdout.buffer.write(data) + + if __name__ == "__main__": main() diff --git a/tests/test_smoke.py b/tests/test_1_smoke.py similarity index 100% rename from tests/test_smoke.py rename to tests/test_1_smoke.py diff --git a/tests/test_tui_init.py b/tests/test_2_tui_init.py similarity index 100% rename from tests/test_tui_init.py rename to tests/test_2_tui_init.py diff --git a/tests/test_tui_cert_lifecycle.py b/tests/test_3_tui_cert_lifecycle.py similarity index 100% rename from tests/test_tui_cert_lifecycle.py rename to tests/test_3_tui_cert_lifecycle.py diff --git a/tests/test_4_api_serve.py b/tests/test_4_api_serve.py new file mode 100644 index 0000000..3358608 --- /dev/null +++ b/tests/test_4_api_serve.py @@ -0,0 +1,334 @@ +"""End-to-end coverage of the REST API exposed by ``ssltui serve``. + +The server is started the way a user starts it -- ``ssltui serve`` running in a +tmux pane (the Textual "serve" UI) -- and shared across the tests in this module +via a module-scoped fixture. Each test drives the public REST API over HTTP with +the Bearer token written at CA init, and every download is compared byte-for-byte +against the corresponding file in the CA data directory. + +Endpoints covered: + GET /api/v1/certs list cert metadata + POST /api/v1/certs issue a cert + GET /api/v1/certs/ cert metadata + POST /api/v1/certs//renew renew a cert + GET /api/v1/certs//cert.pem download leaf cert + GET /api/v1/certs//key.pem download private key + GET /api/v1/certs//chain.pem download chain (leaf + CA) +""" + +from __future__ import annotations + +import importlib.util +import json +import os +import socket +import subprocess +import sys +import time +import urllib.error +import urllib.request +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from conftest import TmuxSession, launch_cmd, tmux_required +from ssltui import config, store +from ssltui.ca import init_ca, issue_cert + +flask_required = pytest.mark.skipif( + importlib.util.find_spec("flask") is None, + reason="requires the optional 'api' extra (flask)", +) + +pytestmark = [tmux_required, flask_required] + +# Baseline cert issued by the fixture; read-only tests operate on this CN. +BASE_CN = "api.local" +BASE_SANS = ["www.api.local", "10.0.0.5"] + + +def _free_port() -> int: + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +def _http( + method: str, url: str, token: str | None = None, body: dict | None = None +) -> tuple[int, bytes, dict[str, str]]: + """Make a request; return (status, raw body, headers) without raising. + + A ``token`` of ``None`` omits the Authorization header entirely. + """ + headers: dict[str, str] = {} + if token is not None: + headers["Authorization"] = f"Bearer {token}" + data = None + if body is not None: + headers["Content-Type"] = "application/json" + data = json.dumps(body).encode() + req = urllib.request.Request(url, data=data, method=method, headers=headers) + try: + with urllib.request.urlopen(req, timeout=10) as r: + return r.status, r.read(), dict(r.headers) + except urllib.error.HTTPError as e: + return e.code, e.read(), dict(e.headers) + + +def _poll(predicate, timeout: float = 20.0, interval: float = 0.25) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(interval) + return False + + +def _ca_files(data_dir: Path, cn: str) -> tuple[Path, Path, Path]: + """The leaf cert, key, and chain files the CA wrote for *cn*.""" + d = config.cert_dir(data_dir, cn) + return d / "cert.crt", d / "cert.key", d / "chain.crt" + + +@pytest.fixture(scope="module") +def server(tmp_path_factory: pytest.TempPathFactory): + """Init a CA, issue a baseline cert, and run ``ssltui serve`` over HTTP.""" + data_dir = tmp_path_factory.mktemp("ca-data") + init_ca(data_dir) + issue_cert(data_dir, cn=BASE_CN, sans=BASE_SANS, key_type="ec") + + port = _free_port() + base = f"http://127.0.0.1:{port}" + token = config.api_token_path(data_dir).read_text().strip() + + session = TmuxSession(f"ssltui_api_{os.getpid()}", "api") + session.start( + launch_cmd( + sys.executable, + data_dir, + extra=["serve", "--host", "127.0.0.1", "--port", str(port)], + ) + ) + + def _up() -> bool: + try: + urllib.request.urlopen(f"{base}/api/v1/certs", timeout=2) + return True + except urllib.error.HTTPError: + return True # 401 still proves the listener is accepting requests + except OSError: + # Covers URLError (connection refused while binding) and a bare + # TimeoutError during the window after listen() but before the + # accept loop starts serving. Keep polling. + return False + + try: + if not _poll(_up, timeout=30.0): + pytest.fail(f"server never came up; pane:\n{session.capture()}") + yield SimpleNamespace(base=base, token=token, data_dir=data_dir) + finally: + session.kill() + + +# --------------------------------------------------------------------------- # +# Authentication +# --------------------------------------------------------------------------- # + + +def test_missing_token_rejected(server) -> None: + status, _, _ = _http("GET", f"{server.base}/api/v1/certs", token=None) + assert status == 401 + + +def test_wrong_token_rejected(server) -> None: + status, _, _ = _http("GET", f"{server.base}/api/v1/certs", token="not-the-token") + assert status == 401 + + +def test_key_download_requires_auth(server) -> None: + # Private keys must never be served without a valid token (CLAUDE.md). + status, _, _ = _http( + "GET", f"{server.base}/api/v1/certs/{BASE_CN}/key.pem", token=None + ) + assert status == 401 + + +# --------------------------------------------------------------------------- # +# Listing & metadata +# --------------------------------------------------------------------------- # + + +def test_list_certs(server) -> None: + status, body, _ = _http("GET", f"{server.base}/api/v1/certs", server.token) + assert status == 200 + certs = json.loads(body) + assert isinstance(certs, list) + cns = {c["cn"] for c in certs} + # The API listing matches what the store reports on disk. + assert cns == {c["cn"] for c in store.list_certs(server.data_dir)} + assert BASE_CN in cns + + +def test_cert_metadata(server) -> None: + status, body, _ = _http( + "GET", f"{server.base}/api/v1/certs/{BASE_CN}", server.token + ) + assert status == 200 + meta = json.loads(body) + on_disk = store.get_cert(server.data_dir, BASE_CN) + assert on_disk is not None + assert meta["cn"] == BASE_CN + assert meta["serial"] == on_disk["serial"] + + +def test_cert_metadata_unknown_404(server) -> None: + status, _, _ = _http("GET", f"{server.base}/api/v1/certs/nope.local", server.token) + assert status == 404 + + +# --------------------------------------------------------------------------- # +# Downloads — every body is compared to the file in the CA directory +# --------------------------------------------------------------------------- # + + +def test_download_cert_matches_disk(server) -> None: + cert, _, _ = _ca_files(server.data_dir, BASE_CN) + status, body, headers = _http( + "GET", f"{server.base}/api/v1/certs/{BASE_CN}/cert.pem", server.token + ) + assert status == 200 + assert body == cert.read_bytes() + assert b"BEGIN CERTIFICATE" in body + # Served as a download with the expected filename. + assert "api.local.crt" in headers.get("Content-Disposition", "") + + +def test_download_key_matches_disk(server) -> None: + _, key, _ = _ca_files(server.data_dir, BASE_CN) + status, body, _ = _http( + "GET", f"{server.base}/api/v1/certs/{BASE_CN}/key.pem", server.token + ) + assert status == 200 + assert body == key.read_bytes() + assert b"PRIVATE KEY" in body + + +def test_download_chain_matches_disk(server) -> None: + _, _, chain = _ca_files(server.data_dir, BASE_CN) + status, body, _ = _http( + "GET", f"{server.base}/api/v1/certs/{BASE_CN}/chain.pem", server.token + ) + assert status == 200 + assert body == chain.read_bytes() + # The chain bundles the leaf followed by the CA cert. + assert body.count(b"BEGIN CERTIFICATE") >= 2 + assert config.ca_cert_path(server.data_dir).read_bytes() in body + + +def test_download_unknown_404(server) -> None: + for what in ("cert.pem", "key.pem", "chain.pem"): + status, _, _ = _http( + "GET", f"{server.base}/api/v1/certs/nope.local/{what}", server.token + ) + assert status == 404, what + + +# --------------------------------------------------------------------------- # +# Issue +# --------------------------------------------------------------------------- # + + +def test_issue_creates_cert_on_disk(server) -> None: + cn = "issued.local" + status, body, _ = _http( + "POST", + f"{server.base}/api/v1/certs", + server.token, + body={"cn": cn, "sans": ["alt.issued.local"], "key_type": "rsa"}, + ) + assert status == 201, body + meta = json.loads(body) + assert meta["cn"] == cn + assert meta["key_type"] == "rsa" + + cert, key, chain = _ca_files(server.data_dir, cn) + assert _poll(lambda: cert.exists() and key.exists() and chain.exists()) + + # Downloading the freshly issued cert returns exactly what is on disk. + status, dl_cert, _ = _http( + "GET", f"{server.base}/api/v1/certs/{cn}/cert.pem", server.token + ) + assert status == 200 + assert dl_cert == cert.read_bytes() + + status, dl_key, _ = _http( + "GET", f"{server.base}/api/v1/certs/{cn}/key.pem", server.token + ) + assert status == 200 + assert dl_key == key.read_bytes() + + +def test_issue_without_cn_rejected(server) -> None: + status, _, _ = _http( + "POST", f"{server.base}/api/v1/certs", server.token, body={"sans": ["x.local"]} + ) + assert status == 400 + + +# --------------------------------------------------------------------------- # +# Renew +# --------------------------------------------------------------------------- # + + +def test_renew_reissues_cert(server) -> None: + cn = "renew.local" + issue_cert(server.data_dir, cn=cn, key_type="ec") + before = store.get_cert(server.data_dir, cn) + assert before is not None + old_serial = before["serial"] + + status, body, _ = _http( + "POST", f"{server.base}/api/v1/certs/{cn}/renew", server.token + ) + assert status == 200, body + meta = json.loads(body) + assert meta["serial"] != old_serial + + # The renewed leaf on disk is what the API now serves. + cert, _, _ = _ca_files(server.data_dir, cn) + status, dl_cert, _ = _http( + "GET", f"{server.base}/api/v1/certs/{cn}/cert.pem", server.token + ) + assert status == 200 + assert dl_cert == cert.read_bytes() + assert store.get_cert(server.data_dir, cn)["serial"] == meta["serial"] + + +def test_renew_unknown_404(server) -> None: + status, _, _ = _http( + "POST", f"{server.base}/api/v1/certs/nope.local/renew", server.token + ) + assert status == 404 + + +# --------------------------------------------------------------------------- # +# Sanity: a downloaded cert is a real X.509 cert openssl can parse +# --------------------------------------------------------------------------- # + + +def test_downloaded_cert_is_valid_x509(server, tmp_path: Path) -> None: + status, body, _ = _http( + "GET", f"{server.base}/api/v1/certs/{BASE_CN}/cert.pem", server.token + ) + assert status == 200 + out_file = tmp_path / "downloaded.crt" + out_file.write_bytes(body) + out = subprocess.run( + ["openssl", "x509", "-in", str(out_file), "-noout", "-subject"], + capture_output=True, + text=True, + check=False, + ) + assert out.returncode == 0, out.stderr + assert BASE_CN in out.stdout diff --git a/tests/test_5_getroot.py b/tests/test_5_getroot.py new file mode 100644 index 0000000..2adaebb --- /dev/null +++ b/tests/test_5_getroot.py @@ -0,0 +1,57 @@ +"""CLI `getroot` command: prints / saves the root CA certificate.""" + +from __future__ import annotations + +import os +import subprocess +import sys +from pathlib import Path + +from ssltui import config +from ssltui.ca import init_ca + + +def _run(args: list[str], data_dir: Path) -> subprocess.CompletedProcess[str]: + env = {**os.environ, "SSLTUI_DIR": str(data_dir)} + return subprocess.run( + [sys.executable, "-m", "ssltui", *args], + capture_output=True, + text=True, + timeout=30, + env=env, + ) + + +def test_getroot_prints_ca_pem(tmp_path: Path) -> None: + data_dir = tmp_path / "ca" + init_ca(data_dir) + + result = _run(["getroot"], data_dir) + assert result.returncode == 0, result.stderr + # stdout is a clean PEM matching the CA cert on disk. + expected = config.ca_cert_path(data_dir).read_text() + assert result.stdout == expected + assert result.stdout.startswith("-----BEGIN CERTIFICATE-----") + + +def test_getroot_out_writes_file(tmp_path: Path) -> None: + data_dir = tmp_path / "ca" + init_ca(data_dir) + out = tmp_path / "exported.crt" + + result = _run(["getroot", "--out", str(out)], data_dir) + assert result.returncode == 0, result.stderr + assert out.exists() + assert out.read_bytes() == config.ca_cert_path(data_dir).read_bytes() + # Cert files are world-readable per the project's permission policy. + assert (out.stat().st_mode & 0o777) == 0o644 + # Nothing is printed to stdout in --out mode (only a status line). + assert "BEGIN CERTIFICATE" not in result.stdout + + +def test_getroot_uninitialised_errors(tmp_path: Path) -> None: + # No CA created in this dir. + result = _run(["getroot"], tmp_path / "empty") + assert result.returncode == 1 + assert "not initialised" in result.stderr.lower() + assert result.stdout == ""