diff --git a/openvpncertupdate.py b/openvpncertupdate.py index 36cfeb4..f0434a2 100644 --- a/openvpncertupdate.py +++ b/openvpncertupdate.py @@ -1,8 +1,6 @@ #!/usr/bin/env python3 """openvpncertupdate — Manage OpenVPN user certificates via EasyRSA.""" -from __future__ import annotations - import argparse import base64 import curses @@ -21,14 +19,20 @@ import subprocess import sys import urllib.error import urllib.request +import warnings from dataclasses import dataclass, field from datetime import date, datetime, timedelta, timezone from enum import Enum, auto from pathlib import Path -from typing import Callable, Optional +from typing import Callable, Dict, List, Optional, Set, Tuple try: - from cryptography.hazmat.primitives.ciphers.aead import AESGCM + with warnings.catch_warnings(): + # cryptography <41 pinned for Python 3.6 warns on every import that 3.6 + # support is deprecated; the pin keeps us on a supported release, so the + # warning is noise that would otherwise pollute CLI/cron stderr each run. + warnings.filterwarnings("ignore", message="Python 3.6 is no longer supported") + from cryptography.hazmat.primitives.ciphers.aead import AESGCM except ImportError: # pragma: no cover AESGCM = None # type: ignore @@ -144,7 +148,7 @@ def _parse_date(raw: str) -> datetime: return dt.replace(tzinfo=timezone.utc) -def _parse_index_line(line: str) -> Optional[tuple[str, datetime, str]]: +def _parse_index_line(line: str) -> Optional[Tuple[str, datetime, str]]: """Return (cn, expiry, email) for valid (V-status) index.txt lines, else None.""" parts = line.rstrip("\n").split("\t") if len(parts) < 6 or parts[0] != "V": @@ -166,7 +170,7 @@ def _parse_index_line(line: str) -> Optional[tuple[str, datetime, str]]: return cn, expiry, email -def _load_current_certs(pki_dir: str) -> list[CertInfo]: +def _load_current_certs(pki_dir: str) -> List[CertInfo]: """Return one CertInfo per CN: its most recent V-status index.txt entry. index.txt is append-only, so a CN can accumulate several V-status lines @@ -177,7 +181,7 @@ def _load_current_certs(pki_dir: str) -> list[CertInfo]: so expiring/all-certs views don't show stale duplicate rows. """ now = datetime.now(tz=timezone.utc) - by_cn: dict[str, CertInfo] = {} + by_cn: Dict[str, CertInfo] = {} with open(os.path.join(pki_dir, "index.txt")) as fh: for line in fh: parsed = _parse_index_line(line) @@ -197,7 +201,7 @@ def load_expiring_certs( pki_dir: str, days_past: int, days_ahead: int, -) -> list[CertInfo]: +) -> List[CertInfo]: """Return current certs whose expiry falls within [-days_past, +days_ahead].""" now = datetime.now(tz=timezone.utc) cutoff_past = now - timedelta(days=days_past) @@ -208,7 +212,7 @@ def load_expiring_certs( return results -def load_all_certs(pki_dir: str) -> list[CertInfo]: +def load_all_certs(pki_dir: str) -> List[CertInfo]: """Return all current certs, sorted by expiry date.""" results = _load_current_certs(pki_dir) results.sort(key=lambda c: c.expires) @@ -257,13 +261,16 @@ class EasyRSAError(Exception): pass -def _run_easyrsa(cmd: list[str], cwd: str, - extra_env: Optional[dict[str, str]] = None) -> None: +def _run_easyrsa(cmd: List[str], cwd: str, + extra_env: Optional[Dict[str, str]] = None) -> None: env = None if extra_env: env = os.environ.copy() env.update(extra_env) - result = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True, env=env) + result = subprocess.run( + cmd, cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + universal_newlines=True, env=env, + ) 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") @@ -272,7 +279,7 @@ def _run_easyrsa(cmd: list[str], cwd: str, ) -def _base_cmd(easyrsa_dir: str, pki_dir: str, ca_passphrase: str) -> list[str]: +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}") @@ -315,7 +322,11 @@ def copy_crl(pki_dir: str, dest_path: str, restorecon_binary: str = "") -> None: # Best-effort, like is_ca_key_encrypted(): a missing/misconfigured # restorecon must not break CRL deployment on non-SELinux systems. try: - subprocess.run([restorecon_binary, dest_path], capture_output=True, text=True) + subprocess.run( + [restorecon_binary, dest_path], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, + universal_newlines=True, + ) except OSError: pass @@ -727,7 +738,7 @@ def show_cert_form( active = ([n for n in _FORM_FIELDS if not (n == "cn" and cn_readonly)] + [_BTN_CONTINUE, _BTN_CANCEL]) - fields: dict[str, InputField] = { + fields: Dict[str, InputField] = { "cn": InputField(win, _FORM_FIELD_Y["cn"], 3, fw, initial=cn), "email": InputField(win, _FORM_FIELD_Y["email"], 3, fw, initial=email), "password": InputField(win, _FORM_FIELD_Y["password"], 3, fw, @@ -857,8 +868,8 @@ class Action(Enum): @dataclass class ScreenResult: action: Action - selected_cns: list[str] = field(default_factory=list) - saved_checked: list[str] = field(default_factory=list) + selected_cns: List[str] = field(default_factory=list) + saved_checked: List[str] = field(default_factory=list) saved_cursor: int = 0 @@ -877,8 +888,8 @@ def _expiry_label(c: CertInfo) -> str: def show_main_screen( - stdscr, certs: list[CertInfo], show_all: bool = False, - initial_checked: Optional[set[str]] = None, + stdscr, certs: List[CertInfo], show_all: bool = False, + initial_checked: Optional[Set[str]] = None, initial_cursor: int = 0, ) -> ScreenResult: """Display cert selection list. Returns when user activates an action.""" @@ -887,7 +898,7 @@ def show_main_screen( n_certs = len(certs) n_items = n_certs + len(_MENU_ITEMS) - checked: set[str] = set(initial_checked) if initial_checked else set() + checked: Set[str] = set(initial_checked) if initial_checked else set() cursor = clamp(initial_cursor, 0, max(0, n_items - 1)) esc_pending = False @@ -1037,7 +1048,7 @@ def show_main_screen( class CursesApp: def __init__(self) -> None: - self._session_log: list[str] = [] + self._session_log: List[str] = [] def _log(self, entry: str) -> None: self._session_log.append(entry) @@ -1062,7 +1073,7 @@ class CursesApp: init_colors() show_all = False - saved_checked: list[str] = [] + saved_checked: List[str] = [] saved_cursor: int = 0 while True: @@ -1297,7 +1308,7 @@ class CursesApp: # ============================================================ -def _format_cert_table(certs: list[CertInfo]) -> str: +def _format_cert_table(certs: List[CertInfo]) -> str: """Render CN / expiry / email as an aligned table, like `ls -l`.""" if not certs: return "(no certificates)" diff --git a/requirements.txt b/requirements.txt index 11d3e7f..29aa242 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1 +1,3 @@ -cryptography>=41.0.0 +cryptography>=41.0.0; python_version >= "3.7" +cryptography>=3.4,<41; python_version < "3.7" +dataclasses>=0.8; python_version < "3.7" diff --git a/tests/test_easyrsa.py b/tests/test_easyrsa.py index 5961e50..2e96da2 100644 --- a/tests/test_easyrsa.py +++ b/tests/test_easyrsa.py @@ -1,3 +1,5 @@ +import subprocess + import pytest from unittest.mock import patch, MagicMock from openvpncertupdate import ( @@ -82,7 +84,9 @@ def test_copy_crl_skips_restorecon_when_binary_empty(mock_chmod, mock_copy, mock 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"], capture_output=True, text=True) + ["restorecon", "/etc/openvpn/crl.pem"], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, + universal_newlines=True) @patch("openvpncertupdate.subprocess.run", side_effect=FileNotFoundError("no restorecon"))