|
|
|
|
@@ -1,8 +1,6 @@
|
|
|
|
|
#!/usr/bin/env python3
|
|
|
|
|
"""openvpncertupdate — Manage OpenVPN user certificates via EasyRSA."""
|
|
|
|
|
|
|
|
|
|
from __future__ import annotations
|
|
|
|
|
|
|
|
|
|
import argparse
|
|
|
|
|
import base64
|
|
|
|
|
import curses
|
|
|
|
|
@@ -21,13 +19,19 @@ import subprocess
|
|
|
|
|
import sys
|
|
|
|
|
import urllib.error
|
|
|
|
|
import urllib.request
|
|
|
|
|
import warnings
|
|
|
|
|
from dataclasses import dataclass, field
|
|
|
|
|
from datetime import date, datetime, timedelta, timezone
|
|
|
|
|
from enum import Enum, auto
|
|
|
|
|
from pathlib import Path
|
|
|
|
|
from typing import Callable, Optional
|
|
|
|
|
from typing import Callable, Dict, List, Optional, Set, Tuple
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
|
with warnings.catch_warnings():
|
|
|
|
|
# cryptography <41 pinned for Python 3.6 warns on every import that 3.6
|
|
|
|
|
# support is deprecated; the pin keeps us on a supported release, so the
|
|
|
|
|
# warning is noise that would otherwise pollute CLI/cron stderr each run.
|
|
|
|
|
warnings.filterwarnings("ignore", message="Python 3.6 is no longer supported")
|
|
|
|
|
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
|
|
|
|
except ImportError: # pragma: no cover
|
|
|
|
|
AESGCM = None # type: ignore
|
|
|
|
|
@@ -144,7 +148,7 @@ def _parse_date(raw: str) -> datetime:
|
|
|
|
|
return dt.replace(tzinfo=timezone.utc)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _parse_index_line(line: str) -> Optional[tuple[str, datetime, str]]:
|
|
|
|
|
def _parse_index_line(line: str) -> Optional[Tuple[str, datetime, str]]:
|
|
|
|
|
"""Return (cn, expiry, email) for valid (V-status) index.txt lines, else None."""
|
|
|
|
|
parts = line.rstrip("\n").split("\t")
|
|
|
|
|
if len(parts) < 6 or parts[0] != "V":
|
|
|
|
|
@@ -166,7 +170,7 @@ def _parse_index_line(line: str) -> Optional[tuple[str, datetime, str]]:
|
|
|
|
|
return cn, expiry, email
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _load_current_certs(pki_dir: str) -> list[CertInfo]:
|
|
|
|
|
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
|
|
|
|
|
@@ -177,7 +181,7 @@ def _load_current_certs(pki_dir: str) -> list[CertInfo]:
|
|
|
|
|
so expiring/all-certs views don't show stale duplicate rows.
|
|
|
|
|
"""
|
|
|
|
|
now = datetime.now(tz=timezone.utc)
|
|
|
|
|
by_cn: dict[str, CertInfo] = {}
|
|
|
|
|
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)
|
|
|
|
|
@@ -197,7 +201,7 @@ def load_expiring_certs(
|
|
|
|
|
pki_dir: str,
|
|
|
|
|
days_past: int,
|
|
|
|
|
days_ahead: int,
|
|
|
|
|
) -> list[CertInfo]:
|
|
|
|
|
) -> List[CertInfo]:
|
|
|
|
|
"""Return current certs whose expiry falls within [-days_past, +days_ahead]."""
|
|
|
|
|
now = datetime.now(tz=timezone.utc)
|
|
|
|
|
cutoff_past = now - timedelta(days=days_past)
|
|
|
|
|
@@ -208,7 +212,7 @@ def load_expiring_certs(
|
|
|
|
|
return results
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def load_all_certs(pki_dir: str) -> list[CertInfo]:
|
|
|
|
|
def load_all_certs(pki_dir: str) -> List[CertInfo]:
|
|
|
|
|
"""Return all current certs, sorted by expiry date."""
|
|
|
|
|
results = _load_current_certs(pki_dir)
|
|
|
|
|
results.sort(key=lambda c: c.expires)
|
|
|
|
|
@@ -257,13 +261,16 @@ class EasyRSAError(Exception):
|
|
|
|
|
pass
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _run_easyrsa(cmd: list[str], cwd: str,
|
|
|
|
|
extra_env: Optional[dict[str, str]] = None) -> None:
|
|
|
|
|
def _run_easyrsa(cmd: List[str], cwd: str,
|
|
|
|
|
extra_env: Optional[Dict[str, str]] = None) -> None:
|
|
|
|
|
env = None
|
|
|
|
|
if extra_env:
|
|
|
|
|
env = os.environ.copy()
|
|
|
|
|
env.update(extra_env)
|
|
|
|
|
result = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True, env=env)
|
|
|
|
|
result = subprocess.run(
|
|
|
|
|
cmd, cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
|
|
|
|
universal_newlines=True, env=env,
|
|
|
|
|
)
|
|
|
|
|
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")
|
|
|
|
|
@@ -272,7 +279,7 @@ def _run_easyrsa(cmd: list[str], cwd: str,
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _base_cmd(easyrsa_dir: str, pki_dir: str, ca_passphrase: str) -> list[str]:
|
|
|
|
|
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}")
|
|
|
|
|
@@ -315,7 +322,11 @@ def copy_crl(pki_dir: str, dest_path: str, restorecon_binary: str = "") -> None:
|
|
|
|
|
# Best-effort, like is_ca_key_encrypted(): a missing/misconfigured
|
|
|
|
|
# restorecon must not break CRL deployment on non-SELinux systems.
|
|
|
|
|
try:
|
|
|
|
|
subprocess.run([restorecon_binary, dest_path], capture_output=True, text=True)
|
|
|
|
|
subprocess.run(
|
|
|
|
|
[restorecon_binary, dest_path],
|
|
|
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
|
|
|
|
universal_newlines=True,
|
|
|
|
|
)
|
|
|
|
|
except OSError:
|
|
|
|
|
pass
|
|
|
|
|
|
|
|
|
|
@@ -727,7 +738,7 @@ def show_cert_form(
|
|
|
|
|
|
|
|
|
|
active = ([n for n in _FORM_FIELDS if not (n == "cn" and cn_readonly)]
|
|
|
|
|
+ [_BTN_CONTINUE, _BTN_CANCEL])
|
|
|
|
|
fields: dict[str, InputField] = {
|
|
|
|
|
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,
|
|
|
|
|
@@ -857,8 +868,8 @@ class Action(Enum):
|
|
|
|
|
@dataclass
|
|
|
|
|
class ScreenResult:
|
|
|
|
|
action: Action
|
|
|
|
|
selected_cns: list[str] = field(default_factory=list)
|
|
|
|
|
saved_checked: list[str] = field(default_factory=list)
|
|
|
|
|
selected_cns: List[str] = field(default_factory=list)
|
|
|
|
|
saved_checked: List[str] = field(default_factory=list)
|
|
|
|
|
saved_cursor: int = 0
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@@ -877,8 +888,8 @@ def _expiry_label(c: CertInfo) -> str:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def show_main_screen(
|
|
|
|
|
stdscr, certs: list[CertInfo], show_all: bool = False,
|
|
|
|
|
initial_checked: Optional[set[str]] = None,
|
|
|
|
|
stdscr, certs: List[CertInfo], show_all: bool = False,
|
|
|
|
|
initial_checked: Optional[Set[str]] = None,
|
|
|
|
|
initial_cursor: int = 0,
|
|
|
|
|
) -> ScreenResult:
|
|
|
|
|
"""Display cert selection list. Returns when user activates an action."""
|
|
|
|
|
@@ -887,7 +898,7 @@ def show_main_screen(
|
|
|
|
|
|
|
|
|
|
n_certs = len(certs)
|
|
|
|
|
n_items = n_certs + len(_MENU_ITEMS)
|
|
|
|
|
checked: set[str] = set(initial_checked) if initial_checked else set()
|
|
|
|
|
checked: Set[str] = set(initial_checked) if initial_checked else set()
|
|
|
|
|
cursor = clamp(initial_cursor, 0, max(0, n_items - 1))
|
|
|
|
|
esc_pending = False
|
|
|
|
|
|
|
|
|
|
@@ -1037,7 +1048,7 @@ def show_main_screen(
|
|
|
|
|
|
|
|
|
|
class CursesApp:
|
|
|
|
|
def __init__(self) -> None:
|
|
|
|
|
self._session_log: list[str] = []
|
|
|
|
|
self._session_log: List[str] = []
|
|
|
|
|
|
|
|
|
|
def _log(self, entry: str) -> None:
|
|
|
|
|
self._session_log.append(entry)
|
|
|
|
|
@@ -1062,7 +1073,7 @@ class CursesApp:
|
|
|
|
|
init_colors()
|
|
|
|
|
|
|
|
|
|
show_all = False
|
|
|
|
|
saved_checked: list[str] = []
|
|
|
|
|
saved_checked: List[str] = []
|
|
|
|
|
saved_cursor: int = 0
|
|
|
|
|
|
|
|
|
|
while True:
|
|
|
|
|
@@ -1297,7 +1308,7 @@ class CursesApp:
|
|
|
|
|
# ============================================================
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _format_cert_table(certs: list[CertInfo]) -> str:
|
|
|
|
|
def _format_cert_table(certs: List[CertInfo]) -> str:
|
|
|
|
|
"""Render CN / expiry / email as an aligned table, like `ls -l`."""
|
|
|
|
|
if not certs:
|
|
|
|
|
return "(no certificates)"
|
|
|
|
|
|