feat: metadata store (CN→email) and .ovpn config builder
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -207,6 +207,56 @@ def copy_crl(pki_dir: str, dest_path: str) -> None:
|
|||||||
os.chmod(dest_path, 0o644)
|
os.chmod(dest_path, 0o644)
|
||||||
|
|
||||||
|
|
||||||
|
# ============================================================
|
||||||
|
# === METADATA ===
|
||||||
|
# ============================================================
|
||||||
|
|
||||||
|
_METADATA_FILE = "openvpncertupdate-metadata.json"
|
||||||
|
|
||||||
|
|
||||||
|
def load_metadata(pki_dir: str) -> dict[str, str]:
|
||||||
|
p = os.path.join(pki_dir, _METADATA_FILE)
|
||||||
|
if not os.path.exists(p):
|
||||||
|
return {}
|
||||||
|
with open(p) as fh:
|
||||||
|
return json.load(fh)
|
||||||
|
|
||||||
|
|
||||||
|
def save_email(pki_dir: str, cn: str, email_addr: str) -> None:
|
||||||
|
data = load_metadata(pki_dir)
|
||||||
|
data[cn] = email_addr
|
||||||
|
with open(os.path.join(pki_dir, _METADATA_FILE), "w") as fh:
|
||||||
|
json.dump(data, fh, indent=2)
|
||||||
|
|
||||||
|
|
||||||
|
def get_email(pki_dir: str, cn: str) -> str:
|
||||||
|
return load_metadata(pki_dir).get(cn, "")
|
||||||
|
|
||||||
|
|
||||||
|
# ============================================================
|
||||||
|
# === CONFIG ===
|
||||||
|
# ============================================================
|
||||||
|
|
||||||
|
|
||||||
|
def build_ovpn(
|
||||||
|
cn: str,
|
||||||
|
pki_dir: str,
|
||||||
|
template_path: str,
|
||||||
|
output_base_dir: str,
|
||||||
|
config_name: str,
|
||||||
|
) -> str:
|
||||||
|
"""Concat template + inline file. Return absolute path to written .ovpn."""
|
||||||
|
inline_path = os.path.join(pki_dir, "inline", "private", f"{cn}.inline")
|
||||||
|
template_content = Path(template_path).read_text()
|
||||||
|
inline_content = Path(inline_path).read_text()
|
||||||
|
today = date.today().strftime("%Y-%m-%d")
|
||||||
|
out_dir = Path(output_base_dir) / f"{cn}_{today}"
|
||||||
|
out_dir.mkdir(parents=True, exist_ok=True)
|
||||||
|
out_file = out_dir / config_name
|
||||||
|
out_file.write_text(template_content + "\n" + inline_content)
|
||||||
|
return str(out_file.resolve())
|
||||||
|
|
||||||
|
|
||||||
# (remaining sections added in later tasks)
|
# (remaining sections added in later tasks)
|
||||||
# ============================================================
|
# ============================================================
|
||||||
|
|
||||||
|
|||||||
28
tests/test_config_builder.py
Normal file
28
tests/test_config_builder.py
Normal file
@@ -0,0 +1,28 @@
|
|||||||
|
import os
|
||||||
|
from pathlib import Path
|
||||||
|
from datetime import date
|
||||||
|
from openvpncertupdate import build_ovpn
|
||||||
|
|
||||||
|
def setup_pki(tmp_path, cn, inline_content="<ca>X</ca>"):
|
||||||
|
pki = tmp_path / "pki"
|
||||||
|
private_dir = pki / "inline" / "private"
|
||||||
|
private_dir.mkdir(parents=True)
|
||||||
|
(private_dir / f"{cn}.inline").write_text(inline_content)
|
||||||
|
return str(pki)
|
||||||
|
|
||||||
|
def test_creates_file_with_combined_content(tmp_path):
|
||||||
|
pki = setup_pki(tmp_path, "alice", "<ca>CERT</ca>")
|
||||||
|
tmpl = tmp_path / "t.ovpn"
|
||||||
|
tmpl.write_text("client\ndev tun\n")
|
||||||
|
result = build_ovpn("alice", pki, str(tmpl), str(tmp_path / "out"), "vpn.ovpn")
|
||||||
|
content = Path(result).read_text()
|
||||||
|
assert "client" in content
|
||||||
|
assert "CERT" in content
|
||||||
|
|
||||||
|
def test_output_path_has_cn_and_date(tmp_path):
|
||||||
|
pki = setup_pki(tmp_path, "bob")
|
||||||
|
tmpl = tmp_path / "t.ovpn"; tmpl.write_text("")
|
||||||
|
today = date.today().strftime("%Y-%m-%d")
|
||||||
|
result = build_ovpn("bob", pki, str(tmpl), str(tmp_path / "out"), "vpn.ovpn")
|
||||||
|
assert f"bob_{today}" in result
|
||||||
|
assert result.endswith("vpn.ovpn")
|
||||||
16
tests/test_metadata.py
Normal file
16
tests/test_metadata.py
Normal file
@@ -0,0 +1,16 @@
|
|||||||
|
from openvpncertupdate import load_metadata, save_email, get_email
|
||||||
|
|
||||||
|
def test_load_empty(tmp_path):
|
||||||
|
assert load_metadata(str(tmp_path)) == {}
|
||||||
|
|
||||||
|
def test_save_and_get(tmp_path):
|
||||||
|
save_email(str(tmp_path), "alice", "alice@example.com")
|
||||||
|
assert get_email(str(tmp_path), "alice") == "alice@example.com"
|
||||||
|
|
||||||
|
def test_update_existing(tmp_path):
|
||||||
|
save_email(str(tmp_path), "alice", "old@example.com")
|
||||||
|
save_email(str(tmp_path), "alice", "new@example.com")
|
||||||
|
assert get_email(str(tmp_path), "alice") == "new@example.com"
|
||||||
|
|
||||||
|
def test_missing_returns_empty(tmp_path):
|
||||||
|
assert get_email(str(tmp_path), "nobody") == ""
|
||||||
Reference in New Issue
Block a user