"""Pheby plugin configuration. Secrets come from environment variables (Hermes convention: ``~/.hermes/.env`` is loaded by Hermes itself before plugins load). Non-secret behavior lives in the platform's ``extra`` block in ``config.yaml`` under ``platforms.pheby``. Resolution precedence for every key (highest wins): 1. environment variable (secrets *must* come from env) 2. ``platforms.pheby.extra.`` in config.yaml 3. built-in default """ from __future__ import annotations import os from dataclasses import dataclass, field from typing import Any, Dict, Optional # Environment variable names ENV_SECRET = "PHEBY_SECRET" # shared credential (required to serve) ENV_BIND_HOST = "PHEBY_BIND_HOST" ENV_PORT = "PHEBY_PORT" ENV_DEBUG = "PHEBY_DEBUG" ENV_LOG_CHAT_CONTENT = "PHEBY_LOG_CHAT_CONTENT" # Defaults DEFAULT_BIND_HOST = "127.0.0.1" # safe behind a local Caddy reverse proxy DEFAULT_PORT = 8620 DEFAULT_RETENTION_DAYS = 7 DEFAULT_MAX_SECRET_LEN = 1024 @dataclass class PhebyConfig: """Resolved runtime configuration for the Pheby server + adapter.""" # Shared secret for WS/HTTP auth. Empty disables the adapter entirely. secret: str = "" bind_host: str = DEFAULT_BIND_HOST port: int = DEFAULT_PORT # Attachment storage directory; a per-instance subdir is created inside. storage_dir: str = "" # Attachment retention in days (0 = keep forever — not recommended). retention_days: int = DEFAULT_RETENTION_DAYS # Verbose protocol logging (still never logs secrets). debug: bool = False # When True, debug logs MAY include chat text and tool previews. log_chat_content: bool = False # Path to a persistent JSON index of registered attachments. Empty = # derive from storage_dir. index_path: str = "" # Extra config passthrough (whole ``extra`` dict) for future keys. extra: Dict[str, Any] = field(default_factory=dict) # ------------------------------------------------------------------ @property def enabled(self) -> bool: """The adapter only serves when a secret is configured.""" return bool(self.secret) @property def attachments_root(self) -> str: if self.storage_dir: return self.storage_dir # Resolved lazily by the attachment store (needs get_hermes_home). return "" def _env_bool(name: str) -> bool: return os.getenv(name, "").strip().lower() in ("1", "true", "yes", "on") def _coerce_int(value: Any, default: int) -> int: try: return int(value) except (TypeError, ValueError): return default def load_config(extra: Optional[Dict[str, Any]] = None) -> PhebyConfig: """Build a :class:`PhebyConfig` from env + ``platforms.pheby.extra``. Environment variables win over YAML ``extra`` keys. Never raises. """ extra = dict(extra or {}) def _pick(env_name: str, key: str, default: Any = "") -> Any: env = os.getenv(env_name, "") if env.strip(): return env.strip() val = extra.get(key) if val is None or (isinstance(val, str) and not val.strip()): return default return val secret = os.getenv(ENV_SECRET, "").strip() if not secret: # A YAML secret is allowed for local testing but strongly discouraged; # env always wins and docs recommend env-only. secret = str(extra.get("secret", "") or "").strip() if len(secret) > DEFAULT_MAX_SECRET_LEN: secret = secret[:DEFAULT_MAX_SECRET_LEN] port = _coerce_int( _pick(ENV_PORT, "port", DEFAULT_PORT), DEFAULT_PORT) if not (0 < port < 65536): port = DEFAULT_PORT retention = _coerce_int( _pick("", "attachment_retention_days", DEFAULT_RETENTION_DAYS), DEFAULT_RETENTION_DAYS) if retention < 0: retention = DEFAULT_RETENTION_DAYS debug = _env_bool(ENV_DEBUG) or bool(extra.get("debug", False)) log_chat = _env_bool(ENV_LOG_CHAT_CONTENT) or bool( extra.get("log_chat_content", False)) cfg = PhebyConfig( secret=secret, bind_host=str(_pick(ENV_BIND_HOST, "bind_host", DEFAULT_BIND_HOST)), port=port, storage_dir=str(_pick("", "attachment_storage_dir", "") or ""), retention_days=retention, debug=bool(debug), log_chat_content=bool(log_chat), index_path=str(_pick("", "attachment_index_path", "") or ""), extra=extra, ) return cfg def check_requirements() -> bool: """Platform-entry dependency check: aiohttp available + secret set.""" try: import aiohttp # noqa: F401 except ImportError: return False return bool(os.getenv(ENV_SECRET, "").strip()) def validate_config(config: Any) -> bool: """Gateway config validation: at minimum a secret must be resolvable.""" extra = getattr(config, "extra", {}) or {} if os.getenv(ENV_SECRET, "").strip(): return True return bool(str(extra.get("secret", "") or "").strip()) def is_connected(config: Any) -> bool: """True when Pheby is configured (env or config.yaml).""" return validate_config(config) def env_enablement() -> Optional[Dict[str, Any]]: """Seed ``PlatformConfig.extra`` from env for env-only setups. Mirrors the ntfy adapter pattern so ``hermes gateway status`` reflects a PHEBY_SECRET-only deployment without instantiating the server. """ secret = os.getenv(ENV_SECRET, "").strip() if not secret: return None seed: Dict[str, Any] = {} host = os.getenv(ENV_BIND_HOST, "").strip() if host: seed["bind_host"] = host port = os.getenv(ENV_PORT, "").strip() if port: seed["port"] = port return seed __all__ = [ "ENV_SECRET", "ENV_BIND_HOST", "ENV_PORT", "ENV_DEBUG", "ENV_LOG_CHAT_CONTENT", "DEFAULT_BIND_HOST", "DEFAULT_PORT", "DEFAULT_RETENTION_DAYS", "PhebyConfig", "load_config", "check_requirements", "validate_config", "is_connected", "env_enablement", ]