openvpncertupdate/openvpncertupdate.py

#!/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()}"
        )


# (remaining sections added in later tasks)
# ============================================================


def main() -> None:
    pass  # replaced in Task 11


if __name__ == "__main__":
    main()