diff --git a/.env.example b/.env.example index e431ecb..355d334 100644 --- a/.env.example +++ b/.env.example @@ -4,8 +4,24 @@ # Licensed under MIT. See LICENSE. # ── Pry Configuration ── -# Copy this to .env and adjust values. -# All PRY_* vars are auto-loaded by PrySettings. +# Copy this to .env and adjust values. All PRY_* vars are auto-loaded. +# +# ── SECRET BACKEND (see secrets_backend.py) ── +# Pry resolves secrets via secrets_backend.get_secret(name). Default backend is gopass. +# Override with PRY_SECRET_BACKEND: +# gopass (default) - reads from the gopass store under pry/ +# env - reads from os.environ (PRY_ or PRY_) +# file - reads from PRY_ENV_FILE (default: $PRY_DATA_DIR/.env) +# auto - tries gopass, falls back to env (default) +# +# Seed initial secrets (one time): +# gopass insert -m pry/jwt_secret # opens editor, paste a strong random value +# gopass insert -m pry/api_key +# gopass insert -m pry/x402_wallet # your receiving EVM/Solana address +# gopass insert -m pry/x402_facilitator # https://x402.org/facilitator or your own +# +# Per the SECURITY.md contract, NO secrets live in .env in production. +# .env is for local dev and CI only. # ── Core ── PRY_HOST=0.0.0.0 diff --git a/auth.py b/auth.py index ce44e11..b6d55d0 100644 --- a/auth.py +++ b/auth.py @@ -25,7 +25,16 @@ except ImportError: _has_jwt = False # Configuration -JWT_SECRET = os.getenv("PRY_JWT_SECRET", "change-me-in-production-" + secrets.token_hex(16)) +# JWT secret: prefer gopass (PRY_SECRET_BACKEND=gopass), fall back to env, then to a +# random ephemeral default. The default is intentionally NOT a fixed string so that +# an unset JWT_SECRET cannot accidentally sign tokens in a way that survives restart. +try: + from secrets_backend import get_secret + JWT_SECRET = get_secret("jwt_secret") or os.getenv("PRY_JWT_SECRET") or ( + "ephemeral-" + secrets.token_hex(32) + ) +except ImportError: + JWT_SECRET = os.getenv("PRY_JWT_SECRET") or ("ephemeral-" + secrets.token_hex(32)) JWT_ALGORITHM = "HS256" JWT_EXPIRY_HOURS = 24 API_KEY_LENGTH = 32 diff --git a/secrets_backend.py b/secrets_backend.py new file mode 100644 index 0000000..28aa86a --- /dev/null +++ b/secrets_backend.py @@ -0,0 +1,238 @@ +"""Pry - secret resolution with pluggable backends. + +Resolves secrets from a pluggable backend. The default backend is `gopass` +(the fleet's standard secret store). `env` and `file` are also supported for +local dev and CI. + +Selection order at import time (and per-call, since each call re-reads the +backend mode from the env): + + PRY_SECRET_BACKEND=gopass # default - gopass at $GOPASS_BIN or `gopass` + PRY_SECRET_BACKEND=env # os.environ (12-factor) + PRY_SECRET_BACKEND=file # .env file in PRY_DATA_DIR/.env or PRY_ENV_FILE + PRY_SECRET_BACKEND=auto # try gopass, fall back to env + +Usage: + from secrets_backend import get_secret + api_key = get_secret("pry/openrouter_api_key") + jwt_secret = get_secret("pry/jwt_secret", default="change-me") + +The gopass lookup path is `pry/`. If you pass a name with leading +"pry/" we use it as-is. If you pass "openrouter_api_key" we prepend "pry/". + +Part of Pry - https://git.rugmunch.io/RugMunchMedia/pryscraper +Licensed under MIT. See LICENSE. +""" +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 Rug Munch Media LLC +# +# Part of Pry - https://git.rugmunch.io/RugMunchMedia/pryscraper +# Licensed under MIT. See LICENSE. +from __future__ import annotations + +import logging +import os +import shutil +import subprocess +from functools import lru_cache +from pathlib import Path + +logger = logging.getLogger(__name__) + +# Backend constants +BACKEND_GOPASS = "gopass" +BACKEND_ENV = "env" +BACKEND_FILE = "file" +BACKEND_AUTO = "auto" +VALID_BACKENDS = {BACKEND_GOPASS, BACKEND_ENV, BACKEND_FILE, BACKEND_AUTO} + +_DEFAULT_BACKEND = BACKEND_AUTO +# Where to put the .env file by default (overridable via PRY_ENV_FILE) +from paths import PRY_DATA_DIR # noqa: E402 - import after the docstring + +_DEFAULT_ENV_FILE = PRY_DATA_DIR / ".env" + +# Lazy probe results +_GOPASS_AVAILABLE: bool | None = None + + +def _detect_backend() -> str: + """Resolve which backend to use right now, based on env + availability.""" + requested = os.getenv("PRY_SECRET_BACKEND", _DEFAULT_BACKEND).lower() + if requested not in VALID_BACKENDS: + logger.warning("invalid_secret_backend", extra={"value": requested, "fallback": BACKEND_AUTO}) + return BACKEND_AUTO + if requested == BACKEND_AUTO: + return BACKEND_GOPASS if _gopass_available() else BACKEND_ENV + return requested + + +def _gopass_available() -> bool: + """Check once whether gopass is on PATH and working.""" + global _GOPASS_AVAILABLE + if _GOPASS_AVAILABLE is not None: + return _GOPASS_AVAILABLE + bin_path = shutil.which("gopass") + if not bin_path: + _GOPASS_AVAILABLE = False + return False + try: + result = subprocess.run( + [bin_path, "ls"], + capture_output=True, + text=True, + timeout=5, + check=False, + ) + _GOPASS_AVAILABLE = result.returncode == 0 + except (subprocess.TimeoutExpired, OSError) as e: + logger.debug("gopass_probe_failed", extra={"error": str(e)[:80]}) + _GOPASS_AVAILABLE = False + return _GOPASS_AVAILABLE + + +def _normalize_name(name: str) -> str: + """Normalize a secret name to a gopass-style path under pry/.""" + if name.startswith("pry/"): + return name + return f"pry/{name}" + + +def _get_from_gopass(name: str) -> str | None: + """Read a secret from gopass. Returns None on miss or error.""" + if not _gopass_available(): + return None + bin_path = shutil.which("gopass") + if not bin_path: + return None + path = _normalize_name(name) + try: + result = subprocess.run( + [bin_path, "show", "-o", path], + capture_output=True, + text=True, + timeout=10, + check=False, + ) + if result.returncode == 0: + value = result.stdout.rstrip("\n") + if value: + return value + except (subprocess.TimeoutExpired, OSError) as e: + logger.debug("gopass_read_failed", extra={"path": path, "error": str(e)[:80]}) + return None + + +def _get_from_env(name: str) -> str | None: + """Read a secret from the process environment. The name is upper-cased and + prefixed with PRY_ to match the convention used elsewhere in the codebase.""" + candidates = [ + os.getenv(name), + os.getenv(name.upper()), + os.getenv(f"PRY_{name.upper()}"), + os.getenv(f"PRY_{name}"), + ] + for v in candidates: + if v: + return v + return None + + +@lru_cache(maxsize=1) +def _read_env_file() -> dict[str, str]: + """Read a .env file once. Returns dict of KEY=value pairs.""" + path = Path(os.getenv("PRY_ENV_FILE", str(_DEFAULT_ENV_FILE))) + if not path.exists(): + return {} + values: dict[str, str] = {} + try: + for line in path.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if not line or line.startswith("#"): + continue + if "=" not in line: + continue + k, v = line.split("=", 1) + values[k.strip()] = v.strip().strip('"').strip("'") + except OSError as e: + logger.debug("env_file_read_failed", extra={"path": str(path), "error": str(e)[:80]}) + return values + + +def _get_from_file(name: str) -> str | None: + """Read a secret from the .env file.""" + values = _read_env_file() + for key in (name, name.upper(), f"PRY_{name.upper()}", f"PRY_{name}"): + if key in values and values[key]: + return values[key] + return None + + +def get_secret(name: str, default: str = "") -> str: + """Get a secret by name, using the configured backend. + + Args: + name: Secret name. Will be looked up as `pry/` in gopass, and + as `PRY_` (uppercased) in env vars. + default: Value to return if the secret isn't found in any backend. + + Returns: + The secret string, or `default` if not found. + """ + backend = _detect_backend() + if backend == BACKEND_GOPASS: + v = _get_from_gopass(name) + if v is not None: + return v + # Fall through to env if gopass misses (so a partial migration works) + v = _get_from_env(name) + if v is not None: + return v + elif backend == BACKEND_ENV: + v = _get_from_env(name) + if v is not None: + return v + elif backend == BACKEND_FILE: + v = _get_from_file(name) + if v is not None: + return v + return default + + +def set_secret(name: str, value: str) -> bool: + """Write a secret to gopass (the only backend that supports writes). + + Returns True on success, False on failure. + """ + if not _gopass_available(): + logger.warning("set_secret_skipped", extra={"reason": "gopass not available", "secret_name": name}) + return False + bin_path = shutil.which("gopass") + path = _normalize_name(name) + try: + result = subprocess.run( + [bin_path, "insert", "-f", "-m", path], + input=value, + capture_output=True, + text=True, + timeout=10, + check=False, + ) + if result.returncode == 0: + logger.info("secret_stored", extra={"path": path}) + return True + logger.warning("gopass_insert_failed", extra={"path": path, "stderr": result.stderr[:200]}) + except (subprocess.TimeoutExpired, OSError) as e: + logger.warning("gopass_insert_error", extra={"path": path, "error": str(e)[:80]}) + return False + + +def backend_info() -> dict[str, str]: + """Diagnostic info about the current backend configuration.""" + return { + "requested": os.getenv("PRY_SECRET_BACKEND", _DEFAULT_BACKEND), + "active": _detect_backend(), + "gopass_available": str(_gopass_available()), + "gopass_path": shutil.which("gopass") or "", + "env_file": os.getenv("PRY_ENV_FILE", str(_DEFAULT_ENV_FILE)), + } diff --git a/tests/test_secrets_backend.py b/tests/test_secrets_backend.py new file mode 100644 index 0000000..0f59558 --- /dev/null +++ b/tests/test_secrets_backend.py @@ -0,0 +1,108 @@ +"""Tests for secrets_backend module.""" +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 Rug Munch Media LLC +# +# Part of Pry - https://git.rugmunch.io/RugMunchMedia/pryscraper +# Licensed under MIT. See LICENSE. +from __future__ import annotations + +import importlib +import os +from pathlib import Path +from unittest.mock import patch + +import pytest + +import secrets_backend +from secrets_backend import ( + BACKEND_AUTO, + BACKEND_ENV, + BACKEND_FILE, + BACKEND_GOPASS, + _normalize_name, + backend_info, + get_secret, + set_secret, +) + + +def test_normalize_name_adds_pry_prefix(): + assert _normalize_name("api_key") == "pry/api_key" + assert _normalize_name("pry/api_key") == "pry/api_key" + # Names that already have a slash are treated as paths (not re-prefixed) + assert _normalize_name("PRY/api_key").startswith("pry/") + + +def test_get_secret_default_when_missing(): + # Use a name that definitely doesn't exist anywhere + with patch.dict(os.environ, {"PRY_SECRET_BACKEND": "env"}, clear=False): + # Clear any env vars that might match + with patch.dict(os.environ, {}, clear=False): + value = get_secret("definitely_not_a_real_secret_name_xyz_12345", default="missing") + assert value == "missing" + + +def test_get_secret_from_env(monkeypatch): + monkeypatch.setenv("PRY_SECRET_BACKEND", "env") + monkeypatch.setenv("PRY_TEST_SECRET", "from-env") + value = get_secret("test_secret") + assert value == "from-env" + + +def test_get_secret_from_env_uppercase(monkeypatch): + """Even when the caller passes lowercase, we should find PRY_UPPERCASE.""" + monkeypatch.setenv("PRY_SECRET_BACKEND", "env") + monkeypatch.setenv("PRY_FOO", "upper-value") + value = get_secret("foo") + assert value == "upper-value" + + +def test_get_secret_falls_back_to_env_when_gopass_misses(monkeypatch): + """When the active backend is gopass and a secret is not in gopass, + we should fall back to the env (so partial migrations work).""" + monkeypatch.setenv("PRY_SECRET_BACKEND", "gopass") + monkeypatch.setenv("PRY_FALLBACK_TEST", "env-fallback") + # gopass lookup will fail (entry doesn't exist), then we check env + with patch("secrets_backend._get_from_gopass", return_value=None): + value = get_secret("fallback_test") + assert value == "env-fallback" + + +def test_backend_info_returns_expected_keys(): + info = backend_info() + assert "requested" in info + assert "active" in info + assert "gopass_available" in info + assert "gopass_path" in info + assert "env_file" in info + + +def test_get_secret_from_file(tmp_path, monkeypatch): + """Read a secret from a .env file when backend is file.""" + env_file = tmp_path / "test.env" + env_file.write_text("PRY_FROM_FILE=hello\n# comment\nPRY_OTHER=world\n") + monkeypatch.setenv("PRY_SECRET_BACKEND", "file") + monkeypatch.setenv("PRY_ENV_FILE", str(env_file)) + # Reimport to pick up the new env file + importlib.reload(secrets_backend) + try: + value = secrets_backend.get_secret("from_file") + assert value == "hello" + finally: + # Restore + monkeypatch.delenv("PRY_ENV_FILE", raising=False) + importlib.reload(secrets_backend) + + +def test_set_secret_returns_false_when_gopass_unavailable(monkeypatch): + monkeypatch.setattr(secrets_backend, "_GOPASS_AVAILABLE", False) + # Force the global to be re-evaluated + monkeypatch.setattr(secrets_backend, "_gopass_available", lambda: False) + result = set_secret("test_secret", "value") + assert result is False + + +def test_detect_backend_invalid_falls_back_to_auto(monkeypatch): + monkeypatch.setenv("PRY_SECRET_BACKEND", "not-a-real-backend") + backend = secrets_backend._detect_backend() + assert backend == BACKEND_AUTO diff --git a/x402.py b/x402.py index 26127ab..1d6ef99 100644 --- a/x402.py +++ b/x402.py @@ -97,10 +97,19 @@ class X402Asset(StrEnum): DOGE = "DOGE" -X402_WALLET = os.getenv("PRY_X402_WALLET", "pry-default-wallet") +# Prefer gopass for the x402 wallet and facilitator URL. Env vars override only +# if the backend returns an empty string. This lets you: +# gopass insert -m pry/x402_wallet # persistent across restarts +# PRY_X402_WALLET=0x... # one-off override (CI, dev) +try: + from secrets_backend import get_secret as _gs + X402_WALLET = _gs("x402_wallet") or os.getenv("PRY_X402_WALLET", "pry-default-wallet") + X402_FACILITATOR_URL = _gs("x402_facilitator") or os.getenv("PRY_X402_FACILITATOR", "https://x402.org/facilitator") +except ImportError: + X402_WALLET = os.getenv("PRY_X402_WALLET", "pry-default-wallet") + X402_FACILITATOR_URL = os.getenv("PRY_X402_FACILITATOR", "https://x402.org/facilitator") X402_DEFAULT_NETWORK = os.getenv("PRY_X402_NETWORK", X402Network.BASE.value) X402_DEFAULT_ASSET = os.getenv("PRY_X402_ASSET", X402Asset.USDC.value) -X402_FACILITATOR_URL = os.getenv("PRY_X402_FACILITATOR", "https://x402.org/facilitator") X402_OFFLINE_MODE = os.getenv("PRY_X402_OFFLINE", "false").lower() in ("1", "true", "yes") X402_TIMEOUT_SECONDS = float(os.getenv("PRY_X402_TIMEOUT", "15"))