mineracks-ckbunker-hsm-sign/ckbunker_hsm_sign/config.py

"""Configuration loading.

Three sources, in precedence order (highest wins):
  1. CLI flags
  2. YAML file (if --config path is provided)
  3. Environment / .env

Each source is optional. The harness fails with a clear error if something
it actually needs is missing at test-run time, not up-front — so running
`hsm_validate.py --tests connectivity` works with almost no config.
"""

from __future__ import annotations

import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

try:
    import yaml
except ImportError:
    yaml = None


@dataclass
class PolicyExpectations:
    auto_approve_per_txn_sats: int = 10_000
    auto_approve_per_period_sats: int = 50_000
    user_authorised_per_txn_sats: int = 100_000
    user_authorised_per_period_sats: int = 500_000
    velocity_minutes: int = 1440
    message_signing: bool = True


@dataclass
class Config:
    url: str = "http://127.0.0.1:9823"
    cf_client_id: str | None = None
    cf_client_secret: str | None = None

    totp_secret: str | None = None
    user: str = "mineracks"
    message_sign_path: str = "m/84'/0'/0'/1"
    message_sign_address: str | None = None

    small_psbt_path: str = "fixtures/small.psbt"
    large_psbt_path: str = "fixtures/large.psbt"

    policy: PolicyExpectations = field(default_factory=PolicyExpectations)

    tests: dict[str, bool] = field(
        default_factory=lambda: {
            "connectivity": True,
            "message_signing": True,
            "rule2_auto_approve": True,
            "rule1_without_totp_rejects": True,
            "rule1_with_totp_signs": True,
            "counters_tracked": True,
        }
    )

    verbose: bool = False
    save_signed_dir: str | None = None


def _load_dotenv(path: Path) -> dict[str, str]:
    """Tiny .env parser — we don't want python-dotenv as a dependency."""
    out: dict[str, str] = {}
    if not path.exists():
        return out
    for line in path.read_text().splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        k, v = line.split("=", 1)
        v = v.strip().strip('"').strip("'")
        out[k.strip()] = v
    return out


def _apply_env(cfg: Config, env: dict[str, str]) -> None:
    def get(k: str, default: Any = None) -> Any:
        return env.get(k, os.environ.get(k, default))

    cfg.url = get("CKBUNKER_URL", cfg.url)
    cfg.cf_client_id = get("CF_ACCESS_CLIENT_ID", cfg.cf_client_id) or None
    cfg.cf_client_secret = get("CF_ACCESS_CLIENT_SECRET", cfg.cf_client_secret) or None
    cfg.totp_secret = get("TOTP_SECRET", cfg.totp_secret) or None
    cfg.user = get("HSM_USER", cfg.user)
    cfg.message_sign_path = get("MESSAGE_SIGN_PATH", cfg.message_sign_path)
    cfg.message_sign_address = get("MESSAGE_SIGN_ADDRESS", cfg.message_sign_address) or None
    cfg.small_psbt_path = get("SMALL_PSBT_PATH", cfg.small_psbt_path)
    cfg.large_psbt_path = get("LARGE_PSBT_PATH", cfg.large_psbt_path)


def _apply_yaml(cfg: Config, data: dict) -> None:
    if not data:
        return
    bunker = data.get("ckbunker", {})
    cfg.url = bunker.get("url", cfg.url)
    cfg.cf_client_id = bunker.get("cf_access_client_id", cfg.cf_client_id)
    cfg.cf_client_secret = bunker.get("cf_access_client_secret", cfg.cf_client_secret)

    hsm = data.get("hsm", {})
    cfg.user = hsm.get("user", cfg.user)
    cfg.message_sign_path = hsm.get("message_sign_path", cfg.message_sign_path)
    cfg.message_sign_address = hsm.get("message_sign_address", cfg.message_sign_address)

    pol = data.get("policy", {}) or {}
    aa = pol.get("auto_approve", {}) or {}
    ua = pol.get("user_authorised", {}) or {}
    cfg.policy.auto_approve_per_txn_sats = aa.get(
        "per_txn_sats", cfg.policy.auto_approve_per_txn_sats
    )
    cfg.policy.auto_approve_per_period_sats = aa.get(
        "per_period_sats", cfg.policy.auto_approve_per_period_sats
    )
    cfg.policy.user_authorised_per_txn_sats = ua.get(
        "per_txn_sats", cfg.policy.user_authorised_per_txn_sats
    )
    cfg.policy.user_authorised_per_period_sats = ua.get(
        "per_period_sats", cfg.policy.user_authorised_per_period_sats
    )
    cfg.policy.velocity_minutes = pol.get("velocity_minutes", cfg.policy.velocity_minutes)
    cfg.policy.message_signing = pol.get("message_signing", cfg.policy.message_signing)

    fx = data.get("fixtures", {}) or {}
    cfg.small_psbt_path = fx.get("small_psbt", cfg.small_psbt_path)
    cfg.large_psbt_path = fx.get("large_psbt", cfg.large_psbt_path)

    tests = data.get("tests", {}) or {}
    for k, v in tests.items():
        if k in cfg.tests:
            cfg.tests[k] = bool(v)

    out = data.get("output", {}) or {}
    cfg.verbose = bool(out.get("verbose", cfg.verbose))
    cfg.save_signed_dir = out.get("save_signed_dir", cfg.save_signed_dir)


def load_config(
    *,
    yaml_path: Path | None = None,
    dotenv_path: Path | None = Path(".env"),
    overrides: dict[str, Any] | None = None,
) -> Config:
    cfg = Config()

    env = _load_dotenv(dotenv_path) if dotenv_path else {}
    _apply_env(cfg, env)

    if yaml_path:
        if yaml is None:
            raise SystemExit("PyYAML required to read --config. pip install PyYAML")
        with open(yaml_path) as f:
            data = yaml.safe_load(f) or {}
        _apply_yaml(cfg, data)

    if overrides:
        for k, v in overrides.items():
            if v is None:
                continue
            if hasattr(cfg, k):
                setattr(cfg, k, v)

    return cfg