Files
openvpncertupdate/openvpncertupdate.py
Vlad Doloman 565a152243 fix: spec compliance for TUI DIALOGS section
- Rename CertFormResult.cancelled → confirmed (True=submit, False=cancel/Esc)
- Change _FORM_FIELDS from list to tuple
- Update _FORM_LABELS to exact values: CN, Email, Password
- Update _FORM_FIELD_Y to {cn:3, email:5, password:7}
- Add password param to show_cert_form (uses generate_password() if empty)
- Add mask=True to password InputField (both init and F5-regen sites)
- Add ord("Q") to show_confirm false-return condition
- Add KEY_UP/KEY_DOWN navigation in show_cert_form
- Add Ctrl-S (0x13) as immediate submit key in show_cert_form
- Update tests/test_dialogs.py: confirmed=True on submit, confirmed=False on cancel

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-24 00:13:46 +03:00

612 lines
19 KiB
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
# ============================================================
# === 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)
# ============================================================
# === 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:
# Find the first non-flag argument after the binary (the actual easyrsa verb)
verb = next((c for c in cmd[1:] if not c.startswith("-")), "unknown")
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)
# ============================================================
# === 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())
# ============================================================
# === 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}"
# ============================================================
# === 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()}"
)
# ============================================================
# === 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
# ============================================================
# === 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"), ord("Q")):
return False
@dataclass
class CertFormResult:
cn: str
email: str
password: str
confirmed: bool
_FORM_FIELDS = ("cn", "email", "password")
_FORM_LABELS = {
"cn": "CN",
"email": "Email",
"password": "Password",
}
_FORM_FIELD_Y = {"cn": 3, "email": 5, "password": 7}
def show_cert_form(
stdscr,
cn: str = "",
email: str = "",
password: str = "",
cn_readonly: bool = False,
) -> CertFormResult:
"""Modal cert-detail form.
Tab / Shift-Tab / Up / Down cycle fields. F5 regenerates password.
Enter on last field or Ctrl-S 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=password if password else generate_password(),
mask=True),
}
focus = 0
def _submit() -> CertFormResult:
final_cn = cn if cn_readonly else fields["cn"].value
return CertFormResult(
cn=final_cn,
email=fields["email"].value,
password=fields["password"].value,
confirmed=True,
)
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 Ctrl-S=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="", confirmed=False)
if key == 0x13: # Ctrl-S: immediate submit
return _submit()
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_DOWN:
focus = (focus + 1) % len(active)
continue
if key == curses.KEY_UP:
focus = (focus - 1) % len(active)
continue
if key == curses.KEY_F5:
fields["password"] = InputField(
win, _FORM_FIELD_Y["password"], 3, fw,
initial=generate_password(), mask=True,
)
continue
if key in (10, 13, curses.KEY_ENTER):
if focus < len(active) - 1:
focus += 1
continue
return _submit()
fields[active[focus]].handle_key(key)
# (remaining sections added in later tasks)
# ============================================================
def main() -> None:
pass # replaced in Task 11
if __name__ == "__main__":
main()