feat: CLI mode (--create, --reissue, --revoke, --gen-crl)
This commit is contained in:
Vlad Doloman
@@ -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,12 +976,135 @@ 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:
|
||||
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()
|
||||
|
||||
|
||||
|
||||
200
tests/test_cli.py
Normal file
200
tests/test_cli.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user