hermes-agent/plugins/dashboard_auth/drain/__init__.py

"""DrainSecretProvider — shared-bearer-secret auth for the drain-control endpoint.

Uses the non-interactive token capability of the ``DashboardAuthProvider``
ABC (``supports_token`` / ``verify_token`` + the route-agnostic ``token_auth``
middleware seam). ``nous-account-service`` (NAS) provisions a **per-agent
unique** shared secret into each deployed agent's environment; this provider
verifies an inbound ``Authorization`` bearer against it with a constant-time
compare and vouches for the caller as the ``drain-control`` principal. No
login, cookie, session or refresh — the interactive ABC methods raise
``NotImplementedError``. A real auth plugin (not an ad-hoc header check on the
route) so the credential lives inside the dashboard auth framework.

Security properties:
* Per-agent unique secret — a leak's blast radius is one agent.
* Fail-CLOSED entropy gate at registration: a weak/short/low-entropy secret is
  never silently accepted (>= 43 url-safe-base64 chars ~= 256 bits, enough
  distinct characters, Shannon entropy floor).
* ``hmac.compare_digest`` on the request path — not a timing oracle.

Configuration — the secret is a CREDENTIAL, so it is env-only (provisioned by
NAS at deploy time); behavioural knobs live in config.yaml::

    HERMES_DASHBOARD_DRAIN_SECRET   # per-agent shared secret (>=43 url-safe-b64 chars)

    dashboard:
      drain_auth:
        scope: drain            # capability label attached to the principal
        min_secret_chars: 43    # entropy bar (default 43 ~= 256 bits)

When the env var is unset the plugin is a no-op (records a skip reason).
"""
from __future__ import annotations

import hmac
import logging
import math
import os
from collections import Counter
from typing import Optional

from hermes_cli.dashboard_auth import (
    DashboardAuthProvider,
    LoginStart,
    Session,
    TokenPrincipal,
)
from plugins.dashboard_auth._shared import load_config_section

logger = logging.getLogger(__name__)

# token_urlsafe(32) produces exactly 43 chars, so a correctly-provisioned
# secret clears the default bar exactly.
_DEFAULT_MIN_SECRET_CHARS = 43
# Rejects degenerate values like "aaaa..." that are long but trivially low-entropy.
_MIN_DISTINCT_CHARS = 16
# Distribution-aware second guard on top of length + distinct-count.
_MIN_SHANNON_BITS = 128.0

# Kept here (not imported from web_server) to avoid a heavy import at plugin load.
DRAIN_ROUTE_PATH = "/api/gateway/drain"

LAST_SKIP_REASON: str = ""


def _shannon_bits(value: str) -> float:
    """Total Shannon entropy (bits) of ``value`` over its character distribution."""
    if not value:
        return 0.0
    n = len(value)
    per_char = -sum((c / n) * math.log2(c / n) for c in Counter(value).values())
    return per_char * n


def assess_secret_strength(
    secret: str, *, min_chars: int = _DEFAULT_MIN_SECRET_CHARS
) -> Optional[str]:
    """Return a human-readable rejection reason if ``secret`` is too weak, else ``None``.

    Checks, in order: length >= ``min_chars``, distinct chars >=
    ``_MIN_DISTINCT_CHARS``, Shannon entropy >= ``_MIN_SHANNON_BITS``.
    """
    if not secret:
        return "secret is empty"
    if len(secret) < min_chars:
        return (
            f"secret too short: {len(secret)} chars (need >= {min_chars}; "
            "use a >=256-bit value, e.g. `python -c \"import secrets; "
            "print(secrets.token_urlsafe(32))\"`)"
        )
    distinct = len(set(secret))
    if distinct < _MIN_DISTINCT_CHARS:
        return (
            f"secret has only {distinct} distinct characters (need >= "
            f"{_MIN_DISTINCT_CHARS}); looks structured/low-entropy"
        )
    bits = _shannon_bits(secret)
    if bits < _MIN_SHANNON_BITS:
        return (
            f"secret entropy too low: {bits:.0f} bits (need >= "
            f"{_MIN_SHANNON_BITS:.0f}); looks structured/repeated"
        )
    return None


class DrainSecretProvider(DashboardAuthProvider):
    """Non-interactive shared-bearer-secret provider for drain control."""

    name = "drain-secret"
    display_name = "Drain Control (service credential)"
    supports_token = True
    supports_session = False

    def __init__(self, *, secret: str, scope: str = "drain") -> None:
        # Defence in depth: construction enforces the entropy bar too, so a
        # caller bypassing register() still can't build a weak provider.
        reason = assess_secret_strength(secret)
        if reason is not None:
            raise ValueError(f"drain secret rejected: {reason}")
        self._secret = secret
        self._scope = scope or "drain"

    # ---- token capability (the only thing this provider implements) --------

    def verify_token(self, *, token: str) -> Optional[TokenPrincipal]:
        """Constant-time compare; ``drain-control`` principal on match, else
        ``None`` so the generic seam falls through / fails closed."""
        if token and hmac.compare_digest(token.encode("utf-8"), self._secret.encode("utf-8")):
            return TokenPrincipal(principal="drain-control", provider=self.name, scopes=(self._scope,))
        return None

    # ---- interactive methods: unsupported (service credential only) --------

    def start_login(self, *, redirect_uri: str) -> LoginStart:
        raise NotImplementedError(
            "DrainSecretProvider is a non-interactive service credential; there is no login flow."
        )

    def complete_login(
        self, *, code: str, state: str, code_verifier: str, redirect_uri: str
    ) -> Session:
        raise NotImplementedError("DrainSecretProvider is a non-interactive service credential.")

    def verify_session(self, *, access_token: str) -> Optional[Session]:
        # Never mints a Session, so never recognises a cookie. Return None
        # (don't raise) so it stacks harmlessly in the cookie-verify loop.
        return None

    def refresh_session(self, *, refresh_token: str) -> Session:
        raise NotImplementedError("DrainSecretProvider is a non-interactive service credential.")

    def revoke_session(self, *, refresh_token: str) -> None:
        return None


# ---- Plugin entry point ----

def _load_config_drain_auth_section() -> dict:
    return load_config_section(logger, "dashboard-auth-drain", "dashboard", "drain_auth")


def register(ctx) -> None:
    """Register ``DrainSecretProvider`` when a strong secret is set.

    No-op (records a skip reason) when ``HERMES_DASHBOARD_DRAIN_SECRET`` is
    unset or fails the entropy gate. On success also registers the drain
    route as token-authable via the generic seam.
    """
    global LAST_SKIP_REASON
    LAST_SKIP_REASON = ""

    secret = os.environ.get("HERMES_DASHBOARD_DRAIN_SECRET", "").strip()
    if not secret:
        LAST_SKIP_REASON = (
            "HERMES_DASHBOARD_DRAIN_SECRET is not set. Set a per-agent >=256-bit secret "
            "(e.g. `python -c \"import secrets; print(secrets.token_urlsafe(32))\"`) to enable "
            "NAS-driven drain coordination; leave it unset to disable the drain endpoint."
        )
        logger.debug("dashboard-auth-drain: %s", LAST_SKIP_REASON)
        return

    section = _load_config_drain_auth_section()
    scope = str(section.get("scope", "drain") or "drain").strip() or "drain"
    try:
        min_chars = int(section.get("min_secret_chars", _DEFAULT_MIN_SECRET_CHARS))
    except (TypeError, ValueError):
        min_chars = _DEFAULT_MIN_SECRET_CHARS

    reason = assess_secret_strength(secret, min_chars=min_chars)
    if reason is not None:
        LAST_SKIP_REASON = (
            f"HERMES_DASHBOARD_DRAIN_SECRET rejected — {reason}. "
            "The drain endpoint stays disabled (fail-closed)."
        )
        logger.warning("dashboard-auth-drain: %s", LAST_SKIP_REASON)
        return

    try:
        provider = DrainSecretProvider(secret=secret, scope=scope)
    except ValueError as exc:
        LAST_SKIP_REASON = f"DrainSecretProvider construction failed: {exc}"
        logger.warning("dashboard-auth-drain: %s", LAST_SKIP_REASON)
        return

    ctx.register_dashboard_auth_provider(provider)

    # Opt the drain endpoint into the token-auth seam so the interactive
    # cookie gate doesn't bounce NAS's bearer call.
    try:
        from hermes_cli.dashboard_auth.token_auth import register_token_route

        register_token_route(DRAIN_ROUTE_PATH)
    except Exception as exc:  # noqa: BLE001 — seam import must not crash plugin load
        logger.warning("dashboard-auth-drain: could not register token route %s: %s", DRAIN_ROUTE_PATH, exc)

    logger.info(
        "dashboard-auth-drain: registered drain service-credential provider (scope=%s, route=%s)",
        scope, DRAIN_ROUTE_PATH,
    )