feat: EasyRSA subprocess wrappers (revoke, build, gen-crl, copy-crl)
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -149,6 +149,64 @@ def generate_password() -> str:
|
||||
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)
|
||||
|
||||
|
||||
# (remaining sections added in later tasks)
|
||||
# ============================================================
|
||||
|
||||
|
||||
67
tests/test_easyrsa.py
Normal file
67
tests/test_easyrsa.py
Normal file
@@ -0,0 +1,67 @@
|
||||
import pytest
|
||||
from unittest.mock import patch, MagicMock
|
||||
from openvpncertupdate import (
|
||||
revoke_issued, build_client_full, gen_crl, copy_crl, EasyRSAError,
|
||||
)
|
||||
|
||||
|
||||
def ok_result():
|
||||
r = MagicMock(); r.returncode = 0; r.stderr = ""; return r
|
||||
|
||||
def err_result():
|
||||
r = MagicMock(); r.returncode = 1; r.stderr = "oops"; return r
|
||||
|
||||
|
||||
@patch("openvpncertupdate.subprocess.run")
|
||||
def test_revoke_includes_required_args(mock_run):
|
||||
mock_run.return_value = ok_result()
|
||||
revoke_issued("/er", "/pki", "alice", "capass")
|
||||
args = mock_run.call_args[0][0]
|
||||
assert args[0] == "/er/easyrsa"
|
||||
assert "--batch" in args
|
||||
assert "--pki=/pki" in args
|
||||
assert "--passin=pass:capass" in args
|
||||
assert "revoke-issued" in args
|
||||
assert "alice" in args
|
||||
|
||||
|
||||
@patch("openvpncertupdate.subprocess.run")
|
||||
def test_revoke_omits_passin_when_empty(mock_run):
|
||||
mock_run.return_value = ok_result()
|
||||
revoke_issued("/er", "/pki", "alice", "")
|
||||
args = mock_run.call_args[0][0]
|
||||
assert not any(a.startswith("--passin") for a in args)
|
||||
|
||||
|
||||
@patch("openvpncertupdate.subprocess.run")
|
||||
def test_revoke_raises_on_failure(mock_run):
|
||||
mock_run.return_value = err_result()
|
||||
with pytest.raises(EasyRSAError, match="revoke-issued"):
|
||||
revoke_issued("/er", "/pki", "alice", "")
|
||||
|
||||
|
||||
@patch("openvpncertupdate.subprocess.run")
|
||||
def test_build_client_full_args(mock_run):
|
||||
mock_run.return_value = ok_result()
|
||||
build_client_full("/er", "/pki", "bob", "keypass", "capass")
|
||||
args = mock_run.call_args[0][0]
|
||||
assert "--passout=pass:keypass" in args
|
||||
assert "--passin=pass:capass" in args
|
||||
assert "build-client-full" in args
|
||||
assert "bob" in args
|
||||
|
||||
|
||||
@patch("openvpncertupdate.subprocess.run")
|
||||
def test_gen_crl_args(mock_run):
|
||||
mock_run.return_value = ok_result()
|
||||
gen_crl("/er", "/pki", "capass")
|
||||
args = mock_run.call_args[0][0]
|
||||
assert "gen-crl" in args
|
||||
|
||||
|
||||
@patch("openvpncertupdate.shutil.copy2")
|
||||
@patch("openvpncertupdate.os.chmod")
|
||||
def test_copy_crl_copies_and_chmods(mock_chmod, mock_copy):
|
||||
copy_crl("/pki", "/etc/openvpn/crl.pem")
|
||||
mock_copy.assert_called_once_with("/pki/crl.pem", "/etc/openvpn/crl.pem")
|
||||
mock_chmod.assert_called_once_with("/etc/openvpn/crl.pem", 0o644)
|
||||
Reference in New Issue
Block a user