pheby-hermes-platform-adapter/plugin/pheby/config.py

"""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.<key>`` 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",
]