From ceaf24414a9974746e2a1c0d1cf4a1f66a9f8fd3 Mon Sep 17 00:00:00 2001 From: Vlad Doloman Date: Wed, 24 Jun 2026 02:53:47 +0300 Subject: [PATCH] feat: CLI mode (--create, --reissue, --revoke, --gen-crl) --- openvpncertupdate.py | 127 ++++++++++++++++++++++++++- tests/test_cli.py | 200 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 326 insertions(+), 1 deletion(-) create mode 100644 tests/test_cli.py diff --git a/openvpncertupdate.py b/openvpncertupdate.py index 4760266..7f734c7 100644 --- a/openvpncertupdate.py +++ b/openvpncertupdate.py @@ -3,6 +3,7 @@ from __future__ import annotations +import argparse import base64 import curses import email.encoders @@ -16,6 +17,7 @@ import secrets import shutil import string import subprocess +import sys import urllib.error import urllib.request from dataclasses import dataclass, field @@ -974,13 +976,136 @@ class CursesApp: ) +# ============================================================ +# === CLI === +# ============================================================ + + +class CliRunner: + """Non-interactive runner for scripting / cron use.""" + + def create(self, cn: str, email_addr: str) -> None: + self._issue(cn, email_addr, is_renewal=False) + + def reissue(self, cn: str, email_addr: str) -> None: + # Prefer the caller-supplied email; fall back to stored metadata. + final_email = email_addr or get_email(EASYRSA_PKI_DIR, cn) + self._issue(cn, final_email, is_renewal=True) + + def revoke(self, cn: str) -> None: + print(f"Revoking {cn}…", file=sys.stderr) + 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: + print(f"error: {exc}", file=sys.stderr) + sys.exit(1) + print(f"Revoked {cn}. CRL updated → {CRL_DEST_PATH}") + + def regen_crl(self) -> None: + print("Regenerating CRL…", file=sys.stderr) + try: + gen_crl(EASYRSA_DIR, EASYRSA_PKI_DIR, CA_PASSPHRASE) + copy_crl(EASYRSA_PKI_DIR, CRL_DEST_PATH) + except EasyRSAError as exc: + print(f"error: {exc}", file=sys.stderr) + sys.exit(1) + print(f"CRL updated → {CRL_DEST_PATH}") + + def _issue(self, cn: str, email_addr: str, is_renewal: bool) -> None: + password = generate_password() + + if is_renewal: + print(f"Revoking existing cert for {cn}…", file=sys.stderr) + try: + revoke_issued(EASYRSA_DIR, EASYRSA_PKI_DIR, cn, CA_PASSPHRASE) + except EasyRSAError as exc: + print(f"error during revoke: {exc}", file=sys.stderr) + sys.exit(1) + + print(f"Building certificate for {cn}…", file=sys.stderr) + try: + build_client_full(EASYRSA_DIR, EASYRSA_PKI_DIR, cn, password, CA_PASSPHRASE) + except EasyRSAError as exc: + if is_renewal: + print( + f"error during build-client-full: {exc}\n" + f"WARNING: {cn} has been revoked but no new cert was built.\n" + f"Re-run --reissue for this CN to issue a new certificate.", + file=sys.stderr, + ) + else: + print(f"error: {exc}", file=sys.stderr) + sys.exit(1) + + if email_addr: + save_email(EASYRSA_PKI_DIR, cn, email_addr) + + try: + ovpn_path = build_ovpn(cn, EASYRSA_PKI_DIR, OVPN_TEMPLATE_PATH, + VPN_CONFIGS_DIR, CONFIG_NAME) + except Exception as exc: + print(f"error building .ovpn: {exc}", file=sys.stderr) + sys.exit(1) + + one_time_url = "" + try: + one_time_url = create_note(password, CRYPTGEON_URL) + except CryptgeonError as exc: + print(f"warning: Cryptgeon failed: {exc}", file=sys.stderr) + + if email_addr: + try: + send_email(email_addr, cn, one_time_url or password, ovpn_path, + MAIL_FROM, MAIL_SUBJECT, EMAIL_TEMPLATE_PATH, MAIL_BINARY) + print(f"Email sent to {email_addr}", file=sys.stderr) + except RuntimeError as exc: + print(f"warning: email failed: {exc}", file=sys.stderr) + + print(f"config: {ovpn_path}") + if one_time_url: + print(f"password-url: {one_time_url}") + else: + print(f"password: {password}") + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="openvpncertupdate", + description="Manage OpenVPN user certificates via EasyRSA.", + ) + group = parser.add_mutually_exclusive_group() + group.add_argument("--create", metavar="CN", help="Create a new certificate for CN") + group.add_argument("--reissue", metavar="CN", help="Revoke and reissue certificate for CN") + group.add_argument("--revoke", metavar="CN", help="Revoke certificate for CN and regenerate CRL") + group.add_argument("--gen-crl", action="store_true", help="Regenerate and copy CRL") + parser.add_argument("--email", metavar="EMAIL", + help="Email address (required for --create; optional for --reissue)") + return parser + + # ============================================================ # === ENTRY POINT === # ============================================================ def main() -> None: - CursesApp().run() + parser = _build_parser() + args = parser.parse_args() + + if args.create: + if not args.email: + parser.error("--email is required with --create") + CliRunner().create(args.create, args.email) + elif args.reissue: + CliRunner().reissue(args.reissue, args.email or "") + elif args.revoke: + CliRunner().revoke(args.revoke) + elif args.gen_crl: + CliRunner().regen_crl() + else: + CursesApp().run() if __name__ == "__main__": diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..a87f63e --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,200 @@ +"""Tests for CLI mode (CliRunner + main() argument dispatch).""" +import sys +from unittest.mock import MagicMock, patch, call +import pytest + +from openvpncertupdate import CliRunner, _build_parser, main + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _patch_issue(monkeypatch, ovpn_path="/out/cn_2026-01-01/client.ovpn", + one_time_url="https://cg.example.com/#/note/abc/xyz"): + """Patch all side-effectful callables used by CliRunner._issue.""" + monkeypatch.setattr("openvpncertupdate.revoke_issued", MagicMock()) + monkeypatch.setattr("openvpncertupdate.build_client_full", MagicMock()) + monkeypatch.setattr("openvpncertupdate.save_email", MagicMock()) + monkeypatch.setattr("openvpncertupdate.build_ovpn", MagicMock(return_value=ovpn_path)) + monkeypatch.setattr("openvpncertupdate.create_note", MagicMock(return_value=one_time_url)) + monkeypatch.setattr("openvpncertupdate.send_email", MagicMock()) + monkeypatch.setattr("openvpncertupdate.generate_password", MagicMock(return_value="Testpass1234567890abcdefgh")) + return { + "revoke_issued": sys.modules["openvpncertupdate"].revoke_issued, + "build_client_full": sys.modules["openvpncertupdate"].build_client_full, + "save_email": sys.modules["openvpncertupdate"].save_email, + "build_ovpn": sys.modules["openvpncertupdate"].build_ovpn, + "create_note": sys.modules["openvpncertupdate"].create_note, + "send_email": sys.modules["openvpncertupdate"].send_email, + } + + +# --------------------------------------------------------------------------- +# CliRunner.create +# --------------------------------------------------------------------------- + +def test_create_calls_build_not_revoke(monkeypatch, capsys): + mocks = _patch_issue(monkeypatch) + CliRunner().create("alice", "alice@example.com") + mocks["revoke_issued"].assert_not_called() + mocks["build_client_full"].assert_called_once() + _, kwargs = mocks["build_client_full"].call_args + # cn is the third positional arg + assert mocks["build_client_full"].call_args.args[2] == "alice" + + +def test_create_saves_email(monkeypatch, capsys): + mocks = _patch_issue(monkeypatch) + CliRunner().create("alice", "alice@example.com") + mocks["save_email"].assert_called_once() + assert mocks["save_email"].call_args.args[1] == "alice" + assert mocks["save_email"].call_args.args[2] == "alice@example.com" + + +def test_create_sends_email_and_prints_url(monkeypatch, capsys): + url = "https://cg.example.com/#/note/abc/xyz" + mocks = _patch_issue(monkeypatch, one_time_url=url) + CliRunner().create("alice", "alice@example.com") + mocks["send_email"].assert_called_once() + out = capsys.readouterr().out + assert "password-url:" in out + assert url in out + assert "password:" not in out + + +def test_create_prints_password_when_cryptgeon_fails(monkeypatch, capsys): + _patch_issue(monkeypatch) + import openvpncertupdate + monkeypatch.setattr("openvpncertupdate.create_note", + MagicMock(side_effect=openvpncertupdate.CryptgeonError("down"))) + CliRunner().create("alice", "alice@example.com") + out = capsys.readouterr().out + assert "password:" in out + assert "password-url:" not in out + + +# --------------------------------------------------------------------------- +# CliRunner.reissue +# --------------------------------------------------------------------------- + +def test_reissue_calls_revoke_then_build(monkeypatch, capsys): + mocks = _patch_issue(monkeypatch) + CliRunner().reissue("bob", "bob@example.com") + mocks["revoke_issued"].assert_called_once() + mocks["build_client_full"].assert_called_once() + # revoke must happen before build + assert mocks["revoke_issued"].call_args.args[2] == "bob" + assert mocks["build_client_full"].call_args.args[2] == "bob" + + +def test_reissue_falls_back_to_stored_email(monkeypatch, capsys): + mocks = _patch_issue(monkeypatch) + monkeypatch.setattr("openvpncertupdate.get_email", + MagicMock(return_value="stored@example.com")) + CliRunner().reissue("bob", "") # no --email supplied + mocks["send_email"].assert_called_once() + assert mocks["send_email"].call_args.args[0] == "stored@example.com" + + +def test_reissue_no_email_no_send(monkeypatch, capsys): + mocks = _patch_issue(monkeypatch) + monkeypatch.setattr("openvpncertupdate.get_email", MagicMock(return_value="")) + CliRunner().reissue("bob", "") + mocks["send_email"].assert_not_called() + out = capsys.readouterr().out + assert "config:" in out + + +# --------------------------------------------------------------------------- +# CliRunner.revoke +# --------------------------------------------------------------------------- + +def test_revoke_calls_revoke_gen_crl_copy_crl(monkeypatch, capsys): + rev = MagicMock() + gcrl = MagicMock() + ccrl = MagicMock() + monkeypatch.setattr("openvpncertupdate.revoke_issued", rev) + monkeypatch.setattr("openvpncertupdate.gen_crl", gcrl) + monkeypatch.setattr("openvpncertupdate.copy_crl", ccrl) + CliRunner().revoke("charlie") + rev.assert_called_once() + gcrl.assert_called_once() + ccrl.assert_called_once() + assert "charlie" in capsys.readouterr().out + + +def test_revoke_exits_on_error(monkeypatch): + import openvpncertupdate + monkeypatch.setattr("openvpncertupdate.revoke_issued", + MagicMock(side_effect=openvpncertupdate.EasyRSAError("fail"))) + with pytest.raises(SystemExit) as exc_info: + CliRunner().revoke("charlie") + assert exc_info.value.code == 1 + + +# --------------------------------------------------------------------------- +# CliRunner.regen_crl +# --------------------------------------------------------------------------- + +def test_regen_crl_calls_gen_and_copy(monkeypatch, capsys): + gcrl = MagicMock() + ccrl = MagicMock() + monkeypatch.setattr("openvpncertupdate.gen_crl", gcrl) + monkeypatch.setattr("openvpncertupdate.copy_crl", ccrl) + CliRunner().regen_crl() + gcrl.assert_called_once() + ccrl.assert_called_once() + assert "CRL" in capsys.readouterr().out + + +# --------------------------------------------------------------------------- +# main() argument dispatch +# --------------------------------------------------------------------------- + +def test_main_create_requires_email(monkeypatch): + monkeypatch.setattr(sys, "argv", ["prog", "--create", "alice"]) + with pytest.raises(SystemExit) as exc_info: + main() + assert exc_info.value.code != 0 + + +def test_main_dispatches_create(monkeypatch): + monkeypatch.setattr(sys, "argv", + ["prog", "--create", "alice", "--email", "a@b.com"]) + runner = MagicMock() + monkeypatch.setattr("openvpncertupdate.CliRunner", lambda: runner) + main() + runner.create.assert_called_once_with("alice", "a@b.com") + + +def test_main_dispatches_reissue(monkeypatch): + monkeypatch.setattr(sys, "argv", ["prog", "--reissue", "bob"]) + runner = MagicMock() + monkeypatch.setattr("openvpncertupdate.CliRunner", lambda: runner) + main() + runner.reissue.assert_called_once_with("bob", "") + + +def test_main_dispatches_revoke(monkeypatch): + monkeypatch.setattr(sys, "argv", ["prog", "--revoke", "charlie"]) + runner = MagicMock() + monkeypatch.setattr("openvpncertupdate.CliRunner", lambda: runner) + main() + runner.revoke.assert_called_once_with("charlie") + + +def test_main_dispatches_gen_crl(monkeypatch): + monkeypatch.setattr(sys, "argv", ["prog", "--gen-crl"]) + runner = MagicMock() + monkeypatch.setattr("openvpncertupdate.CliRunner", lambda: runner) + main() + runner.regen_crl.assert_called_once_with() + + +def test_main_no_args_launches_tui(monkeypatch): + monkeypatch.setattr(sys, "argv", ["prog"]) + app = MagicMock() + monkeypatch.setattr("openvpncertupdate.CursesApp", lambda: app) + main() + app.run.assert_called_once_with()