VaKU/openvpncertupdate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix TUI crash on serial consoles; make index.txt the sole email store

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Vlad Doloman
2026-07-08 15:46:10 +03:00
parent bd17cdd68d
commit 4798ce3f18
6 changed files with 127 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
CLAUDE.md
View File

@@ -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/<CN>_<YYYY-MM-DD>_<NN>/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: `<PKI_DIR>/inline/private/<CN>.inline`
- User emails stored in `<PKI_DIR>/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 `<PKI_DIR>/index.txt` itself. `get_email()` reads it back from there (`load_all_certs()` + CN match); there is no `openvpncertupdate-metadata.json`

80
openvpncertupdate.py
View File

@@ -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:

22
tests/test_cli.py
View File

@@ -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
# ---------------------------------------------------------------------------

16
tests/test_metadata.py
View File

@@ -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") == ""

43
tests/test_pki.py
View File

@@ -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"

35
tests/test_widgets.py
View File

@@ -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)