VaKU/openvpncertupdate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
master
openvpncertupdate/docs/superpowers/plans/2026-06-23-openvpncertupdate.md
Vlad Doloman 6c60a63eb2 Add planning docs 
Track the superpowers planning documents under docs/superpowers/plans/.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-24 23:56:29 +03:00

1931 lines
60 KiB
Markdown
Raw Permalink Blame History

# openvpncertupdate Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Build a Python + curses TUI tool that manages OpenVPN user certificates via EasyRSA — showing expiring/expired certs, re-issuing with fresh keys and passwords, creating new certs, and delivering configs via Cryptgeon + email.
**Architecture:** Everything lives in a single file `openvpncertupdate.py` organized into clearly-delimited sections (SETTINGS → PKI → PASSWORD → EASYRSA → METADATA → CONFIG → CRYPTGEON → MAILER → TUI WIDGETS → TUI DIALOGS → TUI SCREEN → APP → main). Tests live in `tests/` and import from `openvpncertupdate` directly.
**Tech Stack:** Python 3.8+, curses (stdlib), cryptography>=41 (AES-GCM), subprocess + email + urllib (stdlib), msmtp or sendmail (system), EasyRSA 3.2.x.
## Global Constraints
- Single deliverable file: `openvpncertupdate.py`. No `src/` package.
- Python 3.8+ minimum; only external package is `cryptography`.
- All EasyRSA calls run with `--batch`; CA passphrase via `--passin=pass:<CA_PASSPHRASE>` (omit arg when `CA_PASSPHRASE = ""`).
- Re-issue workflow: `revoke-issued` → `build-client-full`. Do NOT use `easyrsa renew`.
- New private key encrypted with generated password via `--passout=pass:<password>`.
- CRL regeneration: `gen-crl` → copy to `CRL_DEST_PATH` → `chmod 644`. NOT auto-run during renewal — only on standalone `r` hotkey revoke.
- Inline file used: `<PKI_DIR>/inline/private/<CN>.inline` (contains key).
- Config output: `./vpn-configs/<CN>_<YYYY-MM-DD>/<CONFIG_NAME>`.
- User email stored in `<PKI_DIR>/openvpncertupdate-metadata.json`.
- Expiry window defaults: `DAYS_PAST=30`, `DAYS_AHEAD=14`.
- Password: 28 alphanumeric chars; pos 1 = uppercase; pos 2 = lowercase; pos 28 = lowercase; pos 3-27 = any alphanumeric; exclude `oO01lI` from all positions; additionally exclude `j` from positions 2 and 28.
- TUI keys: `r`=revoke highlighted; `q`=quit; `↑↓`=navigate; `Space`=toggle checkbox; `Enter`=activate.
- Main menu order (below cert list): `Create New Certificate` (grayed + notice when any checkbox checked) → `Regenerate CRL` → `Exit`.
---
## File Map
```
openvpncertupdate.py # THE single file — all sections below
# === SETTINGS ===
# === PKI === CertInfo dataclass, load_expiring_certs(), _parse_index_line(), _parse_date()
# === PASSWORD === generate_password()
# === EASYRSA === EasyRSAError, _run_easyrsa(), revoke_issued(), build_client_full(), gen_crl(), copy_crl()
# === METADATA === load_metadata(), save_email(), get_email()
# === CONFIG === build_ovpn()
# === CRYPTGEON === CryptgeonError, create_note()
# === MAILER === send_email()
# === TUI WIDGETS === COLOR_* constants, init_colors(), draw_box(), draw_centered(), clamp(), InputField
# === TUI DIALOGS === CertFormResult, show_cert_form(), show_confirm()
# === TUI SCREEN === Action enum, ScreenResult, show_main_screen()
# === APP === CursesApp
# === ENTRY POINT === main()
requirements.txt
email_template.txt
tests/
__init__.py
test_pki.py
test_password.py
test_cryptgeon.py
test_config_builder.py
test_metadata.py
test_easyrsa.py
test_widgets.py
```
All test files import symbols directly from `openvpncertupdate`:
```python
from openvpncertupdate import load_expiring_certs, CertInfo, ...
```
---
## Task 1: Skeleton + Settings + PKI Section
**Files:**
- Create: `openvpncertupdate.py` (settings + PKI section)
- Create: `requirements.txt`
- Create: `tests/__init__.py`
- Create: `tests/test_pki.py`
**Interfaces:**
- Produces:
```python
@dataclass
class CertInfo:
cn: str
expires: datetime # UTC
days_left: int # negative = already expired
def load_expiring_certs(pki_dir: str, days_past: int, days_ahead: int) -> list[CertInfo]:
"""Return valid certs in expiry window, sorted ascending by expiry date."""
def _parse_index_line(line: str) -> tuple[str, datetime] | None:
"""Parse one index.txt line. Returns (cn, expiry) for V-status lines or None."""
```
- [ ] **Step 1: Create `requirements.txt`**
```
cryptography>=41.0.0
```
- [ ] **Step 2: Create `tests/__init__.py`** (empty)
```bash
mkdir -p tests && touch tests/__init__.py
```
- [ ] **Step 3: Write failing tests — `tests/test_pki.py`**
```python
import textwrap
import pytest
from datetime import datetime, timezone, timedelta
from openvpncertupdate import load_expiring_certs, CertInfo, _parse_index_line
def _fmt(dt: datetime) -> str:
return dt.strftime("%y%m%d%H%M%S") + "Z"
def make_pki(tmp_path, now):
pki = tmp_path / "pki"
pki.mkdir()
content = textwrap.dedent(f"""\
V\t{_fmt(now + timedelta(days=10))}\t\t02\tunknown\t/CN=soon
V\t{_fmt(now + timedelta(days=90))}\t\t03\tunknown\t/CN=later
V\t{_fmt(now - timedelta(days=15))}\t\t04\tunknown\t/CN=past15
V\t{_fmt(now - timedelta(days=40))}\t\t05\tunknown\t/CN=past40
R\t{_fmt(now + timedelta(days=10))}\t230101Z,keyCompromise\t06\tunknown\t/CN=revoked
""")
(pki / "index.txt").write_text(content)
return str(pki)
def test_filters_within_window(tmp_path):
now = datetime.now(tz=timezone.utc)
pki = make_pki(tmp_path, now)
certs = load_expiring_certs(pki, days_past=30, days_ahead=14)
cns = {c.cn for c in certs}
assert "soon" in cns # 10d ahead < 14d threshold
assert "later" not in cns # 90d > 14d
assert "past15" in cns # 15d past < 30d threshold
assert "past40" not in cns # 40d past > 30d threshold
assert "revoked" not in cns # R status skipped
def test_sorted_ascending(tmp_path):
now = datetime.now(tz=timezone.utc)
pki = make_pki(tmp_path, now)
certs = load_expiring_certs(pki, days_past=30, days_ahead=14)
dates = [c.expires for c in certs]
assert dates == sorted(dates)
def test_days_left_negative_for_expired(tmp_path):
now = datetime.now(tz=timezone.utc)
pki = make_pki(tmp_path, now)
certs = load_expiring_certs(pki, days_past=30, days_ahead=14)
expired = next(c for c in certs if c.cn == "past15")
assert expired.days_left < 0
def test_parse_4digit_year():
line = "V\t20251231120000Z\t\t07\tunknown\t/CN=future"
result = _parse_index_line(line)
assert result is not None
cn, dt = result
assert cn == "future"
assert dt.year == 2025
```
- [ ] **Step 4: Run tests to confirm they fail**
```bash
python3 -m pytest tests/test_pki.py -v 2>&1 | head -6
```
Expected: `ModuleNotFoundError` or `ImportError` — `openvpncertupdate` doesn't exist yet.
- [ ] **Step 5: Create `openvpncertupdate.py` with SETTINGS + PKI sections**
```python
#!/usr/bin/env python3
"""openvpncertupdate — Manage OpenVPN user certificates via EasyRSA."""
from __future__ import annotations
import base64
import curses
import email.encoders
import email.mime.base
import email.mime.multipart
import email.mime.text
import hashlib
import json
import os
import secrets
import shutil
import string
import subprocess
import urllib.error
import urllib.request
from dataclasses import dataclass, field
from datetime import date, datetime, timedelta, timezone
from enum import Enum, auto
from pathlib import Path
from typing import Optional
try:
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
except ImportError: # pragma: no cover
AESGCM = None # type: ignore
# ============================================================
# SETTINGS — edit these for your environment
# ============================================================
EASYRSA_DIR = "/etc/easy-rsa"
EASYRSA_PKI_DIR = "/etc/easy-rsa/pki"
CA_PASSPHRASE = "" # empty string = no passphrase
OVPN_TEMPLATE_PATH = "./template.ovpn"
CONFIG_NAME = "client.ovpn"
VPN_CONFIGS_DIR = "./vpn-configs"
CRL_DEST_PATH = "/etc/openvpn/crl.pem"
CRYPTGEON_URL = "https://cryptgeon.example.com"
MAIL_FROM = "vpn-admin@example.com"
MAIL_SUBJECT = "Your VPN Configuration"
EMAIL_TEMPLATE_PATH = "./email_template.txt"
MAIL_BINARY = "msmtp" # or "sendmail"
DAYS_PAST = 30
DAYS_AHEAD = 14
# ============================================================
# === PKI ===
# ============================================================
@dataclass
class CertInfo:
cn: str
expires: datetime # UTC
days_left: int # negative = already expired
def _parse_date(raw: str) -> datetime:
"""Parse OpenSSL date YYMMDDHHMMSSZ or YYYYMMDDHHMMSSZ."""
raw = raw.rstrip("Z")
if len(raw) == 12:
yy = int(raw[:2])
year = 2000 + yy if yy < 50 else 1900 + yy
rest = raw[2:]
dt = datetime.strptime(f"{year}{rest}", "%Y%m%d%H%M%S")
else:
dt = datetime.strptime(raw, "%Y%m%d%H%M%S")
return dt.replace(tzinfo=timezone.utc)
def _parse_index_line(line: str) -> Optional[tuple[str, datetime]]:
"""Return (cn, expiry) for valid (V-status) index.txt lines, else None."""
parts = line.rstrip("\n").split("\t")
if len(parts) < 6 or parts[0] != "V":
return None
try:
expiry = _parse_date(parts[1])
except ValueError:
return None
dn = parts[5]
cn = None
for seg in dn.split("/"):
if seg.startswith("CN="):
cn = seg[3:]
break
if not cn:
return None
return cn, expiry
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]."""
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 = parsed
if cutoff_past <= expiry <= cutoff_future:
results.append(CertInfo(
cn=cn,
expires=expiry,
days_left=(expiry - now).days,
))
results.sort(key=lambda c: c.expires)
return results
# ============================================================
# (remaining sections added in later tasks)
# ============================================================
def main() -> None:
pass # replaced in Task 11
if __name__ == "__main__":
main()
```
- [ ] **Step 6: Run PKI tests**
```bash
python3 -m pytest tests/test_pki.py -v
```
Expected: all 4 tests PASS.
- [ ] **Step 7: Commit**
```bash
git add openvpncertupdate.py requirements.txt tests/__init__.py tests/test_pki.py
git commit -m "feat: project skeleton + settings + PKI index.txt reader"
```
---
## Task 2: Password Section
**Files:**
- Modify: `openvpncertupdate.py` (add `=== PASSWORD ===` section)
- Create: `tests/test_password.py`
**Interfaces:**
- Produces:
```python
def generate_password() -> str:
"""Return a 28-char alphanumeric password per the charset rules."""
```
- [ ] **Step 1: Write failing tests — `tests/test_password.py`**
```python
from openvpncertupdate import generate_password
BANNED_ALL = set("oO01lI")
BANNED_POS2_LAST = BANNED_ALL | {"j"}
def test_length():
assert len(generate_password()) == 28
def test_pos1_uppercase():
for _ in range(200):
pw = generate_password()
assert pw[0].isupper(), f"pos1 not uppercase: {pw}"
def test_pos2_lowercase_no_banned():
for _ in range(200):
pw = generate_password()
c = pw[1]
assert c.islower(), f"pos2 not lowercase: {pw}"
assert c not in BANNED_POS2_LAST, f"pos2 banned '{c}': {pw}"
def test_last_lowercase_no_banned():
for _ in range(200):
pw = generate_password()
c = pw[-1]
assert c.islower(), f"last not lowercase: {pw}"
assert c not in BANNED_POS2_LAST, f"last banned '{c}': {pw}"
def test_all_alphanumeric():
for _ in range(200):
pw = generate_password()
assert pw.isalnum(), f"non-alphanumeric in: {pw}"
def test_no_globally_banned():
for _ in range(500):
pw = generate_password()
for c in BANNED_ALL:
assert c not in pw, f"banned '{c}' in: {pw}"
def test_uniqueness():
assert len({generate_password() for _ in range(100)}) == 100
```
- [ ] **Step 2: Run to confirm failure**
```bash
python3 -m pytest tests/test_password.py -v 2>&1 | head -6
```
Expected: `ImportError: cannot import name 'generate_password'`
- [ ] **Step 3: Add PASSWORD section to `openvpncertupdate.py`**
Insert this block before the `# (remaining sections)` comment (replace that comment):
```python
# ============================================================
# === PASSWORD ===
# ============================================================
_BANNED_ALL = frozenset("oO01lI")
_BANNED_POS2_LAST = _BANNED_ALL | frozenset("j")
_UPPER = [c for c in string.ascii_uppercase if c not in _BANNED_ALL]
_LOWER_RESTRICTED = [c for c in string.ascii_lowercase if c not in _BANNED_POS2_LAST]
_LOWER_MID = [c for c in string.ascii_lowercase if c not in _BANNED_ALL]
_DIGITS_MID = [c for c in string.digits if c not in _BANNED_ALL]
_MID = _UPPER + _LOWER_MID + _DIGITS_MID # 55 chars for pos 3-27
def generate_password() -> str:
pw = [
secrets.choice(_UPPER), # pos 1: uppercase
secrets.choice(_LOWER_RESTRICTED), # pos 2: lowercase, no j
]
for _ in range(25): # pos 3-27: any safe alphanumeric
pw.append(secrets.choice(_MID))
pw.append(secrets.choice(_LOWER_RESTRICTED)) # pos 28: lowercase, no j
return "".join(pw)
```
- [ ] **Step 4: Run tests**
```bash
python3 -m pytest tests/test_password.py -v
```
Expected: all 7 tests PASS.
- [ ] **Step 5: Commit**
```bash
git add openvpncertupdate.py tests/test_password.py
git commit -m "feat: password generator with 28-char charset rules"
```
---
## Task 3: EasyRSA Section
**Files:**
- Modify: `openvpncertupdate.py` (add `=== EASYRSA ===` section)
- Create: `tests/test_easyrsa.py`
**Interfaces:**
- Produces:
```python
class EasyRSAError(Exception): ...
def revoke_issued(easyrsa_dir: str, pki_dir: str, cn: str, ca_passphrase: str) -> None: ...
def build_client_full(easyrsa_dir: str, pki_dir: str, cn: str, key_passphrase: str, ca_passphrase: str) -> None: ...
def gen_crl(easyrsa_dir: str, pki_dir: str, ca_passphrase: str) -> None: ...
def copy_crl(pki_dir: str, dest_path: str) -> None: ...
```
- [ ] **Step 1: Write failing tests — `tests/test_easyrsa.py`**
```python
import pytest
from unittest.mock import patch, MagicMock
from openvpncertupdate import (
revoke_issued, build_client_full, gen_crl, copy_crl, EasyRSAError,
)
def ok_result():
r = MagicMock(); r.returncode = 0; r.stderr = ""; return r
def err_result():
r = MagicMock(); r.returncode = 1; r.stderr = "oops"; return r
@patch("openvpncertupdate.subprocess.run")
def test_revoke_includes_required_args(mock_run):
mock_run.return_value = ok_result()
revoke_issued("/er", "/pki", "alice", "capass")
args = mock_run.call_args[0][0]
assert args[0] == "/er/easyrsa"
assert "--batch" in args
assert "--pki=/pki" in args
assert "--passin=pass:capass" in args
assert "revoke-issued" in args
assert "alice" in args
@patch("openvpncertupdate.subprocess.run")
def test_revoke_omits_passin_when_empty(mock_run):
mock_run.return_value = ok_result()
revoke_issued("/er", "/pki", "alice", "")
args = mock_run.call_args[0][0]
assert not any(a.startswith("--passin") for a in args)
@patch("openvpncertupdate.subprocess.run")
def test_revoke_raises_on_failure(mock_run):
mock_run.return_value = err_result()
with pytest.raises(EasyRSAError, match="revoke-issued"):
revoke_issued("/er", "/pki", "alice", "")
@patch("openvpncertupdate.subprocess.run")
def test_build_client_full_args(mock_run):
mock_run.return_value = ok_result()
build_client_full("/er", "/pki", "bob", "keypass", "capass")
args = mock_run.call_args[0][0]
assert "--passout=pass:keypass" in args
assert "--passin=pass:capass" in args
assert "build-client-full" in args
assert "bob" in args
@patch("openvpncertupdate.subprocess.run")
def test_gen_crl_args(mock_run):
mock_run.return_value = ok_result()
gen_crl("/er", "/pki", "capass")
args = mock_run.call_args[0][0]
assert "gen-crl" in args
@patch("openvpncertupdate.shutil.copy2")
@patch("openvpncertupdate.os.chmod")
def test_copy_crl_copies_and_chmods(mock_chmod, mock_copy):
copy_crl("/pki", "/etc/openvpn/crl.pem")
mock_copy.assert_called_once_with("/pki/crl.pem", "/etc/openvpn/crl.pem")
mock_chmod.assert_called_once_with("/etc/openvpn/crl.pem", 0o644)
```
- [ ] **Step 2: Run to confirm failure**
```bash
python3 -m pytest tests/test_easyrsa.py -v 2>&1 | head -6
```
Expected: `ImportError: cannot import name 'EasyRSAError'`
- [ ] **Step 3: Add EASYRSA section to `openvpncertupdate.py`**
```python
# ============================================================
# === EASYRSA ===
# ============================================================
class EasyRSAError(Exception):
pass
def _run_easyrsa(cmd: list[str], cwd: str) -> None:
result = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True)
if result.returncode != 0:
verb = cmd[1] if len(cmd) > 1 else cmd[0]
raise EasyRSAError(
f"{verb} failed (exit {result.returncode}): {result.stderr.strip()}"
)
def _base_cmd(easyrsa_dir: str, pki_dir: str, ca_passphrase: str) -> list[str]:
cmd = [f"{easyrsa_dir}/easyrsa", "--batch", f"--pki={pki_dir}"]
if ca_passphrase:
cmd.append(f"--passin=pass:{ca_passphrase}")
return cmd
def revoke_issued(
easyrsa_dir: str, pki_dir: str, cn: str, ca_passphrase: str
) -> None:
_run_easyrsa(
_base_cmd(easyrsa_dir, pki_dir, ca_passphrase) + ["revoke-issued", cn],
cwd=easyrsa_dir,
)
def build_client_full(
easyrsa_dir: str, pki_dir: str, cn: str,
key_passphrase: str, ca_passphrase: str,
) -> None:
_run_easyrsa(
_base_cmd(easyrsa_dir, pki_dir, ca_passphrase)
+ [f"--passout=pass:{key_passphrase}", "build-client-full", cn],
cwd=easyrsa_dir,
)
def gen_crl(easyrsa_dir: str, pki_dir: str, ca_passphrase: str) -> None:
_run_easyrsa(
_base_cmd(easyrsa_dir, pki_dir, ca_passphrase) + ["gen-crl"],
cwd=easyrsa_dir,
)
def copy_crl(pki_dir: str, dest_path: str) -> None:
shutil.copy2(f"{pki_dir}/crl.pem", dest_path)
os.chmod(dest_path, 0o644)
```
- [ ] **Step 4: Run tests**
```bash
python3 -m pytest tests/test_easyrsa.py -v
```
Expected: all 6 tests PASS.
- [ ] **Step 5: Commit**
```bash
git add openvpncertupdate.py tests/test_easyrsa.py
git commit -m "feat: EasyRSA subprocess wrappers (revoke, build, gen-crl, copy-crl)"
```
---
## Task 4: Metadata + Config Builder Sections
**Files:**
- Modify: `openvpncertupdate.py` (add `=== METADATA ===` and `=== CONFIG ===` sections)
- Create: `tests/test_metadata.py`
- Create: `tests/test_config_builder.py`
**Interfaces:**
- Produces:
```python
def load_metadata(pki_dir: str) -> dict[str, str]: ...
def save_email(pki_dir: str, cn: str, email: str) -> None: ...
def get_email(pki_dir: str, cn: str) -> str: ...
def build_ovpn(cn: str, pki_dir: str, template_path: str,
output_base_dir: str, config_name: str) -> str:
"""Returns absolute path to written .ovpn file."""
```
- [ ] **Step 1: Write failing tests**
`tests/test_metadata.py`:
```python
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") == ""
```
`tests/test_config_builder.py`:
```python
import os
from pathlib import Path
from datetime import date
from openvpncertupdate import build_ovpn
def setup_pki(tmp_path, cn, inline_content="<ca>X</ca>"):
pki = tmp_path / "pki"
private_dir = pki / "inline" / "private"
private_dir.mkdir(parents=True)
(private_dir / f"{cn}.inline").write_text(inline_content)
return str(pki)
def test_creates_file_with_combined_content(tmp_path):
pki = setup_pki(tmp_path, "alice", "<ca>CERT</ca>")
tmpl = tmp_path / "t.ovpn"
tmpl.write_text("client\ndev tun\n")
result = build_ovpn("alice", pki, str(tmpl), str(tmp_path / "out"), "vpn.ovpn")
content = Path(result).read_text()
assert "client" in content
assert "CERT" in content
def test_output_path_has_cn_and_date(tmp_path):
pki = setup_pki(tmp_path, "bob")
tmpl = tmp_path / "t.ovpn"; tmpl.write_text("")
today = date.today().strftime("%Y-%m-%d")
result = build_ovpn("bob", pki, str(tmpl), str(tmp_path / "out"), "vpn.ovpn")
assert f"bob_{today}" in result
assert result.endswith("vpn.ovpn")
```
- [ ] **Step 2: Run to confirm failure**
```bash
python3 -m pytest tests/test_metadata.py tests/test_config_builder.py -v 2>&1 | head -6
```
- [ ] **Step 3: Add METADATA and CONFIG sections to `openvpncertupdate.py`**
```python
# ============================================================
# === 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 ===
# ============================================================
def build_ovpn(
cn: str,
pki_dir: str,
template_path: str,
output_base_dir: str,
config_name: str,
) -> str:
"""Concat template + inline file. Return absolute path to written .ovpn."""
inline_path = os.path.join(pki_dir, "inline", "private", f"{cn}.inline")
template_content = Path(template_path).read_text()
inline_content = Path(inline_path).read_text()
today = date.today().strftime("%Y-%m-%d")
out_dir = Path(output_base_dir) / f"{cn}_{today}"
out_dir.mkdir(parents=True, exist_ok=True)
out_file = out_dir / config_name
out_file.write_text(template_content + "\n" + inline_content)
return str(out_file.resolve())
```
- [ ] **Step 4: Run tests**
```bash
python3 -m pytest tests/test_metadata.py tests/test_config_builder.py -v
```
Expected: all 6 tests PASS.
- [ ] **Step 5: Commit**
```bash
git add openvpncertupdate.py tests/test_metadata.py tests/test_config_builder.py
git commit -m "feat: metadata store (CN→email) and .ovpn config builder"
```
---
## Task 5: Cryptgeon Section
**Files:**
- Modify: `openvpncertupdate.py` (add `=== CRYPTGEON ===` section)
- Create: `tests/test_cryptgeon.py`
**Interfaces:**
- Produces:
```python
class CryptgeonError(Exception): ...
def create_note(content: str, base_url: str) -> str:
"""Encrypt content, POST to Cryptgeon, return one-time URL."""
```
URL format: `<base_url>/#/note/<id>/<base64url-key>`
- [ ] **Step 1: Install cryptography**
```bash
pip3 install 'cryptography>=41.0.0'
```
- [ ] **Step 2: Write failing tests — `tests/test_cryptgeon.py`**
```python
import json
import pytest
from unittest.mock import patch, MagicMock
from openvpncertupdate import create_note, CryptgeonError
def mock_response(body: bytes):
resp = MagicMock()
resp.read.return_value = body
resp.__enter__ = lambda s: s
resp.__exit__ = MagicMock(return_value=False)
return resp
@patch("openvpncertupdate.urllib.request.urlopen")
def test_returns_url_with_id(mock_open):
mock_open.return_value = mock_response(b'{"id":"abc-123"}')
url = create_note("secret", "https://cg.example.com")
assert "abc-123" in url
assert url.startswith("https://cg.example.com/#/note/abc-123/")
@patch("openvpncertupdate.urllib.request.urlopen")
def test_posts_to_correct_endpoint(mock_open):
mock_open.return_value = mock_response(b'{"id":"x"}')
create_note("hello", "https://cg.example.com")
req = mock_open.call_args[0][0]
assert req.full_url == "https://cg.example.com/api/notes/"
body = json.loads(req.data)
assert isinstance(body["contents"], list)
assert body["views"] == 1
assert body["type"] == "text"
@patch("openvpncertupdate.urllib.request.urlopen")
def test_raises_on_http_error(mock_open):
import urllib.error
mock_open.side_effect = urllib.error.HTTPError("", 500, "Error", None, None)
with pytest.raises(CryptgeonError, match="500"):
create_note("secret", "https://cg.example.com")
@patch("openvpncertupdate.urllib.request.urlopen")
def test_different_passwords_give_different_urls(mock_open):
mock_open.return_value = mock_response(b'{"id":"same"}')
url1 = create_note("pass1", "https://cg.example.com")
url2 = create_note("pass2", "https://cg.example.com")
assert url1 != url2
```
- [ ] **Step 3: Run to confirm failure**
```bash
python3 -m pytest tests/test_cryptgeon.py -v 2>&1 | head -6
```
- [ ] **Step 4: Add CRYPTGEON section to `openvpncertupdate.py`**
```python
# ============================================================
# === CRYPTGEON ===
# ============================================================
class CryptgeonError(Exception):
pass
def create_note(content: str, base_url: str) -> str:
"""AES-256-GCM encrypt content, POST to Cryptgeon, return one-time URL.
Encryption matches the Cryptgeon browser client:
raw_key (32 bytes) → URL fragment (base64url, no padding)
aes_key = SHA-256(raw_key) → AES-256-GCM
payload = list(nonce + ciphertext)
"""
if AESGCM is None:
raise CryptgeonError("cryptography package not installed (pip install cryptography)")
raw_key = os.urandom(32)
aes_key = hashlib.sha256(raw_key).digest()
nonce = os.urandom(12)
aesgcm = AESGCM(aes_key)
ciphertext = aesgcm.encrypt(nonce, content.encode("utf-8"), None)
payload = json.dumps({
"contents": list(nonce + ciphertext),
"views": 1,
"type": "text",
}).encode("utf-8")
api_url = base_url.rstrip("/") + "/api/notes/"
req = urllib.request.Request(
api_url,
data=payload,
headers={"Content-Type": "application/json"},
method="POST",
)
try:
with urllib.request.urlopen(req) as resp:
body = json.loads(resp.read())
except urllib.error.HTTPError as exc:
raise CryptgeonError(f"Cryptgeon API returned {exc.code}: {exc.reason}") from exc
except Exception as exc:
raise CryptgeonError(f"Cryptgeon request failed: {exc}") from exc
note_id = body["id"]
key_fragment = base64.urlsafe_b64encode(raw_key).rstrip(b"=").decode()
return f"{base_url.rstrip('/')}/#/note/{note_id}/{key_fragment}"
```
- [ ] **Step 5: Run tests**
```bash
python3 -m pytest tests/test_cryptgeon.py -v
```
Expected: all 4 tests PASS.
- [ ] **Step 6: Commit**
```bash
git add openvpncertupdate.py tests/test_cryptgeon.py requirements.txt
git commit -m "feat: Cryptgeon client with AES-GCM client-side encryption"
```
---
## Task 6: Mailer Section
**Files:**
- Modify: `openvpncertupdate.py` (add `=== MAILER ===` section)
- Create: `tests/test_mailer.py`
- Create: `email_template.txt`
**Interfaces:**
- Produces:
```python
def send_email(
to_address: str, cn: str, one_time_url: str, ovpn_path: str,
mail_from: str, subject: str, template_path: str, mail_binary: str,
) -> None:
"""Compose MIME email from template + attachment, pipe to mail_binary -t.
Template placeholders: {cn}, {url}, {config_name}. Raises RuntimeError on failure."""
```
- [ ] **Step 1: Create `email_template.txt`**
```
Hello,
Your OpenVPN configuration has been updated for: {cn}
Retrieve your one-time password here (link expires after one view):
{url}
Your configuration file is attached as '{config_name}'.
Regards,
VPN Admin
```
- [ ] **Step 2: Write failing tests — `tests/test_mailer.py`**
```python
import pytest
from unittest.mock import patch, MagicMock
from openvpncertupdate import send_email
def make_popen(returncode=0):
proc = MagicMock()
proc.communicate.return_value = (b"", b"")
proc.returncode = returncode
return proc
def test_calls_mail_binary(tmp_path):
tmpl = tmp_path / "t.txt"; tmpl.write_text("{cn} {url} {config_name}")
att = tmp_path / "c.ovpn"; att.write_text("x")
captured = {}
def fake_popen(cmd, **kw): captured["cmd"] = cmd; return make_popen()
with patch("openvpncertupdate.subprocess.Popen", side_effect=fake_popen):
send_email("to@x.com", "alice", "http://u", str(att),
"f@x.com", "Subj", str(tmpl), "msmtp")
assert captured["cmd"] == ["msmtp", "-t"]
def test_body_contains_cn_and_url(tmp_path):
tmpl = tmp_path / "t.txt"; tmpl.write_text("Dear {cn}, see {url} for {config_name}")
att = tmp_path / "c.ovpn"; att.write_text("")
received = {}
def fake_popen(cmd, **kw):
p = MagicMock()
p.communicate = lambda input=None: (received.update({"data": input}), b"")[1:]
p.returncode = 0
return p
with patch("openvpncertupdate.subprocess.Popen", side_effect=fake_popen):
send_email("to@x.com", "bob", "http://secret", str(att),
"f@x.com", "s", str(tmpl), "msmtp")
body = received["data"].decode()
assert "bob" in body
assert "http://secret" in body
def test_raises_on_failure(tmp_path):
tmpl = tmp_path / "t.txt"; tmpl.write_text("{cn} {url} {config_name}")
att = tmp_path / "c.ovpn"; att.write_text("")
with patch("openvpncertupdate.subprocess.Popen", return_value=make_popen(returncode=1)):
with pytest.raises(RuntimeError, match="mail"):
send_email("x@y.com", "cn", "url", str(att),
"f@x.com", "s", str(tmpl), "msmtp")
```
- [ ] **Step 3: Run to confirm failure**
```bash
python3 -m pytest tests/test_mailer.py -v 2>&1 | head -6
```
- [ ] **Step 4: Add MAILER section to `openvpncertupdate.py`**
```python
# ============================================================
# === MAILER ===
# ============================================================
def send_email(
to_address: str,
cn: str,
one_time_url: str,
ovpn_path: str,
mail_from: str,
subject: str,
template_path: str,
mail_binary: str,
) -> None:
"""Compose and send email with .ovpn attachment via msmtp/sendmail."""
config_name = os.path.basename(ovpn_path)
body = Path(template_path).read_text().format(
cn=cn, url=one_time_url, config_name=config_name,
)
msg = email.mime.multipart.MIMEMultipart()
msg["From"] = mail_from
msg["To"] = to_address
msg["Subject"] = subject
msg.attach(email.mime.text.MIMEText(body, "plain"))
with open(ovpn_path, "rb") as fh:
part = email.mime.base.MIMEBase("application", "octet-stream")
part.set_payload(fh.read())
email.encoders.encode_base64(part)
part.add_header("Content-Disposition", f'attachment; filename="{config_name}"')
msg.attach(part)
proc = subprocess.Popen(
[mail_binary, "-t"],
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
)
_, err = proc.communicate(input=msg.as_bytes())
if proc.returncode != 0:
raise RuntimeError(
f"mail delivery failed (exit {proc.returncode}): {err.decode().strip()}"
)
```
- [ ] **Step 5: Run tests**
```bash
python3 -m pytest tests/test_mailer.py -v
```
Expected: all 3 tests PASS.
- [ ] **Step 6: Commit**
```bash
git add openvpncertupdate.py tests/test_mailer.py email_template.txt
git commit -m "feat: MIME email composer with .ovpn attachment via msmtp/sendmail"
```
---
## Task 7: TUI Widget Section
**Files:**
- Modify: `openvpncertupdate.py` (add `=== TUI WIDGETS ===` section)
- Create: `tests/test_widgets.py`
**Interfaces:**
- Produces:
```python
COLOR_NORMAL: int; COLOR_SELECTED: int; COLOR_DISABLED: int
COLOR_TITLE: int; COLOR_ERROR: int; COLOR_SUCCESS: int
def init_colors() -> None: ...
def draw_box(win, title: str = "") -> None: ...
def draw_centered(win, y: int, text: str, attr: int = 0) -> None: ...
def clamp(value: int, lo: int, hi: int) -> int: ...
class InputField:
def __init__(self, win, y: int, x: int, width: int,
initial: str = "", mask: bool = False): ...
def draw(self) -> None: ...
def handle_key(self, key: int) -> None: ...
@property
def value(self) -> str: ...
```
- [ ] **Step 1: Write `tests/test_widgets.py`** (tests only the logic of `InputField` — no curses terminal needed)
```python
from unittest.mock import MagicMock
import curses as _curses
# Stub just enough of the curses API for InputField to import
_curses.color_pair = lambda x: 0
_curses.A_UNDERLINE = 0
_curses.A_BOLD = 0
_curses.KEY_BACKSPACE = 263
_curses.KEY_LEFT = 260
_curses.KEY_RIGHT = 261
_curses.KEY_HOME = 262
_curses.KEY_END = 360
_curses.KEY_DC = 330
from openvpncertupdate import InputField, clamp
def _field(initial="", mask=False):
win = MagicMock()
win.getmaxyx.return_value = (1, 80)
return InputField(win, 0, 0, 40, initial=initial, mask=mask)
def test_initial_value():
assert _field("hello").value == "hello"
def test_backspace_removes_last():
f = _field("hello")
f.handle_key(_curses.KEY_BACKSPACE)
assert f.value == "hell"
def test_type_appends():
f = _field("hi")
f.handle_key(ord("!"))
assert f.value == "hi!"
def test_left_then_insert():
f = _field("ac")
f.handle_key(_curses.KEY_LEFT)
f.handle_key(_curses.KEY_LEFT)
f.handle_key(ord("b"))
assert f.value == "bac"
def test_delete_key():
f = _field("abc")
f.handle_key(_curses.KEY_HOME)
f.handle_key(_curses.KEY_DC)
assert f.value == "bc"
def test_clamp_bounds():
assert clamp(5, 0, 10) == 5
assert clamp(-1, 0, 10) == 0
assert clamp(11, 0, 10) == 10
```
- [ ] **Step 2: Run to confirm failure**
```bash
python3 -m pytest tests/test_widgets.py -v 2>&1 | head -6
```
- [ ] **Step 3: Add TUI WIDGETS section to `openvpncertupdate.py`**
```python
# ============================================================
# === TUI WIDGETS ===
# ============================================================
COLOR_NORMAL = 1
COLOR_SELECTED = 2
COLOR_DISABLED = 3
COLOR_TITLE = 4
COLOR_ERROR = 5
COLOR_SUCCESS = 6
def init_colors() -> None:
curses.start_color()
curses.use_default_colors()
curses.init_pair(COLOR_NORMAL, curses.COLOR_WHITE, -1)
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)
def draw_box(win, title: str = "") -> None:
win.box()
if title:
_, w = win.getmaxyx()
label = f" {title} "
x = max(1, (w - len(label)) // 2)
try:
win.addstr(0, x, label, curses.color_pair(COLOR_TITLE) | curses.A_BOLD)
except curses.error:
pass
def draw_centered(win, y: int, text: str, attr: int = 0) -> None:
_, w = win.getmaxyx()
x = max(0, (w - len(text)) // 2)
try:
win.addstr(y, x, text, attr)
except curses.error:
pass
def clamp(value: int, lo: int, hi: int) -> int:
return max(lo, min(hi, value))
class InputField:
"""Single-line editable text field. draw() needs a live curses window;
handle_key() + value are pure logic."""
def __init__(
self, win, y: int, x: int, width: int,
initial: str = "", mask: bool = False,
) -> None:
self._win = win
self._y = y
self._x = x
self._width = width
self._mask = mask
self._buf = list(initial)
self._cur = len(self._buf)
@property
def value(self) -> str:
return "".join(self._buf)
def draw(self) -> None:
display = ("*" * len(self._buf)) if self._mask else "".join(self._buf)
start = max(0, self._cur - self._width + 1)
visible = display[start: start + self._width].ljust(self._width)
try:
self._win.addstr(self._y, self._x, visible,
curses.color_pair(COLOR_NORMAL) | curses.A_UNDERLINE)
self._win.move(self._y, self._x + min(self._cur - start, self._width - 1))
except curses.error:
pass
def handle_key(self, key: int) -> None:
if key in (curses.KEY_BACKSPACE, 127, 8):
if self._cur > 0:
del self._buf[self._cur - 1]
self._cur -= 1
elif key == curses.KEY_LEFT:
self._cur = clamp(self._cur - 1, 0, len(self._buf))
elif key == curses.KEY_RIGHT:
self._cur = clamp(self._cur + 1, 0, len(self._buf))
elif key == curses.KEY_HOME:
self._cur = 0
elif key == curses.KEY_END:
self._cur = len(self._buf)
elif key == curses.KEY_DC:
if self._cur < len(self._buf):
del self._buf[self._cur]
elif 32 <= key <= 126:
self._buf.insert(self._cur, chr(key))
self._cur += 1
```
- [ ] **Step 4: Run tests**
```bash
python3 -m pytest tests/test_widgets.py -v
```
Expected: all 6 tests PASS.
- [ ] **Step 5: Commit**
```bash
git add openvpncertupdate.py tests/test_widgets.py
git commit -m "feat: TUI widget primitives (colors, InputField, draw helpers)"
```
---
## Task 8: TUI Dialogs Section
**Files:**
- Modify: `openvpncertupdate.py` (add `=== TUI DIALOGS ===` section)
No unit tests for dialog rendering (requires a terminal). Logic is covered by TUI WIDGETS tests.
- [ ] **Step 1: Add TUI DIALOGS section to `openvpncertupdate.py`**
```python
# ============================================================
# === TUI DIALOGS ===
# ============================================================
def show_confirm(stdscr, message: str) -> bool:
"""Y/N modal dialog. Returns True if user presses y/Y."""
sh, sw = stdscr.getmaxyx()
lines = message.splitlines()
h = len(lines) + 6
w = max(max(len(l) for l in lines) + 6, 38)
win = curses.newwin(h, w, (sh - h) // 2, (sw - w) // 2)
win.keypad(True)
draw_box(win, "Confirm")
for i, line in enumerate(lines):
try:
win.addstr(2 + i, 3, line, curses.color_pair(COLOR_NORMAL))
except curses.error:
pass
prompt = " [Y] Confirm [N / Esc] Cancel "
try:
win.addstr(h - 2, max(1, (w - len(prompt)) // 2), prompt,
curses.color_pair(COLOR_NORMAL))
except curses.error:
pass
win.refresh()
while True:
key = win.getch()
if key in (ord("y"), ord("Y")):
return True
if key in (ord("n"), ord("N"), 27, ord("q")):
return False
@dataclass
class CertFormResult:
cn: str
email: str
password: str
cancelled: bool
_FORM_FIELDS = ["cn", "email", "password"]
_FORM_LABELS = {
"cn": "Common Name (CN):",
"email": "Email address:",
"password": "Password [F5 = regenerate]:",
}
_FORM_FIELD_Y = {"cn": 3, "email": 6, "password": 9}
def show_cert_form(
stdscr,
cn: str = "",
email: str = "",
cn_readonly: bool = False,
) -> CertFormResult:
"""Modal cert-detail form.
Tab / Shift-Tab cycle fields. F5 regenerates password.
Enter on last field confirms. Escape cancels."""
sh, sw = stdscr.getmaxyx()
h, w = 16, 62
win = curses.newwin(h, w, (sh - h) // 2, (sw - w) // 2)
win.keypad(True)
fw = w - 6 # field width
active = [n for n in _FORM_FIELDS if not (n == "cn" and cn_readonly)]
fields: dict[str, InputField] = {
"cn": InputField(win, _FORM_FIELD_Y["cn"], 3, fw, initial=cn),
"email": InputField(win, _FORM_FIELD_Y["email"], 3, fw, initial=email),
"password": InputField(win, _FORM_FIELD_Y["password"], 3, fw,
initial=generate_password()),
}
focus = 0
while True:
win.clear()
draw_box(win, "Certificate Details")
for name in _FORM_FIELDS:
fy = _FORM_FIELD_Y[name]
label = _FORM_LABELS[name]
ro = name == "cn" and cn_readonly
attr = curses.color_pair(COLOR_DISABLED if ro else COLOR_NORMAL)
try:
win.addstr(fy - 1, 3, label, attr)
except curses.error:
pass
if ro:
try:
win.addstr(fy, 3, cn.ljust(fw)[:fw],
curses.color_pair(COLOR_DISABLED) | curses.A_UNDERLINE)
except curses.error:
pass
else:
fields[name].draw()
hint = "Tab=next F5=regen password Enter=confirm Esc=cancel"
try:
win.addstr(h - 2, 2, hint[:w - 4], curses.color_pair(COLOR_DISABLED))
except curses.error:
pass
win.refresh()
key = win.getch()
if key == 27:
return CertFormResult(cn=cn, email="", password="", cancelled=True)
if key == 9: # Tab
focus = (focus + 1) % len(active)
continue
if key == curses.KEY_BTAB: # Shift-Tab
focus = (focus - 1) % len(active)
continue
if key == curses.KEY_F5:
fields["password"] = InputField(
win, _FORM_FIELD_Y["password"], 3, fw, initial=generate_password(),
)
continue
if key in (10, 13, curses.KEY_ENTER):
if focus < len(active) - 1:
focus += 1
continue
final_cn = cn if cn_readonly else fields["cn"].value
return CertFormResult(
cn=final_cn,
email=fields["email"].value,
password=fields["password"].value,
cancelled=False,
)
fields[active[focus]].handle_key(key)
```
- [ ] **Step 2: Commit**
```bash
git add openvpncertupdate.py
git commit -m "feat: TUI modal dialogs (confirm Y/N, cert detail form)"
```
---
## Task 9: TUI Main Screen Section
**Files:**
- Modify: `openvpncertupdate.py` (add `=== TUI SCREEN ===` section)
- [ ] **Step 1: Add TUI SCREEN section to `openvpncertupdate.py`**
```python
# ============================================================
# === TUI SCREEN ===
# ============================================================
class Action(Enum):
RENEW_SELECTED = auto()
NEW_CERT = auto()
REGEN_CRL = auto()
REVOKE = auto()
EXIT = auto()
@dataclass
class ScreenResult:
action: Action
selected_cns: list[str] = field(default_factory=list)
_MENU_NEW = "Create New Certificate"
_MENU_CRL = "Regenerate CRL"
_MENU_EXIT = "Exit"
_MENU_ITEMS = [_MENU_NEW, _MENU_CRL, _MENU_EXIT]
def _expiry_label(c: CertInfo) -> str:
if c.days_left < 0:
return f"[EXPIRED {abs(c.days_left)}d ago]"
if c.days_left == 0:
return "[EXPIRES TODAY!]"
return f"[exp in {c.days_left}d] "
def show_main_screen(stdscr, certs: list[CertInfo]) -> ScreenResult:
"""Display cert selection list. Returns when user activates an action."""
init_colors()
curses.curs_set(0)
n_certs = len(certs)
n_items = n_certs + len(_MENU_ITEMS)
checked: set[str] = set()
cursor = 0
def is_cert(idx): return idx < n_certs
def menu_item(idx): return _MENU_ITEMS[idx - n_certs] if idx >= n_certs else None
def draw():
stdscr.clear()
sh, sw = stdscr.getmaxyx()
# Header bar
hdr = " openvpncertupdate — VPN Certificate Manager "
try:
stdscr.addstr(0, 0, hdr.ljust(sw)[:sw],
curses.color_pair(COLOR_TITLE) | curses.A_BOLD)
except curses.error:
pass
# Footer hint
hint = " ↑↓:Navigate Space:Check Enter:Confirm R:Revoke Q:Quit"
try:
stdscr.addstr(sh - 1, 0, hint[:sw], curses.color_pair(COLOR_DISABLED))
except curses.error:
pass
list_h = sh - 2
scroll = clamp(cursor - list_h // 2, 0, max(0, n_items - list_h))
row_y = 1
for i in range(scroll, min(scroll + list_h, n_items)):
if row_y >= sh - 1:
break
is_cur = (i == cursor)
sel_attr = curses.color_pair(COLOR_SELECTED if is_cur else COLOR_NORMAL)
if is_cert(i):
cert = certs[i]
box = "[X]" if cert.cn in checked else "[ ]"
lbl = _expiry_label(cert)
line = f" {box} {lbl:<18} {cert.cn}"
try:
stdscr.addstr(row_y, 0, line.ljust(sw - 1)[:sw - 1], sel_attr)
except curses.error:
pass
else:
item = menu_item(i)
any_check = bool(checked)
if item == _MENU_NEW and any_check:
notice = " ← clear all checkboxes to use"
line = f" {item}{notice}"
attr = curses.color_pair(COLOR_SELECTED if is_cur else COLOR_DISABLED)
else:
line = f" {item}"
attr = sel_attr
try:
stdscr.addstr(row_y, 0, line.ljust(sw - 1)[:sw - 1], attr)
except curses.error:
pass
row_y += 1
stdscr.refresh()
while True:
draw()
key = stdscr.getch()
if key in (ord("q"), ord("Q")):
return ScreenResult(action=Action.EXIT)
elif key == curses.KEY_UP:
cursor = clamp(cursor - 1, 0, n_items - 1)
elif key == curses.KEY_DOWN:
cursor = clamp(cursor + 1, 0, n_items - 1)
elif key == ord(" ") and is_cert(cursor):
cn = certs[cursor].cn
if cn in checked:
checked.discard(cn)
else:
checked.add(cn)
elif key in (10, 13, curses.KEY_ENTER):
if is_cert(cursor):
# Enter on a cert row: add to selection and confirm immediately
checked.add(certs[cursor].cn)
return ScreenResult(action=Action.RENEW_SELECTED,
selected_cns=sorted(checked))
else:
item = menu_item(cursor)
if item == _MENU_NEW and not checked:
return ScreenResult(action=Action.NEW_CERT)
elif item == _MENU_CRL:
return ScreenResult(action=Action.REGEN_CRL)
elif item == _MENU_EXIT:
return ScreenResult(action=Action.EXIT)
# _MENU_NEW when checked → grayed, ignore
elif key in (ord("r"), ord("R")) and is_cert(cursor):
cert = certs[cursor]
if show_confirm(
stdscr,
f"Revoke certificate for:\n CN: {cert.cn}\n\nThis cannot be undone.",
):
return ScreenResult(action=Action.REVOKE, selected_cns=[cert.cn])
```
- [ ] **Step 2: Commit**
```bash
git add openvpncertupdate.py
git commit -m "feat: TUI main selection screen with checkboxes, menu items, revoke hotkey"
```
---
## Task 10: App + Entry Point Sections
**Files:**
- Modify: `openvpncertupdate.py` (add `=== APP ===` section; replace `main()` stub)
- [ ] **Step 1: Replace the `=== APP ===` placeholder and `main()` in `openvpncertupdate.py`**
Find the block:
```python
# ============================================================
# (remaining sections added in later tasks)
# ============================================================
def main() -> None:
pass # replaced in Task 11
if __name__ == "__main__":
main()
```
Replace it with:
```python
# ============================================================
# === APP ===
# ============================================================
class CursesApp:
def run(self) -> None:
curses.wrapper(self._main)
def _main(self, stdscr) -> None:
curses.curs_set(0)
init_colors()
while True:
try:
certs = load_expiring_certs(
EASYRSA_PKI_DIR, DAYS_PAST, DAYS_AHEAD,
)
except FileNotFoundError as exc:
self._fatal(stdscr, f"Cannot read PKI index.txt:\n{exc}")
return
result = show_main_screen(stdscr, certs)
if result.action == Action.EXIT:
return
elif result.action == Action.REGEN_CRL:
self._regen_crl(stdscr)
elif result.action == Action.REVOKE:
self._do_revoke_and_crl(stdscr, result.selected_cns[0])
elif result.action == Action.NEW_CERT:
self._process_cert(stdscr, cn="", email="", is_renewal=False)
elif result.action == Action.RENEW_SELECTED:
for cn in result.selected_cns:
email_addr = get_email(EASYRSA_PKI_DIR, cn)
if not self._process_cert(stdscr, cn=cn, email=email_addr,
is_renewal=True):
break
# Loop back: reload cert list (it changed after renewal/revoke)
# ── Workflows ────────────────────────────────────────────────────────────
def _process_cert(
self, stdscr, cn: str, email: str, is_renewal: bool,
) -> bool:
"""Form → generate → deliver. Returns False if user cancelled."""
form = show_cert_form(stdscr, cn=cn, email=email, cn_readonly=is_renewal)
if form.cancelled:
return False
final_cn = form.cn
self._msg(stdscr, f"Generating certificate for {final_cn}…")
try:
if is_renewal:
revoke_issued(EASYRSA_DIR, EASYRSA_PKI_DIR, final_cn, CA_PASSPHRASE)
build_client_full(EASYRSA_DIR, EASYRSA_PKI_DIR,
final_cn, form.password, CA_PASSPHRASE)
except EasyRSAError as exc:
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)
except Exception as exc:
self._error(stdscr, f"Failed to build .ovpn:\n{exc}")
return True
try:
one_time_url = create_note(form.password, CRYPTGEON_URL)
except CryptgeonError as exc:
self._error(stdscr, f"Cryptgeon error:\n{exc}")
self._show_result(stdscr, final_cn, ovpn_path, form.password,
one_time_url="(Cryptgeon failed — see error above)")
return True
if form.email and show_confirm(
stdscr,
f"Send email to {form.email}?\n\n"
f"Attachment : {os.path.basename(ovpn_path)}\n"
f"Password URL: {one_time_url[:50]}…",
):
try:
send_email(form.email, final_cn, one_time_url, ovpn_path,
MAIL_FROM, MAIL_SUBJECT, EMAIL_TEMPLATE_PATH, MAIL_BINARY)
self._msg(stdscr,
f"Email sent to {form.email}\nConfig: {ovpn_path}",
wait=True)
except RuntimeError as exc:
self._error(stdscr,
f"Email failed:\n{exc}\n\n"
f"URL : {one_time_url}\n"
f"Config: {ovpn_path}")
else:
self._show_result(stdscr, final_cn, ovpn_path, form.password, one_time_url)
return True
def _do_revoke_and_crl(self, stdscr, cn: str) -> None:
self._msg(stdscr, f"Revoking {cn}…")
try:
revoke_issued(EASYRSA_DIR, EASYRSA_PKI_DIR, cn, CA_PASSPHRASE)
gen_crl(EASYRSA_DIR, EASYRSA_PKI_DIR, CA_PASSPHRASE)
copy_crl(EASYRSA_PKI_DIR, CRL_DEST_PATH)
except EasyRSAError as exc:
self._error(stdscr, f"Revoke failed:\n{exc}")
return
self._msg(stdscr, f"{cn} revoked and CRL updated → {CRL_DEST_PATH}", wait=True)
def _regen_crl(self, stdscr) -> None:
self._msg(stdscr, "Regenerating CRL…")
try:
gen_crl(EASYRSA_DIR, EASYRSA_PKI_DIR, CA_PASSPHRASE)
copy_crl(EASYRSA_PKI_DIR, CRL_DEST_PATH)
except EasyRSAError as exc:
self._error(stdscr, f"gen-crl failed:\n{exc}")
return
self._msg(stdscr, f"CRL regenerated → {CRL_DEST_PATH}", wait=True)
# ── Display helpers ───────────────────────────────────────────────────────
def _msg(self, stdscr, text: str, wait: bool = False) -> None:
stdscr.clear()
sh, sw = stdscr.getmaxyx()
lines = text.splitlines()
start_y = max(1, (sh - len(lines)) // 2)
for i, line in enumerate(lines):
try:
stdscr.addstr(start_y + i,
max(0, (sw - len(line)) // 2),
line[:sw], curses.color_pair(COLOR_NORMAL))
except curses.error:
pass
if wait:
try:
stdscr.addstr(sh - 2, 2, "Press any key to continue…",
curses.color_pair(COLOR_DISABLED))
except curses.error:
pass
stdscr.refresh()
if wait:
stdscr.getch()
def _error(self, stdscr, text: str) -> None:
lines = ["── ERROR ──", ""] + text.splitlines() + ["", "Press any key…"]
stdscr.clear()
sh, sw = stdscr.getmaxyx()
start_y = max(1, (sh - len(lines)) // 2)
for i, line in enumerate(lines):
attr = (curses.color_pair(COLOR_ERROR) | curses.A_BOLD
if i == 0 else curses.color_pair(COLOR_NORMAL))
try:
stdscr.addstr(start_y + i,
max(0, (sw - len(line)) // 2),
line[:sw], attr)
except curses.error:
pass
stdscr.refresh()
stdscr.getch()
def _fatal(self, stdscr, text: str) -> None:
self._error(stdscr, text)
def _show_result(
self, stdscr, cn: str, ovpn_path: str, password: str, one_time_url: str,
) -> None:
self._msg(
stdscr,
f"Certificate issued for: {cn}\n"
f"\n"
f"Config file:\n {ovpn_path}\n"
f"\n"
f"Password URL (one-time):\n {one_time_url}\n"
f"\n"
f"Password (shown once):\n {password}",
wait=True,
)
# ============================================================
# === ENTRY POINT ===
# ============================================================
def main() -> None:
CursesApp().run()
if __name__ == "__main__":
main()
```
- [ ] **Step 2: Run the full test suite**
```bash
python3 -m pytest tests/ -v
```
Expected: all tests PASS. Fix any import or namespace errors before committing.
- [ ] **Step 3: Verify import is clean**
```bash
python3 -c "import openvpncertupdate; print('ok')"
```
Expected: `ok` (no error; curses doesn't initialize at import time).
- [ ] **Step 4: Update CLAUDE.md**
```markdown
# CLAUDE.md
This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
## What this is
`openvpncertupdate` is a single-file Python + curses TUI tool for managing OpenVPN user certificates via EasyRSA 3.2.x. It lists expiring/expired certs, re-issues them with fresh keys, creates new certs, and delivers configs via Cryptgeon (one-time password URL) and email.
## Running
```bash
pip install -r requirements.txt # just: cryptography>=41
python3 openvpncertupdate.py
```
Edit the `SETTINGS` block at the top of `openvpncertupdate.py` before first run.
## Tests
```bash
python3 -m pytest tests/ -v
python3 -m pytest tests/test_password.py -v # single file
python3 -m pytest tests/test_pki.py::test_sorted_ascending -v # single test
```
## File layout — sections inside `openvpncertupdate.py`
| Section | Key symbols |
|---|---|
| SETTINGS | all-caps constants |
| PKI | `CertInfo`, `load_expiring_certs()`, `_parse_index_line()` |
| PASSWORD | `generate_password()` |
| EASYRSA | `EasyRSAError`, `revoke_issued()`, `build_client_full()`, `gen_crl()`, `copy_crl()` |
| METADATA | `load_metadata()`, `save_email()`, `get_email()` |
| CONFIG | `build_ovpn()` → `vpn-configs/<CN>_<date>/<CONFIG_NAME>` |
| CRYPTGEON | `CryptgeonError`, `create_note()` |
| MAILER | `send_email()` |
| TUI WIDGETS | `InputField`, `clamp()`, `draw_box()`, `init_colors()`, `COLOR_*` |
| TUI DIALOGS | `show_confirm()`, `show_cert_form()`, `CertFormResult` |
| TUI SCREEN | `show_main_screen()`, `Action`, `ScreenResult` |
| APP | `CursesApp` |
| ENTRY POINT | `main()` |
## Re-issue workflow
1. `revoke-issued <CN>` — archives old key + CSR to `pki/revoked/`
2. `build-client-full <CN> --passout=pass:<pw>` — generates new key + cert
3. CRL **not** auto-updated during renewal; use "Regenerate CRL" menu item or `r` hotkey
## Key constraints
- EasyRSA called with `--batch`; `--passin=pass:<CA_PASSPHRASE>` omitted when `CA_PASSPHRASE=""`
- Cryptgeon: `raw_key=os.urandom(32)`, `aes_key=SHA-256(raw_key)`, AES-256-GCM, URL fragment=`base64url(raw_key)`
- `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); `oO01lI` banned everywhere
- Inline file path: `<PKI_DIR>/inline/private/<CN>.inline`
- User emails stored in `<PKI_DIR>/openvpncertupdate-metadata.json`
```
- [ ] **Step 5: Final commit**
```bash
git add openvpncertupdate.py CLAUDE.md
git commit -m "feat: app orchestrator, entry point, CLAUDE.md — implementation complete"
```
---
## Spec Coverage
| Requirement | Section/Task |
|---|---|
| List expiring/expired certs on launch | PKI (T1), TUI SCREEN (T9) |
| Sorted by expiry date ascending | PKI (T1) |
| Checkboxes; Create New grayed when any checked | TUI SCREEN (T9) |
| Regenerate CRL menu item | TUI SCREEN (T9), APP (T10) |
| `r` hotkey to revoke + confirm | TUI SCREEN (T9) |
| Revoke-only: gen-crl + copy + chmod 644 | EASYRSA (T3), APP (T10) |
| Re-issue: revoke-issued → build-client-full (not renew) | EASYRSA (T3), APP (T10) |
| New cert: editable CN + email | TUI DIALOGS (T8) |
| Re-issue: read-only CN, editable email pre-filled | TUI DIALOGS (T8) |
| 28-char password with charset rules | PASSWORD (T2) |
| F5 to regenerate password in form | TUI DIALOGS (T8) |
| Concat inline + template → vpn-configs/CN_date/CONFIG_NAME | CONFIG (T4) |
| POST password to Cryptgeon → one-time URL | CRYPTGEON (T5) |
| Ask: send email or show URL + path | APP (T10) |
| Email: attach .ovpn + one-time URL via msmtp/sendmail | MAILER (T6) |
| If no email: display URL + config path | APP (T10) |
| Multiple renewals: process in order, return to list | APP (T10) |
| Configurable DAYS_PAST / DAYS_AHEAD | SETTINGS (T1) |
| All paths + CONFIG_NAME in settings | SETTINGS (T1) |
| Cryptgeon URL in settings | SETTINGS (T1) |
Reference in New Issue View Git Blame Copy Permalink