feat: project skeleton + settings + PKI index.txt reader
This commit is contained in:
137
openvpncertupdate.py
Normal file
137
openvpncertupdate.py
Normal file
@@ -0,0 +1,137 @@
|
||||
#!/usr/bin/env python3
|
||||
"""openvpncertupdate — Manage OpenVPN user certificates via EasyRSA."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import curses
|
||||
import email.encoders
|
||||
import email.mime.base
|
||||
import email.mime.multipart
|
||||
import email.mime.text
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import secrets
|
||||
import shutil
|
||||
import string
|
||||
import subprocess
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import date, datetime, timedelta, timezone
|
||||
from enum import Enum, auto
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
try:
|
||||
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
||||
except ImportError: # pragma: no cover
|
||||
AESGCM = None # type: ignore
|
||||
|
||||
# ============================================================
|
||||
# SETTINGS — edit these for your environment
|
||||
# ============================================================
|
||||
|
||||
EASYRSA_DIR = "/etc/easy-rsa"
|
||||
EASYRSA_PKI_DIR = "/etc/easy-rsa/pki"
|
||||
CA_PASSPHRASE = "" # empty string = no passphrase
|
||||
|
||||
OVPN_TEMPLATE_PATH = "./template.ovpn"
|
||||
CONFIG_NAME = "client.ovpn"
|
||||
VPN_CONFIGS_DIR = "./vpn-configs"
|
||||
|
||||
CRL_DEST_PATH = "/etc/openvpn/crl.pem"
|
||||
|
||||
CRYPTGEON_URL = "https://cryptgeon.example.com"
|
||||
|
||||
MAIL_FROM = "vpn-admin@example.com"
|
||||
MAIL_SUBJECT = "Your VPN Configuration"
|
||||
EMAIL_TEMPLATE_PATH = "./email_template.txt"
|
||||
MAIL_BINARY = "msmtp" # or "sendmail"
|
||||
|
||||
DAYS_PAST = 30
|
||||
DAYS_AHEAD = 14
|
||||
|
||||
# ============================================================
|
||||
# === PKI ===
|
||||
# ============================================================
|
||||
|
||||
|
||||
@dataclass
|
||||
class CertInfo:
|
||||
cn: str
|
||||
expires: datetime # UTC
|
||||
days_left: int # negative = already expired
|
||||
|
||||
|
||||
def _parse_date(raw: str) -> datetime:
|
||||
"""Parse OpenSSL date YYMMDDHHMMSSZ or YYYYMMDDHHMMSSZ."""
|
||||
raw = raw.rstrip("Z")
|
||||
if len(raw) == 12:
|
||||
yy = int(raw[:2])
|
||||
year = 2000 + yy if yy < 50 else 1900 + yy
|
||||
rest = raw[2:]
|
||||
dt = datetime.strptime(f"{year}{rest}", "%Y%m%d%H%M%S")
|
||||
else:
|
||||
dt = datetime.strptime(raw, "%Y%m%d%H%M%S")
|
||||
return dt.replace(tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def _parse_index_line(line: str) -> Optional[tuple[str, datetime]]:
|
||||
"""Return (cn, expiry) for valid (V-status) index.txt lines, else None."""
|
||||
parts = line.rstrip("\n").split("\t")
|
||||
if len(parts) < 6 or parts[0] != "V":
|
||||
return None
|
||||
try:
|
||||
expiry = _parse_date(parts[1])
|
||||
except ValueError:
|
||||
return None
|
||||
dn = parts[5]
|
||||
cn = None
|
||||
for seg in dn.split("/"):
|
||||
if seg.startswith("CN="):
|
||||
cn = seg[3:]
|
||||
break
|
||||
if not cn:
|
||||
return None
|
||||
return cn, expiry
|
||||
|
||||
|
||||
def load_expiring_certs(
|
||||
pki_dir: str,
|
||||
days_past: int,
|
||||
days_ahead: int,
|
||||
) -> list[CertInfo]:
|
||||
"""Return valid certs whose expiry falls within [-days_past, +days_ahead]."""
|
||||
now = datetime.now(tz=timezone.utc)
|
||||
cutoff_past = now - timedelta(days=days_past)
|
||||
cutoff_future = now + timedelta(days=days_ahead)
|
||||
results: list[CertInfo] = []
|
||||
with open(os.path.join(pki_dir, "index.txt")) as fh:
|
||||
for line in fh:
|
||||
parsed = _parse_index_line(line)
|
||||
if parsed is None:
|
||||
continue
|
||||
cn, expiry = parsed
|
||||
if cutoff_past <= expiry <= cutoff_future:
|
||||
results.append(CertInfo(
|
||||
cn=cn,
|
||||
expires=expiry,
|
||||
days_left=(expiry - now).days,
|
||||
))
|
||||
results.sort(key=lambda c: c.expires)
|
||||
return results
|
||||
|
||||
|
||||
# ============================================================
|
||||
# (remaining sections added in later tasks)
|
||||
# ============================================================
|
||||
|
||||
|
||||
def main() -> None:
|
||||
pass # replaced in Task 11
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
1
requirements.txt
Normal file
1
requirements.txt
Normal file
@@ -0,0 +1 @@
|
||||
cryptography>=41.0.0
|
||||
0
tests/__init__.py
Normal file
0
tests/__init__.py
Normal file
59
tests/test_pki.py
Normal file
59
tests/test_pki.py
Normal file
@@ -0,0 +1,59 @@
|
||||
import textwrap
|
||||
import pytest
|
||||
from datetime import datetime, timezone, timedelta
|
||||
from openvpncertupdate import load_expiring_certs, CertInfo, _parse_index_line
|
||||
|
||||
|
||||
def _fmt(dt: datetime) -> str:
|
||||
return dt.strftime("%y%m%d%H%M%S") + "Z"
|
||||
|
||||
|
||||
def make_pki(tmp_path, now):
|
||||
pki = tmp_path / "pki"
|
||||
pki.mkdir()
|
||||
content = textwrap.dedent(f"""\
|
||||
V\t{_fmt(now + timedelta(days=10))}\t\t02\tunknown\t/CN=soon
|
||||
V\t{_fmt(now + timedelta(days=90))}\t\t03\tunknown\t/CN=later
|
||||
V\t{_fmt(now - timedelta(days=15))}\t\t04\tunknown\t/CN=past15
|
||||
V\t{_fmt(now - timedelta(days=40))}\t\t05\tunknown\t/CN=past40
|
||||
R\t{_fmt(now + timedelta(days=10))}\t230101Z,keyCompromise\t06\tunknown\t/CN=revoked
|
||||
""")
|
||||
(pki / "index.txt").write_text(content)
|
||||
return str(pki)
|
||||
|
||||
|
||||
def test_filters_within_window(tmp_path):
|
||||
now = datetime.now(tz=timezone.utc)
|
||||
pki = make_pki(tmp_path, now)
|
||||
certs = load_expiring_certs(pki, days_past=30, days_ahead=14)
|
||||
cns = {c.cn for c in certs}
|
||||
assert "soon" in cns # 10d ahead < 14d threshold
|
||||
assert "later" not in cns # 90d > 14d
|
||||
assert "past15" in cns # 15d past < 30d threshold
|
||||
assert "past40" not in cns # 40d past > 30d threshold
|
||||
assert "revoked" not in cns # R status skipped
|
||||
|
||||
|
||||
def test_sorted_ascending(tmp_path):
|
||||
now = datetime.now(tz=timezone.utc)
|
||||
pki = make_pki(tmp_path, now)
|
||||
certs = load_expiring_certs(pki, days_past=30, days_ahead=14)
|
||||
dates = [c.expires for c in certs]
|
||||
assert dates == sorted(dates)
|
||||
|
||||
|
||||
def test_days_left_negative_for_expired(tmp_path):
|
||||
now = datetime.now(tz=timezone.utc)
|
||||
pki = make_pki(tmp_path, now)
|
||||
certs = load_expiring_certs(pki, days_past=30, days_ahead=14)
|
||||
expired = next(c for c in certs if c.cn == "past15")
|
||||
assert expired.days_left < 0
|
||||
|
||||
|
||||
def test_parse_4digit_year():
|
||||
line = "V\t20251231120000Z\t\t07\tunknown\t/CN=future"
|
||||
result = _parse_index_line(line)
|
||||
assert result is not None
|
||||
cn, dt = result
|
||||
assert cn == "future"
|
||||
assert dt.year == 2025
|
||||
Reference in New Issue
Block a user