336 lines
13 KiB
Python
336 lines
13 KiB
Python
import subprocess
|
|
|
|
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
from openvpncertupdate import (
|
|
revoke_issued, build_client_full, gen_crl, copy_crl, EasyRSAError,
|
|
is_ca_key_encrypted, resolve_ca_passphrase, resolve_cert_days, ConfigError,
|
|
has_issued_cert, issued_cert_path,
|
|
)
|
|
|
|
|
|
def ok_result():
|
|
r = MagicMock(); r.returncode = 0; r.stdout = ""; r.stderr = ""; return r
|
|
|
|
def err_result(stdout="", stderr="oops"):
|
|
r = MagicMock(); r.returncode = 1; r.stdout = stdout; r.stderr = stderr; return r
|
|
|
|
|
|
# EasyRSA's print() is `printf '%s\n'` -> stdout, and both user_error() and
|
|
# die() route through it, so this is what a real failure looks like on the wire.
|
|
EASYRSA_USER_ERROR = """
|
|
EasyRSA version 3.2.6
|
|
|
|
Error
|
|
-----
|
|
Unable to revoke as no certificate was found.
|
|
Certificate was expected at:
|
|
* /etc/easy-rsa/pki/issued/alice.crt
|
|
"""
|
|
|
|
|
|
@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_error_reports_easyrsa_stdout_diagnostics(mock_run):
|
|
# EasyRSA writes its diagnostics to stdout; reporting stderr alone left the
|
|
# user with a blank error box.
|
|
mock_run.return_value = err_result(stdout=EASYRSA_USER_ERROR, stderr="")
|
|
with pytest.raises(EasyRSAError) as exc:
|
|
revoke_issued("/er", "/pki", "alice", "")
|
|
msg = str(exc.value)
|
|
assert "Unable to revoke as no certificate was found." in msg
|
|
assert "* /etc/easy-rsa/pki/issued/alice.crt" in msg
|
|
|
|
|
|
@patch("openvpncertupdate.subprocess.run")
|
|
def test_error_strips_banner_and_blank_padding(mock_run):
|
|
mock_run.return_value = err_result(stdout=EASYRSA_USER_ERROR, stderr="")
|
|
with pytest.raises(EasyRSAError) as exc:
|
|
revoke_issued("/er", "/pki", "alice", "")
|
|
lines = str(exc.value).splitlines()
|
|
assert "" not in lines # no blank padding
|
|
assert not any(l.startswith("EasyRSA version") for l in lines)
|
|
assert "-----" not in lines # no rule under "Error"
|
|
|
|
|
|
@patch("openvpncertupdate.subprocess.run")
|
|
def test_error_keeps_both_streams(mock_run):
|
|
# openssl failures land on stderr while EasyRSA's die() text lands on stdout.
|
|
mock_run.return_value = err_result(
|
|
stdout="Easy-RSA error:\n\nFailed to revoke certificate.",
|
|
stderr="unable to load CA private key",
|
|
)
|
|
with pytest.raises(EasyRSAError) as exc:
|
|
revoke_issued("/er", "/pki", "alice", "")
|
|
msg = str(exc.value)
|
|
assert "Failed to revoke certificate." in msg
|
|
assert "unable to load CA private key" in msg
|
|
|
|
|
|
@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)
|
|
|
|
|
|
@patch("openvpncertupdate.subprocess.run")
|
|
@patch("openvpncertupdate.shutil.copy2")
|
|
@patch("openvpncertupdate.os.chmod")
|
|
def test_copy_crl_skips_restorecon_when_binary_empty(mock_chmod, mock_copy, mock_run):
|
|
copy_crl("/pki", "/etc/openvpn/crl.pem", restorecon_binary="")
|
|
mock_run.assert_not_called()
|
|
|
|
|
|
@patch("openvpncertupdate.subprocess.run")
|
|
@patch("openvpncertupdate.shutil.copy2")
|
|
@patch("openvpncertupdate.os.chmod")
|
|
def test_copy_crl_runs_restorecon_when_binary_set(mock_chmod, mock_copy, mock_run):
|
|
copy_crl("/pki", "/etc/openvpn/crl.pem", restorecon_binary="restorecon")
|
|
mock_run.assert_called_once_with(
|
|
["restorecon", "/etc/openvpn/crl.pem"],
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
|
universal_newlines=True)
|
|
|
|
|
|
@patch("openvpncertupdate.subprocess.run", side_effect=FileNotFoundError("no restorecon"))
|
|
@patch("openvpncertupdate.shutil.copy2")
|
|
@patch("openvpncertupdate.os.chmod")
|
|
def test_copy_crl_restorecon_failure_is_non_fatal(mock_chmod, mock_copy, mock_run):
|
|
# Missing/misconfigured restorecon must not break CRL deployment.
|
|
copy_crl("/pki", "/etc/openvpn/crl.pem", restorecon_binary="restorecon") # must not raise
|
|
mock_copy.assert_called_once()
|
|
mock_chmod.assert_called_once()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# has_issued_cert
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_issued_cert_path(tmp_path):
|
|
assert issued_cert_path("/pki", "y.kuts") == "/pki/issued/y.kuts.crt"
|
|
|
|
|
|
def test_has_issued_cert_true_when_file_present(tmp_path):
|
|
issued = tmp_path / "issued"
|
|
issued.mkdir()
|
|
(issued / "y.kuts.crt").write_text("-----BEGIN CERTIFICATE-----\n")
|
|
assert has_issued_cert(str(tmp_path), "y.kuts") is True
|
|
|
|
|
|
def test_has_issued_cert_false_when_index_lists_cert_but_file_is_gone(tmp_path):
|
|
(tmp_path / "issued").mkdir()
|
|
assert has_issued_cert(str(tmp_path), "y.kuts") is False
|
|
|
|
|
|
def test_has_issued_cert_false_when_issued_dir_missing(tmp_path):
|
|
assert has_issued_cert(str(tmp_path), "y.kuts") is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# is_ca_key_encrypted
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _write_ca_key(pki_dir, content):
|
|
private = pki_dir / "private"
|
|
private.mkdir(parents=True, exist_ok=True)
|
|
(private / "ca.key").write_text(content)
|
|
|
|
|
|
def test_is_ca_key_encrypted_true_for_pkcs8_encrypted_header(tmp_path):
|
|
_write_ca_key(tmp_path, "-----BEGIN ENCRYPTED PRIVATE KEY-----\nAAA\n-----END ENCRYPTED PRIVATE KEY-----\n")
|
|
assert is_ca_key_encrypted(str(tmp_path)) is True
|
|
|
|
|
|
def test_is_ca_key_encrypted_true_for_legacy_proc_type_header(tmp_path):
|
|
_write_ca_key(tmp_path, "-----BEGIN RSA PRIVATE KEY-----\nProc-Type: 4,ENCRYPTED\nDEK-Info: AES-256-CBC,...\n\nAAA\n-----END RSA PRIVATE KEY-----\n")
|
|
assert is_ca_key_encrypted(str(tmp_path)) is True
|
|
|
|
|
|
def test_is_ca_key_encrypted_false_for_plain_key(tmp_path):
|
|
_write_ca_key(tmp_path, "-----BEGIN PRIVATE KEY-----\nAAA\n-----END PRIVATE KEY-----\n")
|
|
assert is_ca_key_encrypted(str(tmp_path)) is False
|
|
|
|
|
|
def test_is_ca_key_encrypted_false_when_key_missing(tmp_path):
|
|
assert is_ca_key_encrypted(str(tmp_path)) is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# resolve_ca_passphrase
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_resolve_empty_string_prompts_when_ca_is_encrypted(tmp_path):
|
|
_write_ca_key(tmp_path, "-----BEGIN ENCRYPTED PRIVATE KEY-----\nAAA\n-----END ENCRYPTED PRIVATE KEY-----\n")
|
|
prompt = MagicMock(return_value="typed-pass")
|
|
result = resolve_ca_passphrase("", str(tmp_path), prompt=prompt)
|
|
assert result == "typed-pass"
|
|
prompt.assert_called_once()
|
|
|
|
|
|
def test_resolve_empty_string_skips_prompt_when_ca_not_encrypted(tmp_path):
|
|
_write_ca_key(tmp_path, "-----BEGIN PRIVATE KEY-----\nAAA\n-----END PRIVATE KEY-----\n")
|
|
prompt = MagicMock()
|
|
result = resolve_ca_passphrase("", str(tmp_path), prompt=prompt)
|
|
assert result == ""
|
|
prompt.assert_not_called()
|
|
|
|
|
|
def test_resolve_empty_sentinel_never_checks_or_prompts(tmp_path):
|
|
_write_ca_key(tmp_path, "-----BEGIN ENCRYPTED PRIVATE KEY-----\nAAA\n-----END ENCRYPTED PRIVATE KEY-----\n")
|
|
prompt = MagicMock()
|
|
result = resolve_ca_passphrase("!empty", str(tmp_path), prompt=prompt)
|
|
assert result == ""
|
|
prompt.assert_not_called()
|
|
|
|
|
|
def test_resolve_ask_sentinel_always_prompts_without_checking(tmp_path):
|
|
# no ca.key on disk at all -- must not be checked/read for "!ask"
|
|
prompt = MagicMock(return_value="typed-pass")
|
|
result = resolve_ca_passphrase("!ask", str(tmp_path), prompt=prompt)
|
|
assert result == "typed-pass"
|
|
prompt.assert_called_once()
|
|
|
|
|
|
def test_resolve_literal_passphrase_passed_through_unchanged(tmp_path):
|
|
prompt = MagicMock()
|
|
result = resolve_ca_passphrase("mysecret", str(tmp_path), prompt=prompt)
|
|
assert result == "mysecret"
|
|
prompt.assert_not_called()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# resolve_cert_days
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.parametrize("configured", ["default", "", " ", " default "])
|
|
def test_resolve_cert_days_sentinels_mean_inherit(configured):
|
|
# "" and "default" leave the lifetime to EasyRSA's own EASYRSA_CERT_EXPIRE,
|
|
# so deploying this cannot silently shorten certs on an existing PKI.
|
|
assert resolve_cert_days(configured) == ""
|
|
|
|
|
|
@pytest.mark.parametrize("configured,expected", [
|
|
(90, "90"), ("90", "90"), (" 90 ", "90"), (1, "1"), (3650, "3650"),
|
|
])
|
|
def test_resolve_cert_days_accepts_positive_numbers(configured, expected):
|
|
assert resolve_cert_days(configured) == expected
|
|
|
|
|
|
def test_resolve_cert_days_normalises_leading_zeros():
|
|
# EasyRSA's own pattern rejects "090" with "Number expected", which reads
|
|
# as nonsense to someone who typed only digits.
|
|
assert resolve_cert_days("090") == "90"
|
|
|
|
|
|
def test_resolve_cert_days_rejects_zero():
|
|
with pytest.raises(ConfigError, match="positive"):
|
|
resolve_cert_days("0")
|
|
|
|
|
|
@pytest.mark.parametrize("configured", ["-5", -5, "abc", "9 0", "90.5", 90.5, None])
|
|
def test_resolve_cert_days_rejects_junk(configured):
|
|
with pytest.raises(ConfigError):
|
|
resolve_cert_days(configured)
|
|
|
|
|
|
def test_resolve_cert_days_error_names_the_source():
|
|
# The same resolver serves CERT_DAYS, --days and the TUI field; the message
|
|
# must say which one the user actually touched.
|
|
with pytest.raises(ConfigError, match="--days"):
|
|
resolve_cert_days("0", "--days")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# build_client_full with days parameter
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@patch("openvpncertupdate.subprocess.run")
|
|
def test_build_client_full_includes_days_when_set(mock_run):
|
|
mock_run.return_value = ok_result()
|
|
build_client_full("/er", "/pki", "bob", "keypass", "capass", days="90")
|
|
assert "--days=90" in mock_run.call_args[0][0]
|
|
|
|
|
|
@patch("openvpncertupdate.subprocess.run")
|
|
def test_build_client_full_omits_days_when_inheriting(mock_run):
|
|
mock_run.return_value = ok_result()
|
|
build_client_full("/er", "/pki", "bob", "keypass", "capass", days="")
|
|
assert not any(a.startswith("--days") for a in mock_run.call_args[0][0])
|
|
|
|
|
|
@patch("openvpncertupdate.subprocess.run")
|
|
def test_build_client_full_days_precedes_the_verb(mock_run):
|
|
# --days is a global option: EasyRSA parses it before the command word.
|
|
mock_run.return_value = ok_result()
|
|
build_client_full("/er", "/pki", "bob", "keypass", "capass", days="90")
|
|
args = mock_run.call_args[0][0]
|
|
assert args.index("--days=90") < args.index("build-client-full")
|
|
|
|
|
|
@patch("openvpncertupdate.subprocess.run")
|
|
def test_revoke_and_gen_crl_never_carry_days(mock_run):
|
|
# _base_cmd() is shared with gen-crl, where --days means CRL validity.
|
|
# Putting the flag there would quietly change how long CRLs are valid.
|
|
mock_run.return_value = ok_result()
|
|
revoke_issued("/er", "/pki", "alice", "capass")
|
|
assert not any(a.startswith("--days") for a in mock_run.call_args[0][0])
|
|
gen_crl("/er", "/pki", "capass")
|
|
assert not any(a.startswith("--days") for a in mock_run.call_args[0][0])
|
|
|
|
|
|
@patch("openvpncertupdate.subprocess.run")
|
|
def test_error_verb_detection_survives_days_flag(mock_run):
|
|
# _run_easyrsa picks the verb as the first non-flag arg; "--days=90"
|
|
# must not be mistaken for it.
|
|
mock_run.return_value = err_result(stdout="Error\nboom", stderr="")
|
|
with pytest.raises(EasyRSAError, match="build-client-full"):
|
|
build_client_full("/er", "/pki", "bob", "keypass", "capass", days="90")
|