598 lines
18 KiB
Python
598 lines
18 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")):
|
|
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)
|
|
|
|
|
|
# (remaining sections added in later tasks)
|
|
# ============================================================
|
|
|
|
|
|
def main() -> None:
|
|
pass # replaced in Task 11
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|