VaKU/openvpncertupdate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Restore Python 3.6 compatibility

Browse Source 
Remove Python 3.7+ constructs so the tool runs on 3.6:

- Drop `from __future__ import annotations` (3.7+) and convert all
  builtin-generic annotations (list[]/dict[]/tuple[]/set[]) to their
  typing.List/Dict/Tuple/Set equivalents, since without the future
  import these annotations are evaluated eagerly and builtin generic
  subscription only exists on 3.9+.
- Replace subprocess.run(capture_output=, text=) — both 3.7+ — with
  stdout/stderr=PIPE and universal_newlines=True; update the matching
  test assertion.
- requirements.txt: gate cryptography>=41 behind python_version>=3.7 and
  pin cryptography<41 (last 3.6-compatible line, 40.0.2) plus the
  dataclasses backport for 3.6.
- Suppress cryptography 40.x's per-import "Python 3.6 is no longer
  supported" deprecation warning so it doesn't pollute CLI/cron stderr;
  the <41 pin keeps us on a supported release regardless.

The only cryptography API used is AESGCM (present since 2.0), so no
x509/API-surface changes are needed for the older pin.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Vlad Doloman
2026-07-22 17:12:41 +03:00
parent 0477392b5e
commit b8158e260a
3 changed files with 42 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
openvpncertupdate.py
View File

@@ -1,8 +1,6 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""openvpncertupdate — Manage OpenVPN user certificates via EasyRSA.""" """openvpncertupdate — Manage OpenVPN user certificates via EasyRSA."""
from __future__ import annotations
import argparse import argparse
import base64 import base64
import curses import curses
@@ -21,13 +19,19 @@ import subprocess
import sys import sys
import urllib.error import urllib.error
import urllib.request import urllib.request
import warnings
from dataclasses import dataclass, field from dataclasses import dataclass, field
from datetime import date, datetime, timedelta, timezone from datetime import date, datetime, timedelta, timezone
from enum import Enum, auto from enum import Enum, auto
from pathlib import Path from pathlib import Path
from typing import Callable, Optional from typing import Callable, Dict, List, Optional, Set, Tuple
try: try:
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 from cryptography.hazmat.primitives.ciphers.aead import AESGCM
except ImportError: # pragma: no cover except ImportError: # pragma: no cover
AESGCM = None # type: ignore AESGCM = None # type: ignore
@@ -144,7 +148,7 @@ def _parse_date(raw: str) -> datetime:
return dt.replace(tzinfo=timezone.utc) 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.""" """Return (cn, expiry, email) for valid (V-status) index.txt lines, else None."""
parts = line.rstrip("\n").split("\t") parts = line.rstrip("\n").split("\t")
if len(parts) < 6 or parts[0] != "V": 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 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. """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 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. so expiring/all-certs views don't show stale duplicate rows.
""" """
now = datetime.now(tz=timezone.utc) 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: with open(os.path.join(pki_dir, "index.txt")) as fh:
for line in fh: for line in fh:
parsed = _parse_index_line(line) parsed = _parse_index_line(line)
@@ -197,7 +201,7 @@ def load_expiring_certs(
pki_dir: str, pki_dir: str,
days_past: int, days_past: int,
days_ahead: int, days_ahead: int,
) -> list[CertInfo]: ) -> List[CertInfo]:
"""Return current certs whose expiry falls within [-days_past, +days_ahead].""" """Return current certs whose expiry falls within [-days_past, +days_ahead]."""
now = datetime.now(tz=timezone.utc) now = datetime.now(tz=timezone.utc)
cutoff_past = now - timedelta(days=days_past) cutoff_past = now - timedelta(days=days_past)
@@ -208,7 +212,7 @@ def load_expiring_certs(
return results 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.""" """Return all current certs, sorted by expiry date."""
results = _load_current_certs(pki_dir) results = _load_current_certs(pki_dir)
results.sort(key=lambda c: c.expires) results.sort(key=lambda c: c.expires)
@@ -257,13 +261,16 @@ class EasyRSAError(Exception):
pass pass
def _run_easyrsa(cmd: list[str], cwd: str, def _run_easyrsa(cmd: List[str], cwd: str,
extra_env: Optional[dict[str, str]] = None) -> None: extra_env: Optional[Dict[str, str]] = None) -> None:
env = None env = None
if extra_env: if extra_env:
env = os.environ.copy() env = os.environ.copy()
env.update(extra_env) 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: if result.returncode != 0:
# Find the first non-flag argument after the binary (the actual easyrsa verb) # 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") 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}"] cmd = [f"{easyrsa_dir}/easyrsa", "--batch", f"--pki={pki_dir}"]
if ca_passphrase: if ca_passphrase:
cmd.append(f"--passin=pass:{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 # Best-effort, like is_ca_key_encrypted(): a missing/misconfigured
# restorecon must not break CRL deployment on non-SELinux systems. # restorecon must not break CRL deployment on non-SELinux systems.
try: 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: except OSError:
pass pass
@@ -727,7 +738,7 @@ def show_cert_form(
active = ([n for n in _FORM_FIELDS if not (n == "cn" and cn_readonly)] active = ([n for n in _FORM_FIELDS if not (n == "cn" and cn_readonly)]
+ [_BTN_CONTINUE, _BTN_CANCEL]) + [_BTN_CONTINUE, _BTN_CANCEL])
fields: dict[str, InputField] = { fields: Dict[str, InputField] = {
"cn": InputField(win, _FORM_FIELD_Y["cn"], 3, fw, initial=cn), "cn": InputField(win, _FORM_FIELD_Y["cn"], 3, fw, initial=cn),
"email": InputField(win, _FORM_FIELD_Y["email"], 3, fw, initial=email), "email": InputField(win, _FORM_FIELD_Y["email"], 3, fw, initial=email),
"password": InputField(win, _FORM_FIELD_Y["password"], 3, fw, "password": InputField(win, _FORM_FIELD_Y["password"], 3, fw,
@@ -857,8 +868,8 @@ class Action(Enum):
@dataclass @dataclass
class ScreenResult: class ScreenResult:
action: Action action: Action
selected_cns: list[str] = field(default_factory=list) selected_cns: List[str] = field(default_factory=list)
saved_checked: list[str] = field(default_factory=list) saved_checked: List[str] = field(default_factory=list)
saved_cursor: int = 0 saved_cursor: int = 0
@@ -877,8 +888,8 @@ def _expiry_label(c: CertInfo) -> str:
def show_main_screen( def show_main_screen(
stdscr, certs: list[CertInfo], show_all: bool = False, stdscr, certs: List[CertInfo], show_all: bool = False,
initial_checked: Optional[set[str]] = None, initial_checked: Optional[Set[str]] = None,
initial_cursor: int = 0, initial_cursor: int = 0,
) -> ScreenResult: ) -> ScreenResult:
"""Display cert selection list. Returns when user activates an action.""" """Display cert selection list. Returns when user activates an action."""
@@ -887,7 +898,7 @@ def show_main_screen(
n_certs = len(certs) n_certs = len(certs)
n_items = n_certs + len(_MENU_ITEMS) 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)) cursor = clamp(initial_cursor, 0, max(0, n_items - 1))
esc_pending = False esc_pending = False
@@ -1037,7 +1048,7 @@ def show_main_screen(
class CursesApp: class CursesApp:
def __init__(self) -> None: def __init__(self) -> None:
self._session_log: list[str] = [] self._session_log: List[str] = []
def _log(self, entry: str) -> None: def _log(self, entry: str) -> None:
self._session_log.append(entry) self._session_log.append(entry)
@@ -1062,7 +1073,7 @@ class CursesApp:
init_colors() init_colors()
show_all = False show_all = False
saved_checked: list[str] = [] saved_checked: List[str] = []
saved_cursor: int = 0 saved_cursor: int = 0
while True: 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`.""" """Render CN / expiry / email as an aligned table, like `ls -l`."""
if not certs: if not certs:
return "(no certificates)" return "(no certificates)"

4
requirements.txt
View File

@@ -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"

6
tests/test_easyrsa.py
View File

@@ -1,3 +1,5 @@
import subprocess
import pytest import pytest
from unittest.mock import patch, MagicMock from unittest.mock import patch, MagicMock
from openvpncertupdate import ( 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): 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") copy_crl("/pki", "/etc/openvpn/crl.pem", restorecon_binary="restorecon")
mock_run.assert_called_once_with( 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")) @patch("openvpncertupdate.subprocess.run", side_effect=FileNotFoundError("no restorecon"))