dedupe cert list rows to the last non-revoked entry per CN
index.txt is append-only: a CN left unrevoked after expiring, then reissued, accumulates multiple V-status lines. load_expiring_certs() and load_all_certs() previously returned one CertInfo per line, so such a CN showed as duplicate rows in the TUI main menu (and in --list/--list-all). Extract the existing "last non-revoked entry wins" scan (previously only in get_email()) into a shared _load_current_certs(), and build load_expiring_certs()/load_all_certs()/get_email() on top of it. The date-window filter in load_expiring_certs() now applies to each CN's canonical (most recent) entry rather than to raw lines, so a stale duplicate that happens to fall in the "recently expired" window no longer masks a live reissued cert whose real expiry is outside it. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Vlad Doloman
@@ -51,7 +51,7 @@ 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()`, `get_email()` |
|
||||
| PKI | `CertInfo`, `_load_current_certs()`, `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()` |
|
||||
| CONFIG | `build_ovpn()` → `vpn-configs/<CN>_<YYYY-MM-DD>_<NN>/CONFIG_NAME` |
|
||||
@@ -91,4 +91,5 @@ 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 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`
|
||||
- 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; there is no `openvpncertupdate-metadata.json`
|
||||
- `index.txt` is append-only and a CN can accumulate multiple V-status lines (e.g. left unrevoked after expiring, then reissued) alongside older R-status ones. `_load_current_certs()` is the single place that resolves this: keeps only the last (most recently appended) V-status line per CN. `load_expiring_certs()`, `load_all_certs()`, and `get_email()` all build on it, so the TUI list, `--list`/`--list-all`, and email lookups never show/use a stale duplicate
|
||||
|
||||
@@ -165,71 +165,61 @@ def _parse_index_line(line: str) -> Optional[tuple[str, datetime, str]]:
|
||||
return cn, expiry, email
|
||||
|
||||
|
||||
def _load_current_certs(pki_dir: str) -> list[CertInfo]:
|
||||
"""Return one CertInfo per CN: its most recent V-status index.txt entry.
|
||||
|
||||
index.txt is append-only, so a CN can accumulate several V-status lines
|
||||
(e.g. an old cert left unrevoked after it expired, then reissued) as
|
||||
well as older R-status (revoked) ones. Scan the whole file in order and
|
||||
keep only the last non-revoked line per CN — same rule get_email() uses
|
||||
for picking an email among duplicates, applied here to the row itself
|
||||
so expiring/all-certs views don't show stale duplicate rows.
|
||||
"""
|
||||
now = datetime.now(tz=timezone.utc)
|
||||
by_cn: dict[str, CertInfo] = {}
|
||||
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
|
||||
cn, expiry, email = parsed
|
||||
by_cn[cn] = CertInfo(
|
||||
cn=cn,
|
||||
expires=expiry,
|
||||
days_left=(expiry - now).days,
|
||||
email=email,
|
||||
)
|
||||
return list(by_cn.values())
|
||||
|
||||
|
||||
def load_expiring_certs(
|
||||
pki_dir: str,
|
||||
days_past: int,
|
||||
days_ahead: int,
|
||||
) -> list[CertInfo]:
|
||||
"""Return valid certs whose expiry falls within [-days_past, +days_ahead]."""
|
||||
"""Return current certs whose expiry falls within [-days_past, +days_ahead]."""
|
||||
now = datetime.now(tz=timezone.utc)
|
||||
cutoff_past = now - timedelta(days=days_past)
|
||||
cutoff_future = now + timedelta(days=days_ahead)
|
||||
results: list[CertInfo] = []
|
||||
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
|
||||
cn, expiry, email = parsed
|
||||
if cutoff_past <= expiry <= cutoff_future:
|
||||
results.append(CertInfo(
|
||||
cn=cn,
|
||||
expires=expiry,
|
||||
days_left=(expiry - now).days,
|
||||
email=email,
|
||||
))
|
||||
results = [c for c in _load_current_certs(pki_dir)
|
||||
if cutoff_past <= c.expires <= cutoff_future]
|
||||
results.sort(key=lambda c: c.expires)
|
||||
return results
|
||||
|
||||
|
||||
def load_all_certs(pki_dir: str) -> list[CertInfo]:
|
||||
"""Return all valid (V-status) certs, sorted by expiry date."""
|
||||
now = datetime.now(tz=timezone.utc)
|
||||
results: list[CertInfo] = []
|
||||
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
|
||||
cn, expiry, email = parsed
|
||||
results.append(CertInfo(
|
||||
cn=cn,
|
||||
expires=expiry,
|
||||
days_left=(expiry - now).days,
|
||||
email=email,
|
||||
))
|
||||
"""Return all current certs, sorted by expiry date."""
|
||||
results = _load_current_certs(pki_dir)
|
||||
results.sort(key=lambda c: c.expires)
|
||||
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
|
||||
"""Return the emailAddress on CN's current (most recent V-status) entry."""
|
||||
for c in _load_current_certs(pki_dir):
|
||||
if c.cn == cn:
|
||||
return c.email
|
||||
return ""
|
||||
|
||||
|
||||
# ============================================================
|
||||
|
||||
@@ -109,6 +109,43 @@ def test_load_all_certs_sorted_ascending(tmp_path):
|
||||
assert dates == sorted(dates)
|
||||
|
||||
|
||||
def _make_duplicate_cn_pki(tmp_path, now):
|
||||
"""index.txt with an old (still V-status, never revoked) entry for
|
||||
"dave" plus a newer reissued V-status entry for the same CN."""
|
||||
pki = tmp_path / "pki"
|
||||
pki.mkdir()
|
||||
e_old = (now - timedelta(days=5)).strftime("%y%m%d%H%M%S") + "Z" # expired, left unrevoked
|
||||
e_new = (now + timedelta(days=365)).strftime("%y%m%d%H%M%S") + "Z" # reissued, far future
|
||||
content = (
|
||||
f"V\t{e_old}\t\t01\tunknown\t/CN=dave/emailAddress=old@example.com\n"
|
||||
f"V\t{e_new}\t\t02\tunknown\t/CN=dave/emailAddress=newest@example.com\n"
|
||||
)
|
||||
(pki / "index.txt").write_text(content)
|
||||
return str(pki), e_old, e_new
|
||||
|
||||
|
||||
def test_load_all_certs_dedups_to_last_valid_entry(tmp_path):
|
||||
# A CN left unrevoked after expiring, then reissued, must appear once
|
||||
# — as its most recent entry — not as two rows.
|
||||
now = datetime.now(tz=timezone.utc)
|
||||
pki, _e_old, _e_new = _make_duplicate_cn_pki(tmp_path, now)
|
||||
certs = load_all_certs(pki)
|
||||
dave_rows = [c for c in certs if c.cn == "dave"]
|
||||
assert len(dave_rows) == 1
|
||||
assert dave_rows[0].email == "newest@example.com"
|
||||
assert dave_rows[0].days_left > 300
|
||||
|
||||
|
||||
def test_load_expiring_certs_filters_by_current_entry_not_stale_duplicate(tmp_path):
|
||||
# The stale (unrevoked) old entry for "dave" falls inside the
|
||||
# recently-expired window, but the *current* entry (reissued, far in
|
||||
# the future) doesn't — so "dave" must NOT show up as "expiring soon".
|
||||
now = datetime.now(tz=timezone.utc)
|
||||
pki, _e_old, _e_new = _make_duplicate_cn_pki(tmp_path, now)
|
||||
certs = load_expiring_certs(pki, days_past=30, days_ahead=14)
|
||||
assert "dave" not in {c.cn for c in certs}
|
||||
|
||||
|
||||
def test_get_email_reads_from_index_txt(tmp_path):
|
||||
now = datetime.now(tz=timezone.utc)
|
||||
pki = tmp_path / "pki"
|
||||
|
||||
Reference in New Issue
Block a user