From 4798ce3f18b7022be2f01591754cf91df30c4f2e Mon Sep 17 00:00:00 2001 From: Vlad Doloman Date: Wed, 8 Jul 2026 15:46:10 +0300 Subject: [PATCH] fix TUI crash on serial consoles; make index.txt the sole email store MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit curses.use_default_colors() raises on terminals (e.g. serial consoles using TERM=linux/vt100) whose terminfo lacks default-color support; init_colors() now falls back to an explicit black background and caps the "disabled" color to 8-color terminals. --reissue without --email silently sent no mail when the CN wasn't in openvpncertupdate-metadata.json, even though its email was already embedded in the cert's index.txt subject (via EASYRSA_REQ_EMAIL) and the TUI already fell back to it. Since build_client_full() always writes that subject field whenever an email is known, metadata.json was a redundant, driftable copy — remove it and have get_email() read index.txt directly, picking the last non-revoked entry for a CN since index.txt can contain revoked/duplicate lines for the same CN. Co-Authored-By: Claude Sonnet 5 --- CLAUDE.md | 5 ++- openvpncertupdate.py | 80 ++++++++++++++++++++---------------------- tests/test_cli.py | 22 ++++++------ tests/test_metadata.py | 16 --------- tests/test_pki.py | 43 ++++++++++++++++++++++- tests/test_widgets.py | 35 ++++++++++++++++-- 6 files changed, 127 insertions(+), 74 deletions(-) delete mode 100644 tests/test_metadata.py diff --git a/CLAUDE.md b/CLAUDE.md index 190bb1e..8d9acc8 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -47,10 +47,9 @@ python3 -m pytest tests/test_pki.py::test_sorted_ascending -v # single test |---|---| | SETTINGS | all-caps constants | | SETTINGS OVERRIDE | `ConfigError`, `load_settings_overrides()`, `_OVERRIDABLE_SETTINGS` | -| PKI | `CertInfo`, `load_expiring_certs()`, `load_all_certs()`, `_parse_index_line()` | +| PKI | `CertInfo`, `load_expiring_certs()`, `load_all_certs()`, `_parse_index_line()`, `get_email()` | | PASSWORD | `generate_password()` | | EASYRSA | `EasyRSAError`, `revoke_issued()`, `build_client_full()`, `gen_crl()`, `copy_crl()`, `is_ca_key_encrypted()`, `resolve_ca_passphrase()` | -| METADATA | `load_metadata()`, `save_email()`, `get_email()` | | CONFIG | `build_ovpn()` → `vpn-configs/__/CONFIG_NAME` | | CRYPTGEON | `CryptgeonError`, `create_note()` | | MAILER | `build_mime_message()`, `send_email()` | @@ -88,4 +87,4 @@ python3 -m pytest tests/test_pki.py::test_sorted_ascending -v # single test - `copy_crl()` does `chmod 644` after copy - Password: pos 1=uppercase, pos 2=lowercase (no j), pos 3-27=alphanumeric, pos 28=lowercase (no j); `oO01lIQ5S2Z8B` banned everywhere - Inline file path: `/inline/private/.inline` -- User emails stored in `/openvpncertupdate-metadata.json` +- User emails are not stored separately: `build_client_full()` sets `EASYRSA_REQ_EMAIL` whenever an email is known, which EasyRSA embeds as `emailAddress=` in the cert subject — so it round-trips through `/index.txt` itself. `get_email()` reads it back from there (`load_all_certs()` + CN match); there is no `openvpncertupdate-metadata.json` diff --git a/openvpncertupdate.py b/openvpncertupdate.py index 9c78190..316c48c 100644 --- a/openvpncertupdate.py +++ b/openvpncertupdate.py @@ -212,6 +212,26 @@ def load_all_certs(pki_dir: str) -> list[CertInfo]: return results +def get_email(pki_dir: str, cn: str) -> str: + """Return the emailAddress from CN's most recent V-status index.txt entry. + + index.txt is append-only, so a CN can have several V-status lines (e.g. + issued directly with easyrsa, bypassing this tool's revoke-before-build + renewal flow) as well as older R-status (revoked) ones; scan the whole + file in order and keep the last non-revoked match, or '' if none. + """ + found = "" + with open(os.path.join(pki_dir, "index.txt")) as fh: + for line in fh: + parsed = _parse_index_line(line) + if parsed is None: + continue + line_cn, _expiry, email = parsed + if line_cn == cn: + found = email + return found + + # ============================================================ # === PASSWORD === # ============================================================ @@ -333,32 +353,6 @@ def resolve_ca_passphrase( return configured -# ============================================================ -# === METADATA === -# ============================================================ - -_METADATA_FILE = "openvpncertupdate-metadata.json" - - -def load_metadata(pki_dir: str) -> dict[str, str]: - p = os.path.join(pki_dir, _METADATA_FILE) - if not os.path.exists(p): - return {} - with open(p) as fh: - return json.load(fh) - - -def save_email(pki_dir: str, cn: str, email_addr: str) -> None: - data = load_metadata(pki_dir) - data[cn] = email_addr - with open(os.path.join(pki_dir, _METADATA_FILE), "w") as fh: - json.dump(data, fh, indent=2) - - -def get_email(pki_dir: str, cn: str) -> str: - return load_metadata(pki_dir).get(cn, "") - - # ============================================================ # === CONFIG === # ============================================================ @@ -539,13 +533,20 @@ COLOR_SUCCESS = 6 def init_colors() -> None: curses.start_color() - curses.use_default_colors() - curses.init_pair(COLOR_NORMAL, curses.COLOR_WHITE, -1) + try: + curses.use_default_colors() + bg = -1 + except curses.error: + # Terminal (e.g. a serial console) lacks default-color support + # (terminfo has no "op"/"AX" capability); fall back to explicit black. + bg = curses.COLOR_BLACK + disabled_fg = 8 if curses.COLORS > 8 else curses.COLOR_WHITE + curses.init_pair(COLOR_NORMAL, curses.COLOR_WHITE, bg) curses.init_pair(COLOR_SELECTED, curses.COLOR_BLACK, curses.COLOR_CYAN) - curses.init_pair(COLOR_DISABLED, 8, -1) - curses.init_pair(COLOR_TITLE, curses.COLOR_WHITE, -1) - curses.init_pair(COLOR_ERROR, curses.COLOR_RED, -1) - curses.init_pair(COLOR_SUCCESS, curses.COLOR_GREEN, -1) + curses.init_pair(COLOR_DISABLED, disabled_fg, bg) + curses.init_pair(COLOR_TITLE, curses.COLOR_WHITE, bg) + curses.init_pair(COLOR_ERROR, curses.COLOR_RED, bg) + curses.init_pair(COLOR_SUCCESS, curses.COLOR_GREEN, bg) def draw_box(win, title: str = "") -> None: @@ -1055,11 +1056,8 @@ class CursesApp: elif result.action == Action.NEW_CERT: self._process_cert(stdscr, cn="", email="", is_renewal=False) elif result.action == Action.RENEW_SELECTED: - cert_by_cn = {c.cn: c for c in certs} for cn in result.selected_cns: email_addr = get_email(EASYRSA_PKI_DIR, cn) - if not email_addr and cn in cert_by_cn: - email_addr = cert_by_cn[cn].email if not self._process_cert(stdscr, cn=cn, email=email_addr, is_renewal=True): break @@ -1101,9 +1099,6 @@ class CursesApp: self._error(stdscr, f"EasyRSA error:\n{exc}") return True - if form.email: - save_email(EASYRSA_PKI_DIR, final_cn, form.email) - try: ovpn_path = build_ovpn(final_cn, EASYRSA_PKI_DIR, OVPN_TEMPLATE_PATH, VPN_CONFIGS_DIR, CONFIG_NAME) @@ -1259,7 +1254,8 @@ class CliRunner: def reissue(self, cn: str, email_addr: str, send_email_flag: bool = True, show_eml: bool = False) -> None: - # Prefer the caller-supplied email; fall back to stored metadata. + # Prefer the caller-supplied email; fall back to the emailAddress + # already on the existing cert's subject in index.txt. final_email = email_addr or get_email(EASYRSA_PKI_DIR, cn) self._issue(cn, final_email, is_renewal=True, send_email_flag=send_email_flag, show_eml=show_eml) @@ -1313,9 +1309,6 @@ class CliRunner: print(f"error: {exc}", file=sys.stderr) sys.exit(1) - if email_addr: - save_email(EASYRSA_PKI_DIR, cn, email_addr) - try: ovpn_path = build_ovpn(cn, EASYRSA_PKI_DIR, OVPN_TEMPLATE_PATH, VPN_CONFIGS_DIR, CONFIG_NAME) @@ -1357,6 +1350,9 @@ class CliRunner: print(f"Email sent to {email_addr}", file=sys.stderr) except RuntimeError as exc: print(f"warning: email failed: {exc}", file=sys.stderr) + elif send_email_flag and not email_addr: + print(f"warning: email skipped (no email address on file for {cn})", + file=sys.stderr) print(f"config: {ovpn_path}") if one_time_url: diff --git a/tests/test_cli.py b/tests/test_cli.py index 391e717..1a8eaac 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -15,7 +15,6 @@ def _patch_issue(monkeypatch, ovpn_path="/out/cn_2026-01-01/client.ovpn", """Patch all side-effectful callables used by CliRunner._issue.""" monkeypatch.setattr("openvpncertupdate.revoke_issued", MagicMock()) monkeypatch.setattr("openvpncertupdate.build_client_full", MagicMock()) - monkeypatch.setattr("openvpncertupdate.save_email", MagicMock()) monkeypatch.setattr("openvpncertupdate.build_ovpn", MagicMock(return_value=ovpn_path)) monkeypatch.setattr("openvpncertupdate.create_note", MagicMock(return_value=one_time_url)) monkeypatch.setattr("openvpncertupdate.send_email", MagicMock()) @@ -23,7 +22,6 @@ def _patch_issue(monkeypatch, ovpn_path="/out/cn_2026-01-01/client.ovpn", return { "revoke_issued": sys.modules["openvpncertupdate"].revoke_issued, "build_client_full": sys.modules["openvpncertupdate"].build_client_full, - "save_email": sys.modules["openvpncertupdate"].save_email, "build_ovpn": sys.modules["openvpncertupdate"].build_ovpn, "create_note": sys.modules["openvpncertupdate"].create_note, "send_email": sys.modules["openvpncertupdate"].send_email, @@ -44,12 +42,13 @@ def test_create_calls_build_not_revoke(monkeypatch, capsys): assert mocks["build_client_full"].call_args.args[2] == "alice" -def test_create_saves_email(monkeypatch, capsys): +def test_create_passes_email_to_build_client_full(monkeypatch, capsys): + # build-client-full embeds email into the cert subject (EASYRSA_REQ_EMAIL), + # which is what index.txt / get_email() reads back later — there's no + # separate email store to save to. mocks = _patch_issue(monkeypatch) CliRunner().create("alice", "alice@example.com") - mocks["save_email"].assert_called_once() - assert mocks["save_email"].call_args.args[1] == "alice" - assert mocks["save_email"].call_args.args[2] == "alice@example.com" + assert mocks["build_client_full"].call_args.kwargs["email"] == "alice@example.com" def test_create_sends_email_and_prints_url(monkeypatch, capsys): @@ -88,13 +87,15 @@ def test_reissue_calls_revoke_then_build(monkeypatch, capsys): assert mocks["build_client_full"].call_args.args[2] == "bob" -def test_reissue_falls_back_to_stored_email(monkeypatch, capsys): +def test_reissue_falls_back_to_index_txt_email(monkeypatch, capsys): + # get_email() reads straight from index.txt; this stubs that lookup + # rather than the file itself (index.txt parsing is covered in test_pki.py). mocks = _patch_issue(monkeypatch) monkeypatch.setattr("openvpncertupdate.get_email", - MagicMock(return_value="stored@example.com")) + MagicMock(return_value="fromindex@example.com")) CliRunner().reissue("bob", "") # no --email supplied mocks["send_email"].assert_called_once() - assert mocks["send_email"].call_args.args[0] == "stored@example.com" + assert mocks["send_email"].call_args.args[0] == "fromindex@example.com" def test_reissue_no_email_no_send(monkeypatch, capsys): @@ -102,8 +103,9 @@ def test_reissue_no_email_no_send(monkeypatch, capsys): monkeypatch.setattr("openvpncertupdate.get_email", MagicMock(return_value="")) CliRunner().reissue("bob", "") mocks["send_email"].assert_not_called() - out = capsys.readouterr().out + out, err = capsys.readouterr() assert "config:" in out + assert "warning: email skipped (no email address on file for bob)" in err # --------------------------------------------------------------------------- diff --git a/tests/test_metadata.py b/tests/test_metadata.py deleted file mode 100644 index 6925eb9..0000000 --- a/tests/test_metadata.py +++ /dev/null @@ -1,16 +0,0 @@ -from openvpncertupdate import load_metadata, save_email, get_email - -def test_load_empty(tmp_path): - assert load_metadata(str(tmp_path)) == {} - -def test_save_and_get(tmp_path): - save_email(str(tmp_path), "alice", "alice@example.com") - assert get_email(str(tmp_path), "alice") == "alice@example.com" - -def test_update_existing(tmp_path): - save_email(str(tmp_path), "alice", "old@example.com") - save_email(str(tmp_path), "alice", "new@example.com") - assert get_email(str(tmp_path), "alice") == "new@example.com" - -def test_missing_returns_empty(tmp_path): - assert get_email(str(tmp_path), "nobody") == "" diff --git a/tests/test_pki.py b/tests/test_pki.py index 6e37dba..8c66d45 100644 --- a/tests/test_pki.py +++ b/tests/test_pki.py @@ -1,7 +1,7 @@ import textwrap import pytest from datetime import datetime, timezone, timedelta -from openvpncertupdate import load_all_certs, load_expiring_certs, CertInfo, _parse_index_line +from openvpncertupdate import load_all_certs, load_expiring_certs, CertInfo, _parse_index_line, get_email def _fmt(dt: datetime) -> str: @@ -107,3 +107,44 @@ def test_load_all_certs_sorted_ascending(tmp_path): certs = load_all_certs(pki) dates = [c.expires for c in certs] assert dates == sorted(dates) + + +def test_get_email_reads_from_index_txt(tmp_path): + now = datetime.now(tz=timezone.utc) + pki = tmp_path / "pki" + pki.mkdir() + expiry = now + timedelta(days=5) + line = f"V\t{expiry.strftime('%y%m%d%H%M%S')}Z\t\t01\tunknown\t/CN=carol/emailAddress=carol@example.com\n" + (pki / "index.txt").write_text(line) + assert get_email(str(pki), "carol") == "carol@example.com" + + +def test_get_email_missing_cn_returns_empty(tmp_path): + now = datetime.now(tz=timezone.utc) + pki = make_pki(tmp_path, now) + assert get_email(pki, "nobody") == "" + + +def test_get_email_ignores_revoked_entry(tmp_path): + now = datetime.now(tz=timezone.utc) + pki = make_pki(tmp_path, now) + # "revoked" CN in make_pki() has status R, not V + assert get_email(pki, "revoked") == "" + + +def test_get_email_picks_last_valid_entry_among_duplicates(tmp_path): + # index.txt is append-only: a CN can have several lines (old ones R, + # or multiple V if a cert was reissued without going through this + # tool's revoke-first flow). The most recently appended V line wins. + now = datetime.now(tz=timezone.utc) + pki = tmp_path / "pki" + pki.mkdir() + e1 = (now + timedelta(days=5)).strftime("%y%m%d%H%M%S") + "Z" + e2 = (now + timedelta(days=365)).strftime("%y%m%d%H%M%S") + "Z" + content = ( + f"R\t{e1}\t230101Z,superseded\t01\tunknown\t/CN=dave/emailAddress=old@example.com\n" + f"V\t{e1}\t\t02\tunknown\t/CN=dave/emailAddress=older-valid@example.com\n" + f"V\t{e2}\t\t03\tunknown\t/CN=dave/emailAddress=newest@example.com\n" + ) + (pki / "index.txt").write_text(content) + assert get_email(str(pki), "dave") == "newest@example.com" diff --git a/tests/test_widgets.py b/tests/test_widgets.py index a429ce0..bdeff86 100644 --- a/tests/test_widgets.py +++ b/tests/test_widgets.py @@ -1,4 +1,4 @@ -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch import curses as _curses # Stub just enough of the curses API for InputField to import @@ -12,7 +12,7 @@ _curses.KEY_HOME = 262 _curses.KEY_END = 360 _curses.KEY_DC = 330 -from openvpncertupdate import InputField, clamp +from openvpncertupdate import InputField, clamp, init_colors, COLOR_NORMAL, COLOR_DISABLED def _field(initial="", mask=False): @@ -51,3 +51,34 @@ def test_clamp_bounds(): assert clamp(5, 0, 10) == 5 assert clamp(-1, 0, 10) == 0 assert clamp(11, 0, 10) == 10 + + +def _init_colors_calls(*, use_default_colors_raises, colors): + calls = [] + use_default = MagicMock( + side_effect=_curses.error("use_default_colors() returned ERR") + if use_default_colors_raises else None + ) + with patch.object(_curses, "start_color"), \ + patch.object(_curses, "use_default_colors", use_default), \ + patch.object(_curses, "init_pair", side_effect=lambda *a: calls.append(a)), \ + patch.object(_curses, "COLORS", colors, create=True): + init_colors() + return calls + +def test_init_colors_uses_default_colors_when_supported(): + calls = _init_colors_calls(use_default_colors_raises=False, colors=256) + normal = next(c for c in calls if c[0] == COLOR_NORMAL) + disabled = next(c for c in calls if c[0] == COLOR_DISABLED) + assert normal[2] == -1 + assert disabled[1:] == (8, -1) + +def test_init_colors_falls_back_on_serial_console(): + # Serial consoles (e.g. TERM=linux/vt100) often lack the terminfo + # "op"/"AX" capability, so use_default_colors() raises curses.error + # and only 8 colors are available. + calls = _init_colors_calls(use_default_colors_raises=True, colors=8) + normal = next(c for c in calls if c[0] == COLOR_NORMAL) + disabled = next(c for c in calls if c[0] == COLOR_DISABLED) + assert normal[2] == _curses.COLOR_BLACK + assert disabled[1:] == (_curses.COLOR_WHITE, _curses.COLOR_BLACK)