fix Cryptgeon protocol and add CA passphrase / external config resolution
Cryptgeon's real protocol (verified against upstream occulto/frontend/backend
source) never matched what create_note() sent: it used a SHA-256-derived key
instead of the raw one, a single-blob ciphertext instead of the delimited
AES-GCM--nonce--ciphertext format, an empty meta field instead of a JSON
string, and a hash-bang URL instead of the real /note/<id>#<key> route -
notes uploaded fine but were undecryptable in the browser.
Also adds CA_PASSPHRASE auto-detection/prompting ("" auto-detects an
encrypted CA key and prompts, "!empty"/"!ask" opt out of/force the prompt)
and an external .conf file (--config / CONFIG_PATH / <script>.conf) that can
override the SETTINGS block without editing the script.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
10
CLAUDE.md
10
CLAUDE.md
@@ -31,6 +31,7 @@ Edit the `SETTINGS` block at the top of `openvpncertupdate.py` before first run.
|
|||||||
| `--send-email` | Force email delivery |
|
| `--send-email` | Force email delivery |
|
||||||
| `--no-send-email` | Skip email; print URL to stdout |
|
| `--no-send-email` | Skip email; print URL to stdout |
|
||||||
| `--show-eml` | Print base64-encoded `.eml` to stdout (implies `--no-send-email` unless `--send-email` also given) |
|
| `--show-eml` | Print base64-encoded `.eml` to stdout (implies `--no-send-email` unless `--send-email` also given) |
|
||||||
|
| `--config PATH` | External `.conf` file overriding `SETTINGS` (overrides `CONFIG_PATH`; missing file here is an error) |
|
||||||
|
|
||||||
## Tests
|
## Tests
|
||||||
|
|
||||||
@@ -45,9 +46,10 @@ python3 -m pytest tests/test_pki.py::test_sorted_ascending -v # single test
|
|||||||
| Section | Key symbols |
|
| Section | Key symbols |
|
||||||
|---|---|
|
|---|---|
|
||||||
| SETTINGS | all-caps constants |
|
| SETTINGS | all-caps constants |
|
||||||
|
| SETTINGS OVERRIDE | `ConfigError`, `load_settings_overrides()`, `_OVERRIDABLE_SETTINGS` |
|
||||||
| PKI | `CertInfo`, `load_expiring_certs()`, `load_all_certs()`, `_parse_index_line()` |
|
| PKI | `CertInfo`, `load_expiring_certs()`, `load_all_certs()`, `_parse_index_line()` |
|
||||||
| PASSWORD | `generate_password()` |
|
| PASSWORD | `generate_password()` |
|
||||||
| EASYRSA | `EasyRSAError`, `revoke_issued()`, `build_client_full()`, `gen_crl()`, `copy_crl()` |
|
| EASYRSA | `EasyRSAError`, `revoke_issued()`, `build_client_full()`, `gen_crl()`, `copy_crl()`, `is_ca_key_encrypted()`, `resolve_ca_passphrase()` |
|
||||||
| METADATA | `load_metadata()`, `save_email()`, `get_email()` |
|
| METADATA | `load_metadata()`, `save_email()`, `get_email()` |
|
||||||
| CONFIG | `build_ovpn()` → `vpn-configs/<CN>_<YYYY-MM-DD>_<NN>/CONFIG_NAME` |
|
| CONFIG | `build_ovpn()` → `vpn-configs/<CN>_<YYYY-MM-DD>_<NN>/CONFIG_NAME` |
|
||||||
| CRYPTGEON | `CryptgeonError`, `create_note()` |
|
| CRYPTGEON | `CryptgeonError`, `create_note()` |
|
||||||
@@ -78,9 +80,11 @@ python3 -m pytest tests/test_pki.py::test_sorted_ascending -v # single test
|
|||||||
|
|
||||||
## Key constraints
|
## Key constraints
|
||||||
|
|
||||||
|
- External config file (`load_settings_overrides()`, run once in `main()` right after arg parsing, before dispatch): resolution order is `--config PATH` > `CONFIG_PATH` setting > `<this-script-path>.conf` next to the script. The CLI flag or `CONFIG_PATH` make the path explicit — a missing file there is a fatal `ConfigError`; the default `<script>.conf` path is optional and silently skipped if absent. The file is executed as Python (same syntax as the `SETTINGS` block, so only run trusted files) and only names listed in `_OVERRIDABLE_SETTINGS` are applied — `CONFIG_PATH` itself is deliberately not overridable this way
|
||||||
- Email: set `SMTP_HOST` to use smtplib (SMTP_TLS: `"starttls"`/`"ssl"`/`""`); leave empty to use `MAIL_BINARY`. Auth skipped when `SMTP_USER=""`
|
- Email: set `SMTP_HOST` to use smtplib (SMTP_TLS: `"starttls"`/`"ssl"`/`""`); leave empty to use `MAIL_BINARY`. Auth skipped when `SMTP_USER=""`
|
||||||
- EasyRSA called with `--batch`; `--passin=pass:<CA_PASSPHRASE>` omitted when `CA_PASSPHRASE=""`
|
- EasyRSA called with `--batch`; `--passin=pass:<passphrase>` omitted when the resolved passphrase is empty
|
||||||
- Cryptgeon: `raw_key=os.urandom(32)`, `aes_key=SHA-256(raw_key)`, AES-256-GCM, URL fragment=`base64url(raw_key)`
|
- CA passphrase resolution (`resolve_ca_passphrase()`, run once in `main()` right after arg parsing, before dispatch): `CA_PASSPHRASE=""` → auto-detect via `is_ca_key_encrypted()` (checks `<PKI_DIR>/private/ca.key` PEM header for `ENCRYPTED`) and prompt only if encrypted; `"!empty"` → never check/prompt, passphrase is `""`; `"!ask"` → always prompt, skip detection; any other value → used literally
|
||||||
|
- Cryptgeon: matches the `occulto` browser client — `key=os.urandom(32)` used directly (no derivation) for AES-256-GCM; `contents` = `base64(b"AES-GCM") + "--" + base64(nonce) + "--" + base64(ciphertext)`; `meta` = JSON string `{"type": "text"}`; URL = `<base>/note/<id>#<key.hex()>`
|
||||||
- `copy_crl()` does `chmod 644` after copy
|
- `copy_crl()` does `chmod 644` after copy
|
||||||
- Password: pos 1=uppercase, pos 2=lowercase (no j), pos 3-27=alphanumeric, pos 28=lowercase (no j); `oO01lIQ5S2Z8B` banned everywhere
|
- Password: pos 1=uppercase, pos 2=lowercase (no j), pos 3-27=alphanumeric, pos 28=lowercase (no j); `oO01lIQ5S2Z8B` banned everywhere
|
||||||
- Inline file path: `<PKI_DIR>/inline/private/<CN>.inline`
|
- Inline file path: `<PKI_DIR>/inline/private/<CN>.inline`
|
||||||
|
|||||||
@@ -10,7 +10,7 @@ import email.encoders
|
|||||||
import email.mime.base
|
import email.mime.base
|
||||||
import email.mime.multipart
|
import email.mime.multipart
|
||||||
import email.mime.text
|
import email.mime.text
|
||||||
import hashlib
|
import getpass
|
||||||
import json
|
import json
|
||||||
import os
|
import os
|
||||||
import secrets
|
import secrets
|
||||||
@@ -25,7 +25,7 @@ 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 Optional
|
from typing import Callable, Optional
|
||||||
|
|
||||||
try:
|
try:
|
||||||
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
||||||
@@ -36,9 +36,18 @@ except ImportError: # pragma: no cover
|
|||||||
# SETTINGS — edit these for your environment
|
# SETTINGS — edit these for your environment
|
||||||
# ============================================================
|
# ============================================================
|
||||||
|
|
||||||
|
CONFIG_PATH = "" # explicit path to an external .conf file overriding the settings
|
||||||
|
# below (same syntax as this block). Overridden by --config.
|
||||||
|
# Empty = look for "<this-script-path>.conf" next to the script;
|
||||||
|
# if that default file is absent it is skipped silently. An
|
||||||
|
# explicit CONFIG_PATH or --config that doesn't exist is an error.
|
||||||
|
|
||||||
EASYRSA_DIR = "/etc/easy-rsa"
|
EASYRSA_DIR = "/etc/easy-rsa"
|
||||||
EASYRSA_PKI_DIR = "/etc/easy-rsa/pki"
|
EASYRSA_PKI_DIR = "/etc/easy-rsa/pki"
|
||||||
CA_PASSPHRASE = "" # empty string = no passphrase
|
CA_PASSPHRASE = "" # "" = auto-detect & prompt if the CA key is encrypted
|
||||||
|
# "!empty" = CA key is unencrypted, never check/prompt
|
||||||
|
# "!ask" = always prompt, skip detection
|
||||||
|
# any other value = used as the literal passphrase
|
||||||
|
|
||||||
OVPN_TEMPLATE_PATH = "./template.ovpn"
|
OVPN_TEMPLATE_PATH = "./template.ovpn"
|
||||||
CONFIG_NAME = "client.ovpn"
|
CONFIG_NAME = "client.ovpn"
|
||||||
@@ -62,6 +71,52 @@ SMTP_TLS = "starttls" # "starttls" | "ssl" | "" (plain)
|
|||||||
DAYS_PAST = 30
|
DAYS_PAST = 30
|
||||||
DAYS_AHEAD = 14
|
DAYS_AHEAD = 14
|
||||||
|
|
||||||
|
# Names an external .conf file is allowed to override — keep in sync with the
|
||||||
|
# SETTINGS block above. Deliberately excludes CONFIG_PATH itself.
|
||||||
|
_OVERRIDABLE_SETTINGS = frozenset({
|
||||||
|
"EASYRSA_DIR", "EASYRSA_PKI_DIR", "CA_PASSPHRASE",
|
||||||
|
"OVPN_TEMPLATE_PATH", "CONFIG_NAME", "VPN_CONFIGS_DIR",
|
||||||
|
"CRL_DEST_PATH",
|
||||||
|
"CRYPTGEON_URL",
|
||||||
|
"MAIL_FROM", "MAIL_SUBJECT", "EMAIL_TEMPLATE_PATH", "MAIL_BINARY",
|
||||||
|
"SMTP_HOST", "SMTP_PORT", "SMTP_USER", "SMTP_PASSWORD", "SMTP_TLS",
|
||||||
|
"DAYS_PAST", "DAYS_AHEAD",
|
||||||
|
})
|
||||||
|
|
||||||
|
|
||||||
|
# ============================================================
|
||||||
|
# === SETTINGS OVERRIDE ===
|
||||||
|
# ============================================================
|
||||||
|
|
||||||
|
|
||||||
|
class ConfigError(Exception):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def load_settings_overrides(
|
||||||
|
cli_config: Optional[str], configured_path: str, script_path: str
|
||||||
|
) -> dict:
|
||||||
|
"""Load SETTINGS overrides from an external .conf file (same Python syntax
|
||||||
|
as the SETTINGS block above — it's executed, so only run trusted files).
|
||||||
|
|
||||||
|
Resolution order: --config CLI flag > CONFIG_PATH setting > "<script_path>.conf"
|
||||||
|
next to this script. A path from the CLI flag or CONFIG_PATH is explicit: a
|
||||||
|
missing file raises ConfigError. The default "<script_path>.conf" path is
|
||||||
|
optional: a missing file returns {} silently.
|
||||||
|
"""
|
||||||
|
explicit = bool(cli_config or configured_path)
|
||||||
|
path = cli_config or configured_path or f"{script_path}.conf"
|
||||||
|
|
||||||
|
if not os.path.isfile(path):
|
||||||
|
if explicit:
|
||||||
|
raise ConfigError(f"config file not found: {path}")
|
||||||
|
return {}
|
||||||
|
|
||||||
|
namespace: dict = {}
|
||||||
|
exec(compile(Path(path).read_text(), path, "exec"), namespace)
|
||||||
|
return {name: value for name, value in namespace.items() if name in _OVERRIDABLE_SETTINGS}
|
||||||
|
|
||||||
|
|
||||||
# ============================================================
|
# ============================================================
|
||||||
# === PKI ===
|
# === PKI ===
|
||||||
# ============================================================
|
# ============================================================
|
||||||
@@ -247,6 +302,37 @@ def copy_crl(pki_dir: str, dest_path: str) -> None:
|
|||||||
os.chmod(dest_path, 0o644)
|
os.chmod(dest_path, 0o644)
|
||||||
|
|
||||||
|
|
||||||
|
def is_ca_key_encrypted(pki_dir: str) -> bool:
|
||||||
|
"""Best-effort check of the CA private key's PEM header. Missing/unreadable
|
||||||
|
key => False (let the normal EasyRSA flow surface any real problem)."""
|
||||||
|
try:
|
||||||
|
header = Path(f"{pki_dir}/private/ca.key").read_text()
|
||||||
|
except OSError:
|
||||||
|
return False
|
||||||
|
return "ENCRYPTED" in header
|
||||||
|
|
||||||
|
|
||||||
|
def resolve_ca_passphrase(
|
||||||
|
configured: str, pki_dir: str, prompt: Callable[[str], str] = getpass.getpass
|
||||||
|
) -> str:
|
||||||
|
"""Resolve the SETTINGS CA_PASSPHRASE sentinel into an actual passphrase.
|
||||||
|
|
||||||
|
"" → auto-detect: prompt only if the CA key is actually encrypted
|
||||||
|
"!empty" → explicitly unencrypted; never check, never prompt
|
||||||
|
"!ask" → always prompt, without checking
|
||||||
|
other → used as the literal passphrase
|
||||||
|
"""
|
||||||
|
if configured == "!empty":
|
||||||
|
return ""
|
||||||
|
if configured == "!ask":
|
||||||
|
return prompt("CA passphrase: ")
|
||||||
|
if configured == "":
|
||||||
|
if is_ca_key_encrypted(pki_dir):
|
||||||
|
return prompt("CA key is encrypted; enter CA passphrase: ")
|
||||||
|
return ""
|
||||||
|
return configured
|
||||||
|
|
||||||
|
|
||||||
# ============================================================
|
# ============================================================
|
||||||
# === METADATA ===
|
# === METADATA ===
|
||||||
# ============================================================
|
# ============================================================
|
||||||
@@ -318,25 +404,28 @@ class CryptgeonError(Exception):
|
|||||||
def create_note(content: str, base_url: str) -> str:
|
def create_note(content: str, base_url: str) -> str:
|
||||||
"""AES-256-GCM encrypt content, POST to Cryptgeon, return one-time URL.
|
"""AES-256-GCM encrypt content, POST to Cryptgeon, return one-time URL.
|
||||||
|
|
||||||
Encryption matches the Cryptgeon browser client:
|
Encryption matches the Cryptgeon browser client (occulto's AES.encrypt +
|
||||||
raw_key (32 bytes) → URL fragment (base64url, no padding)
|
API.create):
|
||||||
aes_key = SHA-256(raw_key) → AES-256-GCM
|
key (32 bytes, used directly, no derivation) → URL fragment (hex)
|
||||||
payload = list(nonce + ciphertext)
|
contents = base64(b"AES-GCM") + "--" + base64(nonce) + "--" + base64(ciphertext)
|
||||||
|
meta = JSON string '{"type": "text"}'
|
||||||
"""
|
"""
|
||||||
if AESGCM is None:
|
if AESGCM is None:
|
||||||
raise CryptgeonError("cryptography package not installed (pip install cryptography)")
|
raise CryptgeonError("cryptography package not installed (pip install cryptography)")
|
||||||
|
|
||||||
raw_key = os.urandom(32)
|
key = os.urandom(32)
|
||||||
aes_key = hashlib.sha256(raw_key).digest()
|
nonce = os.urandom(12)
|
||||||
nonce = os.urandom(12)
|
aesgcm = AESGCM(key)
|
||||||
aesgcm = AESGCM(aes_key)
|
|
||||||
ciphertext = aesgcm.encrypt(nonce, content.encode("utf-8"), None)
|
ciphertext = aesgcm.encrypt(nonce, content.encode("utf-8"), None)
|
||||||
|
|
||||||
|
contents = "--".join(
|
||||||
|
base64.b64encode(part).decode("ascii") for part in (b"AES-GCM", nonce, ciphertext)
|
||||||
|
)
|
||||||
|
|
||||||
payload = json.dumps({
|
payload = json.dumps({
|
||||||
"contents": base64.b64encode(nonce + ciphertext).decode("ascii"),
|
"contents": contents,
|
||||||
"meta": "",
|
"meta": json.dumps({"type": "text"}),
|
||||||
"views": 1,
|
"views": 1,
|
||||||
"type": "text",
|
|
||||||
}).encode("utf-8")
|
}).encode("utf-8")
|
||||||
|
|
||||||
api_url = base_url.rstrip("/") + "/api/notes/"
|
api_url = base_url.rstrip("/") + "/api/notes/"
|
||||||
@@ -354,9 +443,8 @@ def create_note(content: str, base_url: str) -> str:
|
|||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
raise CryptgeonError(f"Cryptgeon request failed: {exc}") from exc
|
raise CryptgeonError(f"Cryptgeon request failed: {exc}") from exc
|
||||||
|
|
||||||
note_id = body["id"]
|
note_id = body["id"]
|
||||||
key_fragment = base64.urlsafe_b64encode(raw_key).rstrip(b"=").decode()
|
return f"{base_url.rstrip('/')}/note/{note_id}#{key.hex()}"
|
||||||
return f"{base_url.rstrip('/')}/#/note/{note_id}/{key_fragment}"
|
|
||||||
|
|
||||||
|
|
||||||
# ============================================================
|
# ============================================================
|
||||||
@@ -1296,6 +1384,9 @@ def _build_parser() -> argparse.ArgumentParser:
|
|||||||
parser.add_argument("--show-eml", action="store_true",
|
parser.add_argument("--show-eml", action="store_true",
|
||||||
help="Print the generated email as base64-encoded .eml to stdout "
|
help="Print the generated email as base64-encoded .eml to stdout "
|
||||||
"(implies --no-send-email unless --send-email is also given)")
|
"(implies --no-send-email unless --send-email is also given)")
|
||||||
|
parser.add_argument("--config", metavar="PATH",
|
||||||
|
help="External .conf file overriding SETTINGS (overrides CONFIG_PATH; "
|
||||||
|
"a missing file here is an error)")
|
||||||
return parser
|
return parser
|
||||||
|
|
||||||
|
|
||||||
@@ -1308,6 +1399,16 @@ def main() -> None:
|
|||||||
parser = _build_parser()
|
parser = _build_parser()
|
||||||
args = parser.parse_args()
|
args = parser.parse_args()
|
||||||
|
|
||||||
|
try:
|
||||||
|
overrides = load_settings_overrides(args.config, CONFIG_PATH, __file__)
|
||||||
|
except ConfigError as exc:
|
||||||
|
print(f"error: {exc}", file=sys.stderr)
|
||||||
|
sys.exit(1)
|
||||||
|
globals().update(overrides)
|
||||||
|
|
||||||
|
global CA_PASSPHRASE
|
||||||
|
CA_PASSPHRASE = resolve_ca_passphrase(CA_PASSPHRASE, EASYRSA_PKI_DIR)
|
||||||
|
|
||||||
# --show-eml switches the default to --no-send-email; explicit --send-email overrides.
|
# --show-eml switches the default to --no-send-email; explicit --send-email overrides.
|
||||||
send_email_flag = args.send_email if args.send_email is not None else (not args.show_eml)
|
send_email_flag = args.send_email if args.send_email is not None else (not args.show_eml)
|
||||||
|
|
||||||
|
|||||||
@@ -200,6 +200,65 @@ def test_main_no_args_launches_tui(monkeypatch):
|
|||||||
app.run.assert_called_once_with()
|
app.run.assert_called_once_with()
|
||||||
|
|
||||||
|
|
||||||
|
def test_main_resolves_ca_passphrase_before_dispatch(monkeypatch, tmp_path):
|
||||||
|
"""The configured CA_PASSPHRASE sentinel must be resolved to an actual
|
||||||
|
passphrase before any EasyRSA call, and that resolved value (not the
|
||||||
|
sentinel) must be what reaches them."""
|
||||||
|
monkeypatch.setattr(sys, "argv", ["prog", "--gen-crl"])
|
||||||
|
monkeypatch.setattr("openvpncertupdate.CA_PASSPHRASE", "!ask")
|
||||||
|
monkeypatch.setattr("openvpncertupdate.EASYRSA_DIR", "/er")
|
||||||
|
monkeypatch.setattr("openvpncertupdate.EASYRSA_PKI_DIR", str(tmp_path))
|
||||||
|
mock_gen_crl = MagicMock()
|
||||||
|
monkeypatch.setattr("openvpncertupdate.gen_crl", mock_gen_crl)
|
||||||
|
monkeypatch.setattr("openvpncertupdate.copy_crl", MagicMock())
|
||||||
|
mock_resolve = MagicMock(return_value="typed-pass")
|
||||||
|
monkeypatch.setattr("openvpncertupdate.resolve_ca_passphrase", mock_resolve)
|
||||||
|
|
||||||
|
main()
|
||||||
|
|
||||||
|
mock_resolve.assert_called_once_with("!ask", str(tmp_path))
|
||||||
|
|
||||||
|
|
||||||
|
def test_main_applies_config_overrides_before_dispatch(monkeypatch, tmp_path):
|
||||||
|
"""Overrides returned by load_settings_overrides() must land in the module
|
||||||
|
globals before EasyRSA calls, so a config file can redirect EASYRSA_PKI_DIR etc."""
|
||||||
|
monkeypatch.setattr(sys, "argv", ["prog", "--gen-crl"])
|
||||||
|
monkeypatch.setattr("openvpncertupdate.EASYRSA_DIR", "/er")
|
||||||
|
monkeypatch.setattr("openvpncertupdate.EASYRSA_PKI_DIR", "/original/pki")
|
||||||
|
monkeypatch.setattr("openvpncertupdate.CA_PASSPHRASE", "!empty")
|
||||||
|
mock_gen_crl = MagicMock()
|
||||||
|
monkeypatch.setattr("openvpncertupdate.gen_crl", mock_gen_crl)
|
||||||
|
monkeypatch.setattr("openvpncertupdate.copy_crl", MagicMock())
|
||||||
|
mock_load = MagicMock(return_value={"EASYRSA_PKI_DIR": "/overridden/pki"})
|
||||||
|
monkeypatch.setattr("openvpncertupdate.load_settings_overrides", mock_load)
|
||||||
|
|
||||||
|
main()
|
||||||
|
|
||||||
|
mock_gen_crl.assert_called_once_with("/er", "/overridden/pki", "")
|
||||||
|
|
||||||
|
|
||||||
|
def test_main_passes_config_flag_to_load_overrides(monkeypatch):
|
||||||
|
monkeypatch.setattr(sys, "argv", ["prog", "--gen-crl", "--config", "/explicit/path.conf"])
|
||||||
|
monkeypatch.setattr("openvpncertupdate.CA_PASSPHRASE", "!empty")
|
||||||
|
monkeypatch.setattr("openvpncertupdate.CliRunner", lambda: MagicMock())
|
||||||
|
mock_load = MagicMock(return_value={})
|
||||||
|
monkeypatch.setattr("openvpncertupdate.load_settings_overrides", mock_load)
|
||||||
|
|
||||||
|
main()
|
||||||
|
|
||||||
|
assert mock_load.call_args[0][0] == "/explicit/path.conf"
|
||||||
|
|
||||||
|
|
||||||
|
def test_main_exits_with_error_when_explicit_config_missing(monkeypatch, capsys):
|
||||||
|
monkeypatch.setattr(sys, "argv", ["prog", "--gen-crl", "--config", "/does/not/exist.conf"])
|
||||||
|
|
||||||
|
with pytest.raises(SystemExit) as exc_info:
|
||||||
|
main()
|
||||||
|
|
||||||
|
assert exc_info.value.code != 0
|
||||||
|
assert "/does/not/exist.conf" in capsys.readouterr().err
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
# --send-email / --no-send-email
|
# --send-email / --no-send-email
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|||||||
62
tests/test_config_file.py
Normal file
62
tests/test_config_file.py
Normal file
@@ -0,0 +1,62 @@
|
|||||||
|
import pytest
|
||||||
|
from openvpncertupdate import load_settings_overrides, ConfigError
|
||||||
|
|
||||||
|
|
||||||
|
def _script_path(tmp_path):
|
||||||
|
return str(tmp_path / "openvpncertupdate.py")
|
||||||
|
|
||||||
|
|
||||||
|
def test_default_path_missing_returns_empty_silently(tmp_path):
|
||||||
|
overrides = load_settings_overrides(None, "", _script_path(tmp_path))
|
||||||
|
assert overrides == {}
|
||||||
|
|
||||||
|
|
||||||
|
def test_default_path_found_loads_overrides(tmp_path):
|
||||||
|
conf = tmp_path / "openvpncertupdate.py.conf"
|
||||||
|
conf.write_text('EASYRSA_DIR = "/custom/easyrsa"\n')
|
||||||
|
overrides = load_settings_overrides(None, "", _script_path(tmp_path))
|
||||||
|
assert overrides == {"EASYRSA_DIR": "/custom/easyrsa"}
|
||||||
|
|
||||||
|
|
||||||
|
def test_explicit_cli_path_missing_raises(tmp_path):
|
||||||
|
missing = str(tmp_path / "missing.conf")
|
||||||
|
with pytest.raises(ConfigError, match=missing):
|
||||||
|
load_settings_overrides(missing, "", _script_path(tmp_path))
|
||||||
|
|
||||||
|
|
||||||
|
def test_explicit_configured_path_missing_raises(tmp_path):
|
||||||
|
missing = str(tmp_path / "missing2.conf")
|
||||||
|
with pytest.raises(ConfigError, match=missing):
|
||||||
|
load_settings_overrides(None, missing, _script_path(tmp_path))
|
||||||
|
|
||||||
|
|
||||||
|
def test_cli_path_takes_priority_over_configured_path(tmp_path):
|
||||||
|
cli_conf = tmp_path / "from_cli.conf"
|
||||||
|
cli_conf.write_text('EASYRSA_DIR = "/from/cli"\n')
|
||||||
|
configured_conf = tmp_path / "from_configured.conf"
|
||||||
|
configured_conf.write_text('EASYRSA_DIR = "/from/configured"\n')
|
||||||
|
|
||||||
|
overrides = load_settings_overrides(str(cli_conf), str(configured_conf), _script_path(tmp_path))
|
||||||
|
assert overrides == {"EASYRSA_DIR": "/from/cli"}
|
||||||
|
|
||||||
|
|
||||||
|
def test_unrecognized_and_config_path_names_are_filtered_out(tmp_path):
|
||||||
|
conf = tmp_path / "openvpncertupdate.py.conf"
|
||||||
|
conf.write_text(
|
||||||
|
'EASYRSA_DIR = "/custom/easyrsa"\n'
|
||||||
|
'SOME_RANDOM_NAME = "y"\n'
|
||||||
|
'CONFIG_PATH = "/should/not/apply"\n'
|
||||||
|
)
|
||||||
|
overrides = load_settings_overrides(None, "", _script_path(tmp_path))
|
||||||
|
assert overrides == {"EASYRSA_DIR": "/custom/easyrsa"}
|
||||||
|
|
||||||
|
|
||||||
|
def test_comments_and_non_string_types_supported(tmp_path):
|
||||||
|
conf = tmp_path / "openvpncertupdate.py.conf"
|
||||||
|
conf.write_text(
|
||||||
|
"# a comment\n"
|
||||||
|
"SMTP_PORT = 2525\n"
|
||||||
|
'CA_PASSPHRASE = "" # inline comment\n'
|
||||||
|
)
|
||||||
|
overrides = load_settings_overrides(None, "", _script_path(tmp_path))
|
||||||
|
assert overrides == {"SMTP_PORT": 2525, "CA_PASSPHRASE": ""}
|
||||||
@@ -1,6 +1,8 @@
|
|||||||
|
import base64
|
||||||
import json
|
import json
|
||||||
import pytest
|
import pytest
|
||||||
from unittest.mock import patch, MagicMock
|
from unittest.mock import patch, MagicMock
|
||||||
|
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
||||||
from openvpncertupdate import create_note, CryptgeonError
|
from openvpncertupdate import create_note, CryptgeonError
|
||||||
|
|
||||||
|
|
||||||
@@ -13,11 +15,13 @@ def mock_response(body: bytes):
|
|||||||
|
|
||||||
|
|
||||||
@patch("openvpncertupdate.urllib.request.urlopen")
|
@patch("openvpncertupdate.urllib.request.urlopen")
|
||||||
def test_returns_url_with_id(mock_open):
|
def test_url_uses_path_based_note_route_with_hex_key_fragment(mock_open):
|
||||||
mock_open.return_value = mock_response(b'{"id":"abc-123"}')
|
mock_open.return_value = mock_response(b'{"id":"abc-123"}')
|
||||||
url = create_note("secret", "https://cg.example.com")
|
url = create_note("secret", "https://cg.example.com")
|
||||||
assert "abc-123" in url
|
assert url.startswith("https://cg.example.com/note/abc-123#")
|
||||||
assert url.startswith("https://cg.example.com/#/note/abc-123/")
|
fragment = url.split("#", 1)[1]
|
||||||
|
assert len(fragment) == 64
|
||||||
|
bytes.fromhex(fragment) # must be valid hex, not base64url
|
||||||
|
|
||||||
|
|
||||||
@patch("openvpncertupdate.urllib.request.urlopen")
|
@patch("openvpncertupdate.urllib.request.urlopen")
|
||||||
@@ -26,11 +30,29 @@ def test_posts_to_correct_endpoint(mock_open):
|
|||||||
create_note("hello", "https://cg.example.com")
|
create_note("hello", "https://cg.example.com")
|
||||||
req = mock_open.call_args[0][0]
|
req = mock_open.call_args[0][0]
|
||||||
assert req.full_url == "https://cg.example.com/api/notes/"
|
assert req.full_url == "https://cg.example.com/api/notes/"
|
||||||
|
|
||||||
|
|
||||||
|
@patch("openvpncertupdate.urllib.request.urlopen")
|
||||||
|
def test_meta_is_json_string_describing_text_note(mock_open):
|
||||||
|
mock_open.return_value = mock_response(b'{"id":"x"}')
|
||||||
|
create_note("hello", "https://cg.example.com")
|
||||||
|
req = mock_open.call_args[0][0]
|
||||||
body = json.loads(req.data)
|
body = json.loads(req.data)
|
||||||
assert isinstance(body["contents"], str)
|
assert json.loads(body["meta"]) == {"type": "text"}
|
||||||
assert body["meta"] == ""
|
|
||||||
assert body["views"] == 1
|
assert body["views"] == 1
|
||||||
assert body["type"] == "text"
|
|
||||||
|
|
||||||
|
@patch("openvpncertupdate.urllib.request.urlopen")
|
||||||
|
def test_contents_is_three_part_dash_delimited_base64(mock_open):
|
||||||
|
mock_open.return_value = mock_response(b'{"id":"x"}')
|
||||||
|
create_note("hello", "https://cg.example.com")
|
||||||
|
req = mock_open.call_args[0][0]
|
||||||
|
body = json.loads(req.data)
|
||||||
|
parts = body["contents"].split("--")
|
||||||
|
assert len(parts) == 3
|
||||||
|
alg, nonce, ciphertext = (base64.b64decode(p) for p in parts)
|
||||||
|
assert alg == b"AES-GCM"
|
||||||
|
assert len(nonce) == 12
|
||||||
|
|
||||||
|
|
||||||
@patch("openvpncertupdate.urllib.request.urlopen")
|
@patch("openvpncertupdate.urllib.request.urlopen")
|
||||||
@@ -42,8 +64,28 @@ def test_raises_on_http_error(mock_open):
|
|||||||
|
|
||||||
|
|
||||||
@patch("openvpncertupdate.urllib.request.urlopen")
|
@patch("openvpncertupdate.urllib.request.urlopen")
|
||||||
def test_different_passwords_give_different_urls(mock_open):
|
def test_different_contents_give_different_urls(mock_open):
|
||||||
mock_open.return_value = mock_response(b'{"id":"same"}')
|
mock_open.return_value = mock_response(b'{"id":"same"}')
|
||||||
url1 = create_note("pass1", "https://cg.example.com")
|
url1 = create_note("pass1", "https://cg.example.com")
|
||||||
url2 = create_note("pass2", "https://cg.example.com")
|
url2 = create_note("pass2", "https://cg.example.com")
|
||||||
assert url1 != url2
|
assert url1 != url2
|
||||||
|
|
||||||
|
|
||||||
|
@patch("openvpncertupdate.urllib.request.urlopen")
|
||||||
|
def test_decrypts_like_the_browser_client(mock_open):
|
||||||
|
"""Reproduces the note/[id] view: key from the URL fragment must be the
|
||||||
|
exact key used to encrypt (no extra hashing step), or GCM auth fails."""
|
||||||
|
mock_open.return_value = mock_response(b'{"id":"abc"}')
|
||||||
|
url = create_note("my secret password", "https://cg.example.com")
|
||||||
|
req = mock_open.call_args[0][0]
|
||||||
|
body = json.loads(req.data)
|
||||||
|
|
||||||
|
key_hex = url.split("#", 1)[1]
|
||||||
|
key = bytes.fromhex(key_hex)
|
||||||
|
|
||||||
|
alg_b64, nonce_b64, ciphertext_b64 = body["contents"].split("--")
|
||||||
|
nonce = base64.b64decode(nonce_b64)
|
||||||
|
ciphertext = base64.b64decode(ciphertext_b64)
|
||||||
|
|
||||||
|
plaintext = AESGCM(key).decrypt(nonce, ciphertext, None)
|
||||||
|
assert plaintext == b"my secret password"
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ import pytest
|
|||||||
from unittest.mock import patch, MagicMock
|
from unittest.mock import patch, MagicMock
|
||||||
from openvpncertupdate import (
|
from openvpncertupdate import (
|
||||||
revoke_issued, build_client_full, gen_crl, copy_crl, EasyRSAError,
|
revoke_issued, build_client_full, gen_crl, copy_crl, EasyRSAError,
|
||||||
|
is_ca_key_encrypted, resolve_ca_passphrase,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@@ -65,3 +66,75 @@ def test_copy_crl_copies_and_chmods(mock_chmod, mock_copy):
|
|||||||
copy_crl("/pki", "/etc/openvpn/crl.pem")
|
copy_crl("/pki", "/etc/openvpn/crl.pem")
|
||||||
mock_copy.assert_called_once_with("/pki/crl.pem", "/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)
|
mock_chmod.assert_called_once_with("/etc/openvpn/crl.pem", 0o644)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 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()
|
||||||
|
|||||||
Reference in New Issue
Block a user