ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 10 Packages Projects Releases Wiki Activity

Merge branch 'simp/r3-18-E' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-03 00:08:51 -07:00
parent 8d4a925c98 7443080d5a
commit cebbc9ef97
15 changed files with 546 additions and 901 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/commands.py
+65 -138
View File
@@ -10,7 +10,6 @@ from __future__ import annotations
import logging
import re
from dataclasses import dataclass
from typing import Any
from utils import is_truthy_value
from hermes_constants import INDICATOR_STYLES
@@ -21,7 +20,6 @@ logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class CommandDef:
"""Definition of a single slash command."""
name: str # canonical name without slash: "background"
description: str # human-readable description
category: str # "Session", "Configuration", etc.
@@ -31,21 +29,17 @@ class CommandDef:
cli_only: bool = False # only available in CLI
gateway_only: bool = False # only available in gateway/messaging
gateway_config_gate: str | None = None # config dotpath; truthy overrides cli_only for gateway
# Mid-run (agent busy) gateway behavior for gateway/run.py's Guard-2 dispatcher:
# "dispatch" — run while busy (normal handler, or the mid-run
# variant named by ``busy_handler``).
# "reject" — refuse mid-run; generic "Agent is running" unless
# ``busy_handler`` names a specific reject message.
# "interrupt_then_dispatch" — interrupt the agent first (/stop, /new, /reset);
# Guard 1 (platforms/base.py) routes these.
# Mid-run (agent busy) gateway behavior (gateway/run.py Guard-2 dispatcher): "dispatch" = run
# while busy (normal handler or the ``busy_handler`` variant); "reject" = refuse mid-run
# (generic "Agent is running" unless ``busy_handler`` names a reject message);
# "interrupt_then_dispatch" = interrupt first (/stop, /new, /reset; Guard 1, platforms/base.py).
busy_policy: str = "reject"
busy_handler: str | None = None # key of a special mid-run handler in Guard-2 table
# Key in ``hermes_cli.slash_exec.EXECUTORS`` (pure formatter of the core text). A
# string key, not a callable, keeps this module import-light for the gateway.
# Key in ``hermes_cli.slash_exec.EXECUTORS`` (a string, not a callable: keeps this module
# import-light for the gateway).
execute: str | None = None
argument_mode: str | None = None # desktop composer: options|text|mixed; None inferred
# Desktop availability: None = offered; "hidden" = runs but out of the popover;
# otherwise a reason (terminal / messaging / settings / …).
# Desktop availability: None = offered; "hidden" = runs but out of the popover; else a reason.
desktop: str | None = None
@@ -74,8 +68,7 @@ COMMAND_REGISTRY: list[CommandDef] = [
cli_only=True, args_hint="[initial text]", aliases=("compose",)),
CommandDef("undo", "Back up N user turns and re-prompt (default 1)", "Session",
args_hint="[N]"),
CommandDef("title", "Set a title for the current session", "Session",
args_hint="[name]"),
CommandDef("title", "Set a title for the current session", "Session", args_hint="[name]"),
CommandDef("handoff", "Hand off this session to a messaging platform (Telegram, Discord, etc.)", "Session",
args_hint="<platform>", cli_only=True, argument_mode="options"),
CommandDef("branch", "Branch the current session (explore a different path)", "Session",
@@ -97,8 +90,7 @@ COMMAND_REGISTRY: list[CommandDef] = [
CommandDef("stop", "Kill all running background processes", "Session",
busy_policy="interrupt_then_dispatch", busy_handler="stop"),
CommandDef("pause", "Pause new work globally (emergency stop); '/pause off' resumes", "Session",
gateway_only=True, args_hint="[reason | off]",
busy_policy="dispatch"),
gateway_only=True, args_hint="[reason | off]", busy_policy="dispatch"),
CommandDef("approve", "Approve a pending dangerous command", "Session",
gateway_only=True, args_hint="[session|always]", busy_policy="dispatch",
desktop="messaging"),
@@ -113,11 +105,9 @@ COMMAND_REGISTRY: list[CommandDef] = [
aliases=("tasks",), busy_policy="dispatch"),
CommandDef("journey", "Open the learning journey timeline",
"Session", aliases=("learning", "memory-graph"), cli_only=True,
args_hint="[list|delete <id>|edit <id>]",
subcommands=("list", "delete", "edit")),
args_hint="[list|delete <id>|edit <id>]", subcommands=("list", "delete", "edit")),
CommandDef("queue", "Queue a prompt for the next turn (doesn't interrupt)", "Session",
aliases=("q",), args_hint="<prompt>",
busy_policy="dispatch", busy_handler="queue"),
aliases=("q",), args_hint="<prompt>", busy_policy="dispatch", busy_handler="queue"),
CommandDef("steer", "Inject a message after the next tool call without interrupting", "Session",
args_hint="<prompt>", busy_policy="dispatch", busy_handler="steer"),
CommandDef("goal", "Set a standing goal Hermes works on across turns until achieved", "Session",
@@ -144,12 +134,10 @@ COMMAND_REGISTRY: list[CommandDef] = [
CommandDef("status", "Show session, model, token, and context info", "Session",
busy_policy="dispatch"),
CommandDef("egress", "Show Docker egress proxy status", "Session",
args_hint="[status]", subcommands=("status",),
busy_policy="dispatch", busy_handler="egress",
execute="egress"),
args_hint="[status]", subcommands=("status",), busy_policy="dispatch",
busy_handler="egress", execute="egress"),
CommandDef("context", "Show detailed context window view with usage gauge, category breakdown, compression stats, and throughput", "Session",
aliases=("ctx",), args_hint="[all]", subcommands=("all",),
busy_policy="dispatch"),
aliases=("ctx",), args_hint="[all]", subcommands=("all",), busy_policy="dispatch"),
CommandDef("whoami", "Show your slash command access (admin / user)", "Info"),
CommandDef("profile", "Show active profile name and home directory", "Info",
busy_policy="dispatch", execute="profile"),
@@ -157,8 +145,6 @@ COMMAND_REGISTRY: list[CommandDef] = [
gateway_only=True, aliases=("set-home",), desktop="terminal"),
CommandDef("resume", "Resume a previously-named session", "Session",
args_hint="[name]", argument_mode="mixed"),
# Configuration
CommandDef("sessions", "Browse and resume previous sessions", "Session"),
# Configuration
@@ -168,10 +154,8 @@ COMMAND_REGISTRY: list[CommandDef] = [
args_hint="[model] [--provider name] [--global|--session] [--refresh]",
busy_policy="reject", busy_handler="model", desktop="hidden"),
CommandDef("codex-runtime", "Toggle codex app-server runtime for OpenAI/Codex models",
"Configuration", aliases=("codex_runtime",),
args_hint="[auto|codex_app_server]",
"Configuration", aliases=("codex_runtime",), args_hint="[auto|codex_app_server]",
busy_policy="reject", busy_handler="codex-runtime"),
CommandDef("personality", "Set a predefined personality", "Configuration",
args_hint="[name]", argument_mode="options"),
CommandDef("statusbar", "Toggle the context/model status bar", "Configuration",
@@ -186,16 +170,14 @@ COMMAND_REGISTRY: list[CommandDef] = [
args_hint="[staged|all|session] [--stat] [path...]",
subcommands=("staged", "all", "session")),
CommandDef("verbose", "Cycle tool progress display: off -> new -> all -> verbose",
"Configuration", cli_only=True,
gateway_config_gate="display.tool_progress_command",
"Configuration", cli_only=True, gateway_config_gate="display.tool_progress_command",
busy_policy="dispatch", desktop="terminal"),
CommandDef("focus", "Toggle focus view — show only your prompt and the final response",
"Configuration", cli_only=True, args_hint="[on|off|status]",
subcommands=("on", "off", "status")),
CommandDef("footer", "Toggle gateway runtime-metadata footer on final replies",
"Configuration", args_hint="[on|off|status]",
subcommands=("on", "off", "status"), busy_policy="dispatch",
desktop="terminal"),
"Configuration", args_hint="[on|off|status]", subcommands=("on", "off", "status"),
busy_policy="dispatch", desktop="terminal"),
CommandDef("yolo", "Toggle YOLO mode (skip all dangerous command approvals)",
"Configuration", busy_policy="dispatch"),
CommandDef("approvals", "Show or set the persistent dangerous-command approval mode",
@@ -218,8 +200,7 @@ COMMAND_REGISTRY: list[CommandDef] = [
args_hint="[on|off|tts|status]", subcommands=("on", "off", "tts", "status"),
desktop="composer-voice"),
CommandDef("wake", "Toggle the 'Hey Hermes' wake word listener", "Configuration",
cli_only=True, args_hint="[on|off|status]",
subcommands=("on", "off", "status")),
cli_only=True, args_hint="[on|off|status]", subcommands=("on", "off", "status")),
CommandDef("busy", "Control how messages behave while Hermes is working", "Configuration",
args_hint="[queue|steer|interrupt|status]",
subcommands=("queue", "steer", "interrupt", "status"),
@@ -227,8 +208,7 @@ COMMAND_REGISTRY: list[CommandDef] = [
# Tools & Skills
CommandDef("tools", "Manage tools: /tools [list|disable|enable] [name...]", "Tools & Skills",
args_hint="[list|disable|enable] [name...]", cli_only=True,
argument_mode="options"),
args_hint="[list|disable|enable] [name...]", cli_only=True, argument_mode="options"),
CommandDef("toolsets", "List available toolsets", "Tools & Skills",
cli_only=True, desktop="terminal"),
CommandDef("skills", "Search, install, inspect, or manage skills",
@@ -238,8 +218,7 @@ COMMAND_REGISTRY: list[CommandDef] = [
"pending", "approve", "reject", "diff", "approval"),
desktop="settings"),
CommandDef("memory", "Review pending memory writes / toggle the approval gate",
"Tools & Skills",
args_hint="[pending|approve|reject|approval] [id|on|off]",
"Tools & Skills", args_hint="[pending|approve|reject|approval] [id|on|off]",
subcommands=("pending", "approve", "reject", "approval")),
CommandDef("bundles", "List skill bundles (aliases /<name> for multiple skills)",
"Tools & Skills", execute="bundles"),
@@ -321,13 +300,10 @@ COMMAND_REGISTRY: list[CommandDef] = [
# Exit
CommandDef("quit", "Exit the CLI (use --delete to also remove session history)", "Exit",
cli_only=True, aliases=("exit",), args_hint="[--delete]",
desktop="terminal"),
]
cli_only=True, aliases=("exit",), args_hint="[--delete]", desktop="terminal")]
# Distinguishes ``mixed`` (subcommands plus free-text) from ``options``. A bare
# ``args_hint`` with no subcommands is always ``text`` — no tokens needed for that.
# Distinguishes ``mixed`` (subcommands plus free-text) from ``options``; no subcommands => ``text``.
_PROSE_HINTS = ("<prompt>", "[text", "instructions", "[interval]", "<what")
@@ -347,12 +323,9 @@ def command_desktop_meta(cmd: CommandDef) -> dict[str, str | None]:
return {"argument_mode": infer_argument_mode(cmd), "desktop": cmd.desktop}
# --- Derived lookups (built once at import) ---------------------------------
# Every name and alias -> its CommandDef.
_COMMAND_LOOKUP: dict[str, CommandDef] = {
key: cmd for cmd in COMMAND_REGISTRY for key in (cmd.name, *cmd.aliases)
}
key: cmd for cmd in COMMAND_REGISTRY for key in (cmd.name, *cmd.aliases)}
def resolve_command(name: str) -> CommandDef | None:
@@ -361,21 +334,19 @@ def resolve_command(name: str) -> CommandDef | None:
def _build_description(cmd: CommandDef) -> str:
"""Build a CLI-facing description string including usage hint."""
"""CLI-facing description including the usage hint."""
if not cmd.args_hint:
return cmd.description
return f"{cmd.description} (usage: /{cmd.name} {cmd.args_hint})"
# Backwards-compatible flat dict ("/command" -> description) and the same
# grouped by category. Both exclude gateway_only commands.
# Flat "/command" -> description, and the same grouped by category; both exclude gateway_only.
COMMANDS: dict[str, str] = {}
COMMANDS_BY_CATEGORY: dict[str, dict[str, str]] = {}
# Subcommands lookup: "/cmd" -> ["sub1", ...]; explicit ``subcommands`` first (in
# registry order), then pipe patterns in args_hint ("[on|off|status]") as fallback.
SUBCOMMANDS: dict[str, list[str]] = {
f"/{_cmd.name}": list(_cmd.subcommands) for _cmd in COMMAND_REGISTRY if _cmd.subcommands
}
f"/{_cmd.name}": list(_cmd.subcommands) for _cmd in COMMAND_REGISTRY if _cmd.subcommands}
for _cmd in COMMAND_REGISTRY:
if _cmd.gateway_only:
continue
@@ -392,30 +363,19 @@ for _cmd in COMMAND_REGISTRY:
SUBCOMMANDS[f"/{_cmd.name}"] = _m.group(0).split("|")
# /help renderer sub-groups for the ~46-command "Session" category. Category
# itself is load-bearing for gateway help, so commands are not re-tagged;
# unlisted Session commands fall under the base header. Bare names (no /).
# /help sub-groups for the large "Session" category (category itself is load-bearing for gateway
# help, so commands are not re-tagged); unlisted Session commands fall under the base header.
HELP_SESSION_SUBGROUPS: dict[str, tuple[str, ...]] = {
"Context": (
"compress", "compact", "context", "ctx", "status",
),
"Context": ("compress", "compact", "context", "ctx", "status"),
"Background & Automation": (
"bg", "btw", "agents", "tasks", "queue", "q", "steer",
"goal", "subgoal", "heartbeat", "hb", "refine", "loop", "proactive",
"moa", "journey", "learning", "memory-graph",
),
}
# --- Gateway helpers --------------------------------------------------------
"bg", "btw", "agents", "tasks", "queue", "q", "steer", "goal", "subgoal", "heartbeat", "hb",
"refine", "loop", "proactive", "moa", "journey", "learning", "memory-graph")}
# All names + aliases the gateway dispatches. Config-gated commands are
# included; their handler checks the gate at runtime.
GATEWAY_KNOWN_COMMANDS: frozenset[str] = frozenset(
name
for cmd in COMMAND_REGISTRY
if not cmd.cli_only or cmd.gateway_config_gate
for name in (cmd.name, *cmd.aliases)
)
name for cmd in COMMAND_REGISTRY if not cmd.cli_only or cmd.gateway_config_gate
for name in (cmd.name, *cmd.aliases))
def is_gateway_known_command(name: str | None) -> bool:
@@ -424,33 +384,26 @@ def is_gateway_known_command(name: str | None) -> bool:
if not name:
return False
return name in GATEWAY_KNOWN_COMMANDS or any(
plugin_name == name for plugin_name, _d, _h in _iter_plugin_command_entries()
)
plugin_name == name for plugin_name, _d, _h in _iter_plugin_command_entries())
# Commands with explicit mid-run handling (busy_policy != "reject"). Kept
# under its historical name for introspection/tests; the real bypass set is
# every resolvable command (see should_bypass_active_session).
ACTIVE_SESSION_BYPASS_COMMANDS: frozenset[str] = frozenset(
cmd.name for cmd in COMMAND_REGISTRY if cmd.busy_policy != "reject"
)
cmd.name for cmd in COMMAND_REGISTRY if cmd.busy_policy != "reject")
def is_interrupt_then_dispatch(command_name: str | None) -> bool:
"""True when *command_name* (or alias) has busy_policy "interrupt_then_dispatch";
Guard 1 (gateway/platforms/base.py) routes these through the cancel-handoff path."""
"""Guard 1 (gateway/platforms/base.py) routes these through the cancel-handoff path."""
cmd = resolve_command(command_name) if command_name else None
return cmd is not None and cmd.busy_policy == "interrupt_then_dispatch"
def should_bypass_active_session(command_name: str | None) -> bool:
"""True for any resolvable slash command.
Every recognized command is dispatched mid-run (by its Guard-2 handler or the
"busy — wait or /stop first" catch-all), never queued: gateway.run's safety net
discards command text reaching the pending queue, so a queued /model would
interrupt the agent AND vanish with a zero-char response.
"""
"""True for any resolvable slash command: every recognized command is dispatched mid-run
(Guard-2 handler or the "busy" catch-all), never queued — gateway.run's safety net discards
command text reaching the pending queue, so a queued /model would vanish."""
return resolve_command(command_name) is not None if command_name else False
@@ -461,29 +414,17 @@ def _resolve_config_gates() -> set[str]:
if not gated:
return set()
try:
from hermes_cli.config import read_raw_config
from hermes_cli.config import cfg_get, read_raw_config
cfg = read_raw_config()
except Exception:
return set()
def _lookup(dotpath: str) -> Any:
val: Any = cfg
for key in dotpath.split("."):
if not isinstance(val, dict):
return None
val = val.get(key)
return val
return {
cmd.name for cmd in gated
if is_truthy_value(_lookup(cmd.gateway_config_gate), default=False)
}
return {cmd.name for cmd in gated
if is_truthy_value(cfg_get(cfg, *cmd.gateway_config_gate.split(".")), default=False)}
def _is_gateway_available(cmd: CommandDef, config_overrides: set[str] | None = None) -> bool:
"""True if *cmd* appears in gateway surfaces: not ``cli_only``, or its config gate
is truthy. Pass *config_overrides* from ``_resolve_config_gates()`` to avoid
re-reading config per command."""
"""Not ``cli_only``, or its config gate is truthy (*config_overrides* from
``_resolve_config_gates()`` avoids re-reading config per command)."""
if not cmd.cli_only:
return True
if not cmd.gateway_config_gate:
@@ -501,53 +442,39 @@ def gateway_help_lines() -> list[str]:
continue
args = f" {cmd.args_hint}" if cmd.args_hint else ""
# Skip internal aliases like reload_mcp (underscore variant of the name).
alias_parts = [
f"`/{a}`" for a in cmd.aliases
if not (a.replace("-", "_") == cmd.name.replace("-", "_") and a != cmd.name)
]
alias_parts = [f"`/{a}`" for a in cmd.aliases
if not (a.replace("-", "_") == cmd.name.replace("-", "_") and a != cmd.name)]
alias_note = f" (alias: {', '.join(alias_parts)})" if alias_parts else ""
lines.append(f"`/{cmd.name}{args}` -- {cmd.description}{alias_note}")
return lines
def _iter_plugin_command_entries() -> list[tuple[str, str, str]]:
"""(name, description, args_hint) for plugin slash commands registered via
``PluginContext.register_command``; surfaced like ``CommandDef`` entries. Lazy so
importing this module never forces plugin discovery."""
"""(name, description, args_hint) for ``PluginContext.register_command`` slash commands.
Lazy so importing this module never forces plugin discovery."""
try:
from hermes_cli.plugins import get_plugin_commands
commands = get_plugin_commands() or {}
except Exception:
return []
return [
(name, str(meta.get("description") or f"Run /{name}"),
str(meta.get("args_hint") or "").strip())
for name, meta in commands.items()
if isinstance(name, str) and isinstance(meta, dict)
]
return [(name, str(meta.get("description") or f"Run /{name}"),
str(meta.get("args_hint") or "").strip())
for name, meta in commands.items() if isinstance(name, str) and isinstance(meta, dict)]
# --- Lazy re-exports (PEP 562) ----------------------------------------------
# Platform derivations and the prompt_toolkit completer live in sibling modules;
# their names stay importable from here so ``from hermes_cli.commands import X``
# and ``patch("hermes_cli.commands.X")`` keep working, while this registry module
# stays import-light for the gateway (no prompt_toolkit, no plugin discovery).
_LAZY_EXPORTS: dict[str, str] = {
name: "hermes_cli.commands_platforms"
for name in (
"_CMD_NAME_LIMIT", "_DEFAULT_TELEGRAM_MENU_MAX_COMMANDS", "_SLACK_RESERVED_COMMANDS",
"_SLACK_VIA_HERMES_ONLY", "_TELEGRAM_MENU_PRIORITY", "_clamp_command_names",
"_collect_gateway_skill_entries", "_iter_gateway_skills", "_prioritize_telegram_menu_candidates",
"_requires_argument", "_sanitize_slack_name", "_sanitize_telegram_name",
"_telegram_command_menu_config", "_truncate_desc", "discord_skill_commands_by_category",
"slack_app_manifest", "slack_native_slashes", "slack_subcommand_map", "telegram_bot_commands",
"telegram_menu_commands", "telegram_menu_max_commands",
)
}
_LAZY_EXPORTS.update({
name: "hermes_cli.commands_completion"
for name in ("SlashCommandCompleter", "SlashCommandAutoSuggest", "_file_size_label", "_short_desc")
})
# Lazy re-exports (PEP 562): platform derivations and the prompt_toolkit completer live in
# sibling modules; their names stay importable here (``from hermes_cli.commands import X``,
# ``patch("hermes_cli.commands.X")``) while this module stays import-light for the gateway.
_LAZY_EXPORTS: dict[str, str] = {name: "hermes_cli.commands_platforms" for name in (
"_CMD_NAME_LIMIT", "_DEFAULT_TELEGRAM_MENU_MAX_COMMANDS", "_SLACK_RESERVED_COMMANDS",
"_SLACK_VIA_HERMES_ONLY", "_TELEGRAM_MENU_PRIORITY", "_clamp_command_names",
"_collect_gateway_skill_entries", "_iter_gateway_skills", "_prioritize_telegram_menu_candidates",
"_requires_argument", "_sanitize_slack_name", "_sanitize_telegram_name",
"_telegram_command_menu_config", "_truncate_desc", "discord_skill_commands_by_category",
"slack_app_manifest", "slack_native_slashes", "slack_subcommand_map", "telegram_bot_commands",
"telegram_menu_commands", "telegram_menu_max_commands")}
_LAZY_EXPORTS.update({name: "hermes_cli.commands_completion" for name in (
"SlashCommandCompleter", "SlashCommandAutoSuggest", "_file_size_label", "_short_desc")})
def __getattr__(name: str):
hermes_cli/commands_platforms.py
+82 -134
View File
@@ -48,12 +48,9 @@ def _truncate_desc(desc: str, limit: int) -> str:
def _clamp_command_names(
entries: Sequence[tuple[str, ...]], reserved: set[str]) -> list[tuple[str, ...]]:
"""Enforce the 32-char Telegram/Discord name limit with collision avoidance.
Over-long names are truncated; if that collides with *reserved* or an
earlier entry, a 31-char prefix + digit ``0``-``9`` is tried, and the entry
is silently dropped when all ten are taken. Duplicate names are dropped.
Elements beyond ``(name, desc)`` pass through unchanged.
"""Enforce the 32-char Telegram/Discord name limit: over-long names are truncated; on a
collision with *reserved* or an earlier entry a 31-char prefix + digit ``0``-``9`` is tried,
then the entry is silently dropped. Duplicates are dropped; extra tuple elements pass through.
"""
used: set[str] = set(reserved)
result: list = []
@@ -79,14 +76,9 @@ def _clamp_command_names(
# --- Telegram ---------------------------------------------------------------
def telegram_bot_commands(*, include_plugins: bool = True) -> list[tuple[str, str]]:
"""Return (command_name, description) pairs for Telegram setMyCommands.
Names are Telegram-sanitized; aliases are skipped (one entry per canonical
command). Built-ins requiring arguments are **included** (their handlers show
usage when selected bare); plugin commands requiring arguments are **excluded**
because plugins may lack a no-arg fallback. Callers needing source metadata
pass ``include_plugins=False`` and use :func:`_collect_gateway_skill_entries`.
"""
"""(command_name, description) pairs for Telegram setMyCommands: sanitized canonical names
only (no aliases). Built-ins needing arguments are included (their handlers show usage when
selected bare); plugin commands needing arguments are excluded (may lack a no-arg fallback)."""
pairs = [(cmd.name, cmd.description) for cmd in _gateway_available_commands()]
if include_plugins:
pairs += [(n, d) for n, d, hint in _iter_plugin_command_entries()
@@ -94,28 +86,22 @@ def telegram_bot_commands(*, include_plugins: bool = True) -> list[tuple[str, st
return [(tg, desc) for name, desc in pairs if (tg := _sanitize_telegram_name(name))]
# Telegram allows 100 BotCommands; the 60-slot default keeps every built-in plus
# common skill commands visible under the ~4KB payload limit. Tunable via
# platforms.telegram.extra.command_menu.max_commands.
# Telegram allows 100 BotCommands; the 60-slot default keeps every built-in plus common skill
# commands under the ~4KB payload limit (platforms.telegram.extra.command_menu.max_commands).
_DEFAULT_TELEGRAM_MENU_MAX_COMMANDS = 60
_TELEGRAM_BOT_API_MAX_COMMANDS = 100
# priority_mode -> rank tables consulted in order ("configured" = user list,
# "default" = _TELEGRAM_MENU_PRIORITY); unranked candidates keep stable order after.
_TELEGRAM_PRIORITY_TIERS: dict[str, tuple[str, ...]] = {
"prepend": ("configured", "default"),
"append": ("default", "configured"),
"prepend": ("configured", "default"), "append": ("default", "configured"),
"replace": ("configured",)}
# Built-ins that must survive Telegram's small visible menu cap; everything
# else stays dispatchable when typed manually. Order = rank.
# Built-ins that must survive Telegram's small visible menu cap (everything else stays
# dispatchable when typed). Order = rank: everyday, maintenance, mid-turn control, operational.
_TELEGRAM_MENU_PRIORITY = (
# Most-typed everyday commands first.
"help", "new", "stop", "status", "egress", "resume", "sessions", "model",
# Maintenance / diagnostics.
"debug", "restart", "update", "verbose", "commands",
# Mid-turn session control.
"approve", "deny", "queue", "steer", "bg", "btw",
# Lower-priority but still useful operational built-ins.
"reasoning", "usage", "platforms", "platform", "profile", "whoami")
@@ -129,7 +115,6 @@ def _telegram_command_menu_config() -> dict[str, Any]:
for key in ("platforms", "telegram", "extra", "command_menu"):
node = node.get(key) if isinstance(node, Mapping) else None
menu_cfg: Mapping[str, Any] = node if isinstance(node, Mapping) else {}
try:
max_commands = int(menu_cfg.get("max_commands", _DEFAULT_TELEGRAM_MENU_MAX_COMMANDS))
except (TypeError, ValueError):
@@ -163,12 +148,11 @@ def _sanitized_rank(raw_names: Sequence[str]) -> dict[str, int]:
def _prioritize_telegram_menu_candidates(
candidates: list[tuple[str, str, str, str]]) -> list[tuple[str, str, str, str]]:
"""Order ``(final_name, description, source, raw_name)`` candidates; the default
priority applies to core only, "replace" mode ignores it entirely. ``raw_name``
is the pre-clamp name so a configured long command stays addressable."""
"""Order ``(final_name, description, source, raw_name)`` candidates; the default priority
applies to core only, "replace" mode ignores it. ``raw_name`` is the pre-clamp name so a
configured long command stays addressable."""
# Lazy origin import: tests patch ``hermes_cli.commands._telegram_command_menu_config``.
from hermes_cli.commands import _telegram_command_menu_config as menu_config
menu_cfg = menu_config()
configured_rank = _sanitized_rank(menu_cfg["priority"])
default_rank = _sanitized_rank(_TELEGRAM_MENU_PRIORITY)
@@ -192,14 +176,12 @@ def _prioritize_telegram_menu_candidates(
def _iter_gateway_skills(platform: str):
"""Yield ``(cmd_key, info, rel_parts)`` for skills eligible as gateway slash commands.
Scan roots: local ``SKILLS_DIR`` plus configured ``skills.external_dirs`` /
trusted project skills dirs; anything elsewhere or under ``SKILLS_DIR/.hub``
is skipped, as are skills disabled for *platform*. Paths are resolved on both
sides (symlinked roots such as macOS ``/var`` -> ``/private/var`` still match)
and matched per path component (``/my-skills`` never admits
``/my-skills-extra``). ``rel_parts`` is the skill dir relative to its root
(``("creative", "ascii-art")``) for category derivation. Iterates
``sorted(skill_cmds)`` so first-wins collision handling is alphabetical.
Scan roots: ``SKILLS_DIR`` plus ``skills.external_dirs`` / trusted project skills dirs;
anything elsewhere or under ``SKILLS_DIR/.hub`` is skipped, as are skills disabled for
*platform*. Paths are resolved on both sides (symlinked roots still match) and matched per
path component (``/my-skills`` never admits ``/my-skills-extra``). ``rel_parts`` is the skill
dir relative to its root (``("creative", "ascii-art")``). Alphabetical so first-wins
collision handling is deterministic.
"""
from pathlib import Path
@@ -223,7 +205,6 @@ def _iter_gateway_skills(platform: str):
continue
except Exception:
pass
skill_cmds = get_skill_commands()
for cmd_key in sorted(skill_cmds):
info = skill_cmds[cmd_key]
@@ -245,45 +226,41 @@ def _collect_gateway_skill_entries(
) -> tuple[list[tuple[str, str, str, str]], int]:
"""Collect plugin + skill entries for a gateway platform.
Plugin slash commands come first and are never trimmed; skill commands
(alphabetical, see :func:`_iter_gateway_skills`) fill the remaining
*max_slots* — ``None`` returns every candidate for a caller applying its own
cap. *reserved_names* (built-in names) is mutated in place as names are
claimed. *sanitize_name* runs before clamping and may return "" to skip an
entry. Returns ``(entries, hidden_count)`` with entries of
``(name, description, cmd_key, raw_name)`` — ``cmd_key`` is the original skill
key ("" for plugins); ``raw_name`` the sanitized pre-clamp name used for
configured-priority matching.
Plugin slash commands come first and are never trimmed; skill commands (alphabetical) fill
the remaining *max_slots* (``None`` = every candidate, caller caps). *reserved_names* is
mutated in place as names are claimed; *sanitize_name* runs before clamping and may return
"" to skip. Returns ``(entries, hidden_count)``, entries ``(name, description, cmd_key,
raw_name)`` — ``cmd_key`` "" for plugins; ``raw_name`` is the sanitized pre-clamp name used
for configured-priority matching (both survive a clamp-induced rename).
"""
sanitize = sanitize_name or (lambda n: n)
# Tier 1: plugin slash commands (never trimmed). No cmd_key — "" placeholder.
plugin_entries: list[tuple[str, str, str, str]] = []
try:
def _entries(rows) -> list[tuple[str, str, str, str]]:
"""Sanitize + truncate ``(raw_name, description, cmd_key)`` rows, then clamp against
``reserved_names``; any failure in the lazy source yields the rows collected so far."""
out: list[tuple[str, str, str, str]] = []
try:
for raw, desc, cmd_key in rows:
if name := sanitize(raw):
out.append((name, _truncate_desc(desc, desc_limit), cmd_key, name))
except Exception:
pass
return _clamp_command_names(out, reserved_names)
def _plugin_rows():
from hermes_cli.plugins import get_plugin_commands
plugin_cmds = get_plugin_commands()
for cmd_name in sorted(plugin_cmds):
meta = plugin_cmds[cmd_name]
if platform == "telegram" and _requires_argument(str(meta.get("args_hint") or "")):
continue
if name := sanitize(cmd_name):
desc = _truncate_desc(meta.get("description", "Plugin command"), desc_limit)
plugin_entries.append((name, desc, "", name))
except Exception:
pass
plugin_entries = _clamp_command_names(plugin_entries, reserved_names)
reserved_names.update(n for n, *_rest in plugin_entries)
yield cmd_name, meta.get("description", "Plugin command"), ""
# Tier 2: skill commands (the only tier trimmed at the cap). cmd_key and
# raw_name survive any clamp-induced rename.
skill_entries: list[tuple[str, str, str, str]] = []
try:
for cmd_key, info, _rel in _iter_gateway_skills(platform):
if name := sanitize(cmd_key.lstrip("/")):
desc = _truncate_desc(info.get("description", ""), desc_limit)
skill_entries.append((name, desc, cmd_key, name))
except Exception:
pass
skill_entries = _clamp_command_names(skill_entries, reserved_names)
plugin_entries = _entries(_plugin_rows())
reserved_names.update(n for n, *_rest in plugin_entries)
skill_entries = _entries(
(cmd_key.lstrip("/"), info.get("description", ""), cmd_key)
for cmd_key, info, _rel in _iter_gateway_skills(platform))
if max_slots is None:
return plugin_entries + skill_entries, 0
@@ -293,24 +270,17 @@ def _collect_gateway_skill_entries(
def telegram_menu_commands(max_commands: int = 100) -> tuple[list[tuple[str, str]], int]:
"""Return ``(menu_commands, hidden_count)`` for Telegram, capped to the Bot API limit.
Tier order: core CommandDefs, plugin slash commands, skill commands
(alphabetical; hub and telegram-disabled skills excluded). Tiers keep their
relative order unless named in ``platforms.telegram.extra.command_menu.priority``
— applied to the combined list *before* the cap, so a prioritized dynamic
command can displace an unprioritized core command.
"""
"""``(menu_commands, hidden_count)`` for Telegram, capped to the Bot API limit. Tier order:
core CommandDefs, plugin slash commands, skill commands (alphabetical; hub and
telegram-disabled skills excluded). Tiers keep relative order unless named in
``platforms.telegram.extra.command_menu.priority`` — applied *before* the cap, so a
prioritized dynamic command can displace an unprioritized core command."""
# Lazy origin import: tests patch ``hermes_cli.commands.telegram_bot_commands``.
from hermes_cli.commands import telegram_bot_commands as bot_commands
core_commands = list(bot_commands(include_plugins=False))
entries, hidden_count = _collect_gateway_skill_entries(
platform="telegram",
max_slots=None,
reserved_names={n for n, _ in core_commands},
desc_limit=40,
sanitize_name=_sanitize_telegram_name)
platform="telegram", max_slots=None, reserved_names={n for n, _ in core_commands},
desc_limit=40, sanitize_name=_sanitize_telegram_name)
candidates = [(name, desc, "core", name) for name, desc in core_commands]
candidates += [(name, desc, "skill" if cmd_key else "plugin", raw)
for name, desc, cmd_key, raw in entries]
@@ -325,27 +295,23 @@ def telegram_menu_commands(max_commands: int = 100) -> tuple[list[tuple[str, str
def discord_skill_commands_by_category(
reserved_names: set[str],
) -> tuple[dict[str, list[tuple[str, str, str]]], list[tuple[str, str, str]], int]:
"""Return ``(categories, uncategorized, hidden_count)`` for Discord ``/skill`` autocomplete.
"""``(categories, uncategorized, hidden_count)`` for Discord ``/skill`` autocomplete.
Skills nested >= 2 levels under a scan root (``creative/ascii-art/SKILL.md``)
are grouped under ``categories[top_level]``; root-level skills are
*uncategorized*. Entries are ``(name, description, cmd_key)`` with names
clamped to 32 chars and descriptions to 100; eligibility follows
:func:`_iter_gateway_skills`. No per-group cap — the caller flattens everything
into one autocomplete callback; ``hidden_count`` only reports 32-char clamp
collisions against reserved names or earlier skills.
Skills nested >= 2 levels under a scan root (``creative/ascii-art/SKILL.md``) group under
``categories[top_level]``; root-level skills are *uncategorized*. Entries are
``(name, description, cmd_key)``, names clamped to 32 chars, descriptions to 100. No
per-group cap (the caller flattens into one autocomplete callback); ``hidden_count`` only
reports 32-char clamp collisions against reserved names or earlier skills.
"""
categories: dict[str, list[tuple[str, str, str]]] = {}
uncategorized: list[tuple[str, str, str]] = []
# clamped name -> origin. Reserved (gateway-builtin) names carry a sentinel so
# the warning distinguishes "collided with a reserved command" from "two
# skills collided on the 32-char clamp" — the rename-worthy case.
# clamped name -> origin; reserved names carry a sentinel so the warning distinguishes a
# reserved-command collision from two skills colliding on the clamp (the rename-worthy case).
names_used: dict[str, str] = dict.fromkeys(reserved_names, "<reserved>")
hidden = 0
try:
for cmd_key, info, rel_parts in _iter_gateway_skills("discord"):
# On collision the first (alphabetical) skill wins and the loser is
# dropped from the picker; warn loudly so skill authors can discover it.
# First (alphabetical) skill wins; the loser is dropped from the picker — warn loudly.
discord_name = cmd_key.lstrip("/")[:32]
prior = names_used.get(discord_name)
if prior == "<reserved>":
@@ -369,7 +335,6 @@ def discord_skill_commands_by_category(
continue
names_used[discord_name] = cmd_key
entry = (discord_name, _truncate_desc(info.get("description", ""), 100), cmd_key)
# creative/ascii-art/SKILL.md -> category "creative"; root-level = uncategorized.
if len(rel_parts) >= 2:
categories.setdefault(rel_parts[0], []).append(entry)
else:
@@ -381,27 +346,22 @@ def discord_skill_commands_by_category(
# --- Slack native slash commands --------------------------------------------
# Slack slash names: lowercase a-z, 0-9, hyphens, underscores, max 32 chars;
# an app manifest accepts up to 50 slash commands.
# Slack slash names: lowercase a-z, 0-9, hyphens, underscores, max 32 chars; an app manifest
# accepts up to 50 slash commands. Reserved = Slack built-ins apps cannot register
# (https://slack.com/help/articles/201259356-Use-built-in-slash-commands).
_SLACK_MAX_SLASH_COMMANDS = 50
_SLACK_NAME_LIMIT = 32
_SLACK_INVALID_CHARS = re.compile(r"[^a-z0-9_\-]")
_SLACK_RESERVED_COMMANDS = frozenset({
# Built-in Slack slash commands that cannot be registered by apps.
# https://slack.com/help/articles/201259356-Use-built-in-slash-commands
"me", "status", "away", "dnd", "shrug", "remind", "msg", "feed",
"who", "collapse", "expand", "leave", "join", "open", "search",
"topic", "mute", "pro", "shortcuts"})
"me", "status", "away", "dnd", "shrug", "remind", "msg", "feed", "who", "collapse", "expand",
"leave", "join", "open", "search", "topic", "mute", "pro", "shortcuts"})
# Canonical commands intentionally NOT given a native Slack slash slot: Slack caps
# apps at 50 and the registry is at that ceiling, so rather than let the clamp
# silently drop whichever command sorts last (breaking the Telegram-parity test),
# low-frequency commands go through ``/hermes <command>`` on Slack only (native
# everywhere else). Rule: when a new canonical command tips past the cap, demote a
# rarer one-off lookup here (version, whoami, platform, diff, update, ...) rather
# than a recurring interactive surface (context, loop, save, approvals). Keep
# TIGHT — the parity test reads this set. Aliases are never pinned ahead of
# canonicals (/bg and /btw became canonical so they win first-pass slots).
# Canonical commands deliberately routed through ``/hermes <command>`` on Slack only: the registry
# sits at Slack's 50-slash cap, so rather than let the clamp silently drop whichever command sorts
# last (breaking the Telegram-parity test), low-frequency ones are demoted here. Rule: when a new
# canonical command tips past the cap, demote a rarer one-off lookup (version, whoami, diff, ...)
# rather than a recurring interactive surface (context, loop, save, approvals). Keep TIGHT — the
# parity test reads this set. Aliases are never pinned ahead of canonicals.
_SLACK_VIA_HERMES_ONLY = frozenset({
"topup", "moa", "debug", "egress", "init", "version", "diff", "update", "heartbeat",
"refine", "review", "pause", "whoami", "platform", "insights"})
@@ -413,16 +373,11 @@ def _sanitize_slack_name(raw: str) -> str:
def slack_native_slashes() -> list[tuple[str, str, str]]:
"""Return (slash_name, description, usage_hint) triples for Slack.
Every gateway-available command (canonical names first, then aliases, then
plugin commands) becomes a standalone Slack slash, clamped to the 50-command
cap with duplicate avoidance. Names colliding with a Slack built-in
(``/status``, ``/me``, ...) or in _SLACK_VIA_HERMES_ONLY are skipped.
``/hermes`` is always first so ``/hermes <command>`` works for anything dropped.
"""
"""(slash_name, description, usage_hint) triples for Slack: every gateway-available command
(canonical names first so they win slots at the cap, then aliases, then plugins) becomes a
standalone slash, deduped and clamped to the 50-command cap; Slack built-ins and
_SLACK_VIA_HERMES_ONLY are skipped. ``/hermes`` is always first for anything dropped."""
available = _gateway_available_commands()
# Canonical names first so they win slots at the cap; aliases second; plugins third.
wanted = [(cmd.name, cmd.description, cmd.args_hint or "") for cmd in available]
wanted += [(alias, f"Alias for /{cmd.name} — {cmd.description}", cmd.args_hint or "")
for cmd in available for alias in cmd.aliases]
@@ -433,12 +388,9 @@ def slack_native_slashes() -> list[tuple[str, str, str]]:
seen = {"hermes"}
for name, desc, hint in wanted:
slack_name = _sanitize_slack_name(name)
if (
not slack_name
or slack_name in seen
or slack_name in _SLACK_RESERVED_COMMANDS
or slack_name in _SLACK_VIA_HERMES_ONLY
or len(entries) >= _SLACK_MAX_SLASH_COMMANDS):
if (not slack_name or slack_name in seen or slack_name in _SLACK_RESERVED_COMMANDS
or slack_name in _SLACK_VIA_HERMES_ONLY
or len(entries) >= _SLACK_MAX_SLASH_COMMANDS):
continue
# Slack description cap is 2000 chars; keep it short.
entries.append((slack_name, desc[:140], hint[:100]))
@@ -448,12 +400,8 @@ def slack_native_slashes() -> list[tuple[str, str, str]]:
def slack_app_manifest(
request_url: str = "https://hermes-agent.local/slack/commands") -> dict[str, Any]:
"""Return the ``features.slash_commands`` manifest portion for all gateway slashes.
``request_url`` is schema-required but ignored in Socket Mode (a placeholder is
fine). Only this portion is returned so we stay decoupled from the rest of the
manifest users configure once in the Slack UI.
"""
"""``features.slash_commands`` manifest portion only (decoupled from the rest of the manifest
users configure in the Slack UI); ``request_url`` is schema-required, ignored in Socket Mode."""
slashes = []
for name, desc, usage in slack_native_slashes():
entry = {"command": f"/{name}", "description": desc or f"Run /{name}",
hermes_cli/dashboard_auth/__init__.py
+7 -14
View File
@@ -1,24 +1,17 @@
"""Dashboard authentication provider framework.
The auth gate engages only when the dashboard binds to a non-loopback host
without ``--insecure``; every request must then carry a verified session from
a registered ``DashboardAuthProvider``. The Nous provider
(``plugins/dashboard-auth-nous/``) is the default; third parties register
theirs via the plugin hook ``ctx.register_dashboard_auth_provider``.
"""
"""Dashboard authentication provider framework. The auth gate engages only when the dashboard
binds to a non-loopback host without ``--insecure``; every request must then carry a verified
session from a registered ``DashboardAuthProvider`` (Nous provider is the default; third parties
register theirs via the plugin hook ``ctx.register_dashboard_auth_provider``)."""
from hermes_cli.dashboard_auth.base import (
DashboardAuthProvider, Session, TokenPrincipal, LoginStart, InvalidCodeError,
InvalidCredentialsError, ProviderError, RefreshExpiredError, assert_protocol_compliance,
classify_jwks_lookup_error,
)
classify_jwks_lookup_error)
from hermes_cli.dashboard_auth.registry import (
register_provider, get_provider, list_providers, list_token_providers,
list_session_providers, clear_providers,
)
list_session_providers, clear_providers)
__all__ = [
"DashboardAuthProvider", "Session", "TokenPrincipal", "LoginStart", "InvalidCodeError",
"InvalidCredentialsError", "ProviderError", "RefreshExpiredError", "assert_protocol_compliance",
"classify_jwks_lookup_error", "register_provider", "get_provider", "list_providers",
"list_token_providers", "list_session_providers", "clear_providers",
]
"list_token_providers", "list_session_providers", "clear_providers"]
hermes_cli/dashboard_auth/audit.py
+7 -16
View File
@@ -1,10 +1,7 @@
"""Audit log for dashboard-auth events.
Profile-aware location: ``$HERMES_HOME/logs/dashboard-auth.log``, one JSON
object per line. Token-like fields are stripped before serialisation so
refresh tokens / JWTs never reach disk. Minimal import surface (no
``hermes_constants`` at import time) so early-loading middleware can import it.
"""
"""Audit log for dashboard-auth events: ``$HERMES_HOME/logs/dashboard-auth.log``, one JSON object
per line. Token-like fields are stripped before serialisation so refresh tokens / JWTs never
reach disk. Minimal import surface (no ``hermes_constants`` at import time) so early-loading
middleware can import it."""
from __future__ import annotations
import datetime as _dt
@@ -26,7 +23,6 @@ _REDACTED_FIELDS: frozenset = frozenset({
class AuditEvent(enum.Enum):
"""Event types; values are the literal ``event`` field on the JSON line."""
LOGIN_START = "login_start"
LOGIN_SUCCESS = "login_success"
LOGIN_FAILURE = "login_failure"
@@ -47,19 +43,14 @@ class AuditEvent(enum.Enum):
def _resolve_log_path() -> Path:
"""``$HERMES_HOME/logs/dashboard-auth.log`` (lazy leaf import: honours
profile overrides and the native-Windows ``%LOCALAPPDATA%`` fallback)."""
"""Lazy leaf import: honours profile overrides + the native-Windows ``%LOCALAPPDATA%`` fallback."""
from hermes_constants import get_hermes_home
return get_hermes_home() / "logs" / "dashboard-auth.log"
def audit_log(event: AuditEvent, **fields: Any) -> None:
"""Append one event; token-like fields dropped, log dir created.
Write failures are logged at WARNING but never raise — auth must not
fail because the audit logger broke.
"""
"""Append one event; token-like fields dropped, log dir created. Write failures are logged at
WARNING but never raise — auth must not fail because the audit logger broke."""
entry = {
"ts": _dt.datetime.now(_dt.timezone.utc).isoformat(),
"event": event.value,
hermes_cli/dashboard_auth/base.py
+45 -80
View File
@@ -8,12 +8,8 @@ from typing import Optional
@dataclass(frozen=True)
class Session:
"""A verified interactive identity (from ``complete_login`` / ``verify_session``).
All fields mandatory; providers without orgs set ``org_id=""``. The tokens
are opaque to Hermes.
"""
"""A verified interactive identity (from ``complete_login`` / ``verify_session``). All fields
mandatory; providers without orgs set ``org_id=""``. The tokens are opaque to Hermes."""
user_id: str
email: str
display_name: str
@@ -26,13 +22,9 @@ class Session:
@dataclass(frozen=True)
class TokenPrincipal:
"""A verified non-interactive (service-to-service) caller — the token analog
of :class:`Session`: one bearer token on one request, no login/cookie/refresh.
``principal`` is an opaque stable caller id; ``scopes`` empty means
"unscoped" (a route MAY enforce a required scope).
"""
"""A verified non-interactive (service-to-service) caller — the token analog of
:class:`Session`: one bearer token on one request, no login/cookie/refresh. ``principal`` is
an opaque stable caller id; ``scopes`` empty means "unscoped" (a route MAY enforce one)."""
principal: str
provider: str
scopes: tuple[str, ...] = ()
@@ -40,14 +32,9 @@ class TokenPrincipal:
@dataclass(frozen=True)
class LoginStart:
"""First leg of the OAuth round trip.
``redirect_url`` is the IDP's authorize endpoint; ``cookie_payload`` maps
cookie name -> serialised PKCE/CSRF state that the auth route sets
(HttpOnly, Secure over HTTPS, TTL <= 10 min, ``SameSite=None`` over HTTPS
so it survives the cross-site redirect chain — see ``cookies.set_pkce_cookie``).
"""
"""First leg of the OAuth round trip: ``redirect_url`` is the IDP's authorize endpoint;
``cookie_payload`` maps cookie name -> serialised PKCE/CSRF state that the auth route sets
(HttpOnly, Secure + ``SameSite=None`` over HTTPS, TTL <= 10 min; see ``set_pkce_cookie``)."""
redirect_url: str
cookie_payload: dict[str, str]
@@ -61,33 +48,28 @@ class InvalidCodeError(Exception):
class InvalidCredentialsError(Exception):
"""Username/password rejected. The route answers a generic 401 (never
distinguishing unknown user from wrong password — no username oracle)."""
"""Username/password rejected. The route answers a generic 401 (no username oracle)."""
class RefreshExpiredError(Exception):
"""This provider rejects the refresh token. Not proof of ownership in a
multi-provider deployment, so middleware tries the remaining providers and
forces re-login only after every reachable one rejects it."""
"""This provider rejects the refresh token. Not proof of ownership in a multi-provider
deployment: middleware tries the rest and forces re-login only after every reachable one
rejects it."""
def classify_jwks_lookup_error(exc: BaseException) -> Exception:
"""Map a ``PyJWKClient.get_signing_key_from_jwt`` failure to the protocol.
Only a genuine transport failure (``PyJWKClientConnectionError``, or an
unexpected JWKS shape) is a :class:`ProviderError` (503, never forces logout).
A non-JWT bearer (``DecodeError``), a JWKS with no key for this ``kid``
(``PyJWKSetError``) or any other invalid token is simply not verifiable by this
provider -> :class:`InvalidCodeError` (``verify_session`` returns ``None``).
Folding "cannot parse" into "cannot reach" once made every opaque bearer a
fast 503 against a healthy Portal.
"""
"""Map a ``PyJWKClient.get_signing_key_from_jwt`` failure to the protocol. Only a genuine
transport failure (``PyJWKClientConnectionError``, or an unexpected JWKS shape) is a
:class:`ProviderError` (503, never forces logout). A non-JWT bearer (``DecodeError``), a JWKS
with no key for this ``kid`` (``PyJWKSetError``) or any other invalid token is simply not
verifiable by this provider -> :class:`InvalidCodeError` (``verify_session`` returns ``None``).
Folding "cannot parse" into "cannot reach" once made every opaque bearer a fast 503."""
try:
import jwt
except Exception: # pragma: no cover - jwt is a hard dep of these providers
return ProviderError(f"JWKS lookup failed: {exc!r}")
# Order matters: DecodeError/PyJWKSetError are checked before their
# PyJWKClientError / InvalidTokenError parents.
# Order matters: DecodeError/PyJWKSetError before their PyJWKClientError/InvalidTokenError
# parents.
if isinstance(exc, jwt.PyJWKClientConnectionError):
return ProviderError(f"JWKS lookup failed: {exc}")
if isinstance(exc, (jwt.DecodeError, jwt.PyJWKSetError)):
@@ -102,29 +84,21 @@ def classify_jwks_lookup_error(exc: BaseException) -> Exception:
class DashboardAuthProvider(ABC):
"""Protocol every dashboard-auth provider plugin implements.
Lifecycle: ``start_login`` (redirect URL + PKCE state) -> IDP ->
``complete_login`` (code + verifier -> Session) -> ``verify_session`` per
request -> ``refresh_session`` near expiry -> ``revoke_session`` on logout
(best-effort, must not raise).
Lifecycle: ``start_login`` (redirect URL + PKCE state) -> IDP -> ``complete_login`` (code +
verifier -> Session) -> ``verify_session`` per request -> ``refresh_session`` near expiry ->
``revoke_session`` on logout (best-effort, must not raise). Failure semantics: ``start_login``
/ ``complete_login`` raise ``ProviderError`` when the IDP is unreachable, ``complete_login``
``InvalidCodeError`` on a bad code/state; ``verify_session`` returns ``None`` for
expired/unknown tokens (middleware refreshes) and raises ``ProviderError`` when unreachable
(503); ``refresh_session`` raises ``RefreshExpiredError`` when the token is invalid for that
provider (a foreign opaque token looks expired, so middleware tries the rest) and
``ProviderError`` on network failure (503, cookies kept).
Failure semantics:
* ``start_login`` / ``complete_login`` raise ``ProviderError`` when the IDP
is unreachable; ``complete_login`` raises ``InvalidCodeError`` on a bad
code/state.
* ``verify_session`` returns ``None`` for expired/unknown tokens (middleware
refreshes) and raises ``ProviderError`` when unreachable (503).
* ``refresh_session`` raises ``RefreshExpiredError`` when the token is invalid
for that provider (an opaque foreign token looks expired, so middleware
tries the rest) and ``ProviderError`` on network failure (503, cookies kept).
Subclasses MUST set ``name`` (stable lowercase id) and ``display_name``.
Capability flags: ``supports_password`` (credential form +
``complete_password_login``; OAuth methods may be ``NotImplementedError``
stubs), ``supports_token`` (``verify_token`` for the token-auth seam),
``supports_session`` (False for token-only credentials such as drain, never
offered a login). Everything downstream of login is identical for every session.
Subclasses MUST set ``name`` (stable lowercase id) and ``display_name``. Capability flags:
``supports_password`` (credential form + ``complete_password_login``; OAuth methods may be
``NotImplementedError`` stubs), ``supports_token`` (``verify_token`` for the token-auth seam),
``supports_session`` (False for token-only credentials such as drain, never offered a login).
"""
name: str = ""
display_name: str = ""
supports_password: bool = False
@@ -136,8 +110,7 @@ class DashboardAuthProvider(ABC):
@abstractmethod
def complete_login(
self, *, code: str, state: str, code_verifier: str, redirect_uri: str,
) -> Session: ...
self, *, code: str, state: str, code_verifier: str, redirect_uri: str) -> Session: ...
@abstractmethod
def verify_session(self, *, access_token: str) -> Optional[Session]: ...
@@ -149,35 +122,27 @@ class DashboardAuthProvider(ABC):
def revoke_session(self, *, refresh_token: str) -> None: ...
def complete_password_login(self, *, username: str, password: str) -> "Session":
"""Verify a username/password pair and mint a :class:`Session`.
Only called when ``supports_password`` is True. Raise
``InvalidCredentialsError`` on rejection (SHOULD be constant time for
unknown users — no timing oracle) and ``ProviderError`` when the store is
unreachable. The default raises so a mis-flagged provider fails loudly.
"""
"""Verify a username/password pair and mint a :class:`Session` (only called when
``supports_password``). Raise ``InvalidCredentialsError`` on rejection (SHOULD be constant
time for unknown users — no timing oracle) and ``ProviderError`` when the store is
unreachable. The default raises so a mis-flagged provider fails loudly."""
raise NotImplementedError(
f"{type(self).__name__} does not support password login "
"(set supports_password = True and override "
"complete_password_login)")
"(set supports_password = True and override complete_password_login)")
def verify_token(self, *, token: str) -> "Optional[TokenPrincipal]":
"""Verify a non-interactive bearer token; return its principal.
Mirrors ``verify_session``: return ``None`` (never raise) for an
unrecognised token so the seam falls through; raise ``ProviderError`` ONLY
for a genuine backing-store outage. Shared secrets MUST be compared with
``hmac.compare_digest``. The default raises so a mis-flagged provider fails
loudly.
"""
"""Verify a non-interactive bearer token; return its principal. Mirrors ``verify_session``:
return ``None`` (never raise) for an unrecognised token so the seam falls through; raise
``ProviderError`` ONLY for a genuine backing-store outage. Shared secrets MUST be compared
with ``hmac.compare_digest``. The default raises so a mis-flagged provider fails loudly."""
raise NotImplementedError(
f"{type(self).__name__} does not support token auth "
"(set supports_token = True and override verify_token)")
def assert_protocol_compliance(cls: type) -> None:
"""Raise ``TypeError`` if ``cls`` doesn't fully implement the provider protocol
(call it from every provider plugin's unit tests)."""
"""Raise ``TypeError`` if ``cls`` doesn't fully implement the protocol (call it from every
provider plugin's unit tests)."""
for attr in ("name", "display_name"):
if not getattr(cls, attr, ""):
raise TypeError(f"{cls.__name__} missing or empty attribute: {attr!r}")
hermes_cli/dashboard_auth/cookies.py
+41 -77
View File
@@ -1,24 +1,16 @@
"""Cookie helpers for dashboard auth.
All HttpOnly, ``SameSite=Lax`` unless noted, Path = proxy prefix or /:
- hermes_session_at access token; Max-Age = token TTL (~15 min)
- hermes_session_rt rotating refresh token; written only when the provider
returned one, always cleared on logout/expiry
- hermes_session_provider non-secret routing hint so an RT is not handed to the
wrong provider when several auth plugins are enabled
- hermes_session_pkce PKCE state + CSRF nonce + provider hint, 10 min.
``SameSite=None; Secure`` over HTTPS: it is set on the
/auth/login 302 and must survive the cross-site redirect
chain (Chromium drops Lax cookies set on such a 302,
crbug 40508226); plain HTTP falls back to Lax.
- hermes_sso_attempt one-shot auto-SSO loop-guard marker (60 s)
``Secure`` only when ``request.url.scheme`` is https (honours ``X-Forwarded-Proto``
only from uvicorn's ``forwarded_allow_ips``). Cookie-prefix hardening per
draft-west-cookie-prefixes: bare name over HTTP; ``__Host-`` on gated HTTPS with
Path=/; ``__Secure-`` behind a proxy prefix (``__Host-`` forbids Path != /). Setters
and readers BOTH resolve the name via :func:`_resolved_name` — a mismatch silently
breaks sessions.
All HttpOnly, ``SameSite=Lax`` unless noted, Path = proxy prefix or /: ``hermes_session_at``
(access token; Max-Age = token TTL), ``hermes_session_rt`` (rotating refresh token; written only
when the provider returned one, always cleared on logout/expiry), ``hermes_session_provider``
(non-secret routing hint so an RT is not handed to the wrong provider), ``hermes_session_pkce``
(PKCE state + CSRF nonce + provider hint, 10 min; ``SameSite=None; Secure`` over HTTPS because it
is set on the /auth/login 302 and must survive the cross-site redirect chain — Chromium drops Lax
cookies set on such a 302, crbug 40508226), ``hermes_sso_attempt`` (auto-SSO loop guard, 60 s).
``Secure`` only when ``request.url.scheme`` is https. Cookie-prefix hardening per
draft-west-cookie-prefixes: bare name over HTTP; ``__Host-`` on gated HTTPS with Path=/;
``__Secure-`` behind a proxy prefix (``__Host-`` forbids Path != /). Setters and readers BOTH
resolve the name via :func:`_resolved_name` — a mismatch silently breaks sessions.
"""
from __future__ import annotations
@@ -41,16 +33,14 @@ SSO_ATTEMPT_COOKIE = "hermes_sso_attempt"
# Name variants a reader may have to try; most strict first.
_NAME_VARIANTS = ("__Host-", "__Secure-", "")
# RT cookie lifetime is a generous browser-side upper bound; the provider's own
# RT TTL is the real authority (an expired RT -> RefreshExpiredError -> re-login).
# RT cookie lifetime is a generous browser-side upper bound; the provider's own RT TTL is the
# real authority (an expired RT -> RefreshExpiredError -> re-login).
_RT_MAX_AGE = 30 * 24 * 60 * 60
_PKCE_MAX_AGE = 10 * 60
# Long enough for one portal round trip / back-button; short enough that a
# user returning later gets a fresh silent attempt rather than a stuck /login.
# Long enough for one portal round trip / back-button; short enough that a user returning later
# gets a fresh silent attempt rather than a stuck /login.
_SSO_ATTEMPT_MAX_AGE = 60
# Cheap pre-filter: legacy wire forms always contain ``%`` or ``;``, which are
# outside the base64url alphabet, so they never reach the base64 decoder.
# Cheap pre-filter: legacy wire forms always contain ``%`` or ``;`` (outside base64url).
_B64URL_RE = re.compile(r"^[A-Za-z0-9_-]+={0,2}$")
@@ -62,8 +52,7 @@ def _resolved_name(bare: str, *, use_https: bool, prefix: str) -> str:
def _cookie_path(prefix: str) -> str:
"""``Path=/hermes`` under a proxy prefix (so the cookie does not leak to
sibling apps), else ``/``."""
"""``Path=/hermes`` under a proxy prefix (no leak to sibling apps), else ``/``."""
return prefix if prefix else "/"
@@ -101,13 +90,9 @@ def set_session_provider_cookie(
def set_session_cookies(
response: Response, *, access_token: str, refresh_token: str, access_token_expires_in: int,
use_https: bool, prefix: str = "", provider: str = "") -> None:
"""Set the session cookies.
``access_token_expires_in`` is seconds (the provider's reported TTL). An
empty ``refresh_token`` means "don't persist the RT cookie" — a literal
empty cookie would be dead state at best, attack surface at worst.
``prefix`` is the normalised X-Forwarded-Prefix or ``""``.
"""
"""``access_token_expires_in`` is seconds (the provider's reported TTL). An empty
``refresh_token`` means "don't persist the RT cookie" — a literal empty cookie would be dead
state at best, attack surface at worst."""
_set(response, SESSION_AT_COOKIE, access_token, max_age=access_token_expires_in,
use_https=use_https, prefix=prefix)
if refresh_token:
@@ -119,14 +104,10 @@ def set_session_cookies(
def _clear_cookie_variants(
response: Response, bare_name: str, *, prefix: str,
https_samesite: Literal["lax", "strict", "none"], bare_attrs: dict) -> None:
"""Emit Max-Age=0 deletions for every plausible name variant of a cookie.
We don't know which variant the setter used (it depends on the setting
request's shape). Prefixed names are rejected by the browser unless they
carry ``Secure`` (``__Host-`` additionally needs ``Path=/``), so those
deletions always carry them; the bare deletion mirrors the setter's shape
(``bare_attrs``), which works on both HTTP and HTTPS origins.
"""
"""Emit Max-Age=0 deletions for every plausible name variant (the setting request's shape is
unknown). Prefixed names are rejected by the browser unless they carry ``Secure`` (``__Host-``
additionally ``Path=/``), so those deletions always do; the bare deletion mirrors the setter's
shape (``bare_attrs``), which works on both HTTP and HTTPS origins."""
for variant, path in (("__Host-", "/"), ("__Secure-", _cookie_path(prefix))):
response.set_cookie(
f"{variant}{bare_name}", "", max_age=0, path=path, httponly=True,
@@ -143,41 +124,32 @@ def clear_session_cookies(response: Response, *, prefix: str = "") -> None:
def encode_pkce_payload(parts: dict[str, str]) -> str:
"""Serialise PKCE segments to the wire value ``base64url(JSON)``, no padding.
The urlsafe alphabet is a strict subset of RFC 6265 cookie-octets, so
http.cookies never quotes it (strict proxies such as Go net/http reject the
quoted form) and no value can collide with a delimiter. Padding ``=`` would
trigger quoting; the parser restores it.
"""
"""Wire value ``base64url(JSON)``, no padding. The urlsafe alphabet is a strict subset of RFC
6265 cookie-octets, so http.cookies never quotes it (strict proxies such as Go net/http reject
the quoted form) and no value can collide with a delimiter; padding ``=`` would trigger
quoting, the parser restores it."""
raw = json.dumps(parts, separators=(",", ":"), sort_keys=True)
return base64.urlsafe_b64encode(raw.encode("utf-8")).decode("ascii").rstrip("=")
def set_pkce_cookie(
response: Response, *, payload: dict[str, str], use_https: bool, prefix: str = "") -> None:
"""Set the PKCE cookie (``payload`` is the segment dict; see module docstring
for the SameSite=None rationale)."""
"""``payload`` is the segment dict; see module docstring for the SameSite=None rationale."""
_set(response, PKCE_COOKIE, encode_pkce_payload(payload), max_age=_PKCE_MAX_AGE,
use_https=use_https, prefix=prefix, attrs=_pkce_attrs(use_https=use_https, prefix=prefix))
def clear_pkce_cookie(response: Response, *, use_https: bool, prefix: str = "") -> None:
"""Delete every PKCE cookie variant; the bare deletion mirrors the setter's
shape for the active origin, the prefixed ones carry ``Secure; SameSite=None``."""
"""Delete every PKCE cookie variant (prefixed ones carry ``Secure; SameSite=None``)."""
_clear_cookie_variants(
response, PKCE_COOKIE, prefix=prefix, https_samesite="none",
bare_attrs=_pkce_attrs(use_https=use_https, prefix=prefix))
def _read_with_fallback(request: Request, bare_name: str) -> Optional[str]:
"""Read a cookie trying every prefix variant (the reading request may not
have the same shape as the one that set it)."""
for variant in _NAME_VARIANTS:
value = request.cookies.get(f"{variant}{bare_name}")
if value is not None:
return value
return None
"""Try every prefix variant (the reading request may not match the setting request's shape)."""
return next((v for v in (request.cookies.get(f"{p}{bare_name}") for p in _NAME_VARIANTS)
if v is not None), None)
def read_session_cookies(request: Request) -> Tuple[Optional[str], Optional[str]]:
@@ -197,18 +169,12 @@ def read_pkce_cookie(request: Request) -> Optional[str]:
def parse_pkce_payload(raw: str) -> dict[str, str]:
"""Decode a PKCE cookie value into its segment dict (inverse of
:func:`encode_pkce_payload`). EVERY reader must go through this — reading the
raw wire value parses zero segments and silently disables the check it feeds
(provider dispatch, CSRF state, broker binding).
Compatibility ladder for cookies minted by an older server mid-upgrade:
1. base64url(JSON) — current.
2. Flat form with raw ``;`` delimiters — split WITHOUT unquoting (the ``next``
segment carries its own URL-encoding; a ``%3B`` inside it is not a delimiter).
3. URL-encoded flat form — unquote once, split.
A NEW cookie hitting an OLD server fails the state check; the user retries.
"""
"""Inverse of :func:`encode_pkce_payload`. EVERY reader must go through this — reading the raw
wire value parses zero segments and silently disables the check it feeds (provider dispatch,
CSRF state, broker binding). Compatibility ladder for cookies minted by an older server
mid-upgrade: 1. base64url(JSON); 2. flat form with raw ``;`` delimiters, split WITHOUT
unquoting (the ``next`` segment carries its own URL-encoding); 3. URL-encoded flat form,
unquote once then split. A NEW cookie hitting an OLD server fails the state check."""
if _B64URL_RE.match(raw):
try:
padded = raw + "=" * (-len(raw) % 4)
@@ -233,14 +199,12 @@ def read_sso_attempt_cookie(request: Request) -> Optional[str]:
def clear_sso_attempt_cookie(response: Response, *, prefix: str = "") -> None:
"""Delete the auto-SSO marker (every variant) so it never suppresses a
later silent attempt."""
"""Delete the auto-SSO marker (every variant) so it never suppresses a later silent attempt."""
_clear_cookie_variants(
response, SSO_ATTEMPT_COOKIE, prefix=prefix, https_samesite="lax",
bare_attrs=_common_attrs(use_https=False, prefix=prefix))
def detect_https(request: Request) -> bool:
"""``Secure`` flag decision: ``request.url.scheme == "https"`` (honours
``X-Forwarded-Proto`` under uvicorn ``proxy_headers=True``)."""
"""``Secure`` flag decision (honours ``X-Forwarded-Proto`` under uvicorn ``proxy_headers``)."""
return request.url.scheme == "https"
hermes_cli/dashboard_auth/middleware.py
+72 -96
View File
@@ -26,48 +26,33 @@ from hermes_cli.dashboard_auth.cookies import (
from hermes_cli.dashboard_auth.prefix import prefix_from_request
from hermes_cli.dashboard_auth.public_paths import PUBLIC_API_PATHS
from hermes_cli.dashboard_auth.request_utils import (
access_token_max_age as _expires_in_seconds,
client_ip as _client_ip,
extract_bearer as _extract_bearer,
is_safe_next_path,
scan_session_providers,
access_token_max_age as _expires_in_seconds, client_ip as _client_ip,
extract_bearer as _extract_bearer, is_safe_next_path, scan_session_providers,
unreachable_response)
_log = logging.getLogger(__name__)
# Prefix-matched (``path == p or path.startswith(p)``) bypass list: auth
# bootstrap routes and static asset mounts. ``/assets/`` with the trailing
# slash matches ``/assets/foo.css`` but not ``/assetsleak``.
# Prefix-matched bypass list: auth bootstrap routes and static asset mounts. ``/assets/`` with
# the trailing slash matches ``/assets/foo.css`` but not ``/assetsleak``.
_GATE_PUBLIC_PREFIXES: tuple[str, ...] = (
"/auth/login",
"/auth/callback",
"/auth/native/authorize",
"/auth/native/token",
"/auth/native/refresh",
"/auth/password-login",
"/auth/logout",
"/login",
"/api/auth/providers",
"/api/mcp/oauth/callback/",
"/assets/",
"/favicon.ico",
"/ds-assets/",
"/fonts/",
"/fonts-terminal/")
"/auth/login", "/auth/callback", "/auth/native/authorize", "/auth/native/token",
"/auth/native/refresh", "/auth/password-login", "/auth/logout", "/login",
"/api/auth/providers", "/api/mcp/oauth/callback/",
"/assets/", "/favicon.ico", "/ds-assets/", "/fonts/", "/fonts-terminal/")
def _path_is_public(path: str) -> bool:
"""True if ``path`` bypasses the gate: :data:`PUBLIC_API_PATHS` (shared with the
legacy middleware) matched exactly so ``/api/status`` never exposes
``/api/status/extension``; :data:`_GATE_PUBLIC_PREFIXES` prefix-matched."""
""":data:`PUBLIC_API_PATHS` (shared with the legacy middleware) matched exactly so
``/api/status`` never exposes ``/api/status/extension``; :data:`_GATE_PUBLIC_PREFIXES`
prefix-matched."""
return path in PUBLIC_API_PATHS or any(
path == p or path.startswith(p) for p in _GATE_PUBLIC_PREFIXES)
def _safe_next_target(request: Request) -> str:
"""URL-encoded ``next`` value for the login redirect, or ``""``. Only same-origin
paths outside the auth flow and ``/api`` are kept (query preserved); dropped
deep links fall back to the SPA's ``sessionStorage["hermes.lastLocation"]``."""
"""URL-encoded ``next`` value for the login redirect, or ``""``. Only same-origin paths outside
the auth flow and ``/api`` are kept (query preserved); dropped deep links fall back to the
SPA's ``sessionStorage["hermes.lastLocation"]``."""
path = request.url.path
if not path or not is_safe_next_path(path):
return ""
@@ -76,32 +61,27 @@ def _safe_next_target(request: Request) -> str:
def _unauth_response(request: Request, *, reason: str) -> Response:
"""API routes -> 401 JSON with ``login_url``; HTML routes -> 302 -> /login.
fetch() follows a 302 opaquely into the cross-origin OAuth dance, so API routes
never get redirects; the SPA's 401 handler navigates to ``login_url`` on
``unauthenticated`` / ``session_expired``. Both carry ``next=`` and the prefix."""
"""API routes -> 401 JSON with ``login_url``; HTML routes -> 302 -> /login. fetch() follows a
302 opaquely into the cross-origin OAuth dance, so API routes never get redirects; the SPA's
401 handler navigates to ``login_url`` on ``unauthenticated`` / ``session_expired``."""
next_param = _safe_next_target(request)
prefix = prefix_from_request(request)
login_url = f"{prefix}/login?next={next_param}" if next_param else f"{prefix}/login"
if request.url.path.startswith("/api/"):
error_code = (
"session_expired" if reason == "invalid_or_expired_session" else "unauthenticated")
expired = reason == "invalid_or_expired_session"
return JSONResponse(
{"error": error_code, "detail": "Unauthorized", "reason": reason,
"login_url": login_url},
status_code=401)
{"error": "session_expired" if expired else "unauthenticated", "detail": "Unauthorized",
"reason": reason, "login_url": login_url}, status_code=401)
return RedirectResponse(url=login_url, status_code=302)
def _auto_sso_response(request: Request) -> Response | None:
"""302 straight to ``/auth/login`` on an unauthenticated HTML load, or ``None``.
Only for a document load (not ``/api/*``) when exactly one interactive
provider is registered and it is OAuth-style (a password provider must render
the form), and the one-shot loop-guard cookie is absent — a present marker
means the portal had no session last time: clear it and fall back to ``/login``
rather than ping-pong. Convenience, not a security check: ``/auth/login`` runs
the unchanged PKCE flow.
Only for a document load (not ``/api/*``) when exactly one interactive OAuth-style provider is
registered (a password provider must render the form) and the one-shot loop-guard cookie is
absent — a present marker means the portal had no session last time: clear it and fall back
to ``/login`` rather than ping-pong. Convenience, not a security check.
"""
if request.url.path.startswith("/api/"):
return None
@@ -127,37 +107,57 @@ def _auto_sso_response(request: Request) -> Response | None:
def _verify_access_token(
request: Request, *, access_token: str, provider_hint: str | None = None, audit: bool = True):
"""Run ``verify_session`` across the provider stack; Session or ``None``.
``audit=False`` is the native-app bearer path (no cookie, no server-side refresh
— the desktop rotates via ``/auth/native/refresh``); 503-on-outage semantics
come from :func:`scan_session_providers`."""
"""Run ``verify_session`` across the provider stack; Session or ``None``. ``audit=False`` is
the native-app bearer path (no cookie, no server-side refresh — the desktop rotates via
``/auth/native/refresh``); 503-on-outage semantics come from :func:`scan_session_providers`."""
def _audit_unreachable(provider):
if audit:
audit_log(
AuditEvent.SESSION_VERIFY_FAILURE, provider=provider.name,
reason="provider_unreachable", ip=_client_ip(request))
audit_log(AuditEvent.SESSION_VERIFY_FAILURE, provider=provider.name,
reason="provider_unreachable", ip=_client_ip(request))
return scan_session_providers(
provider_hint, lambda p: p.verify_session(access_token=access_token),
phase="verify" if audit else "bearer verify", log=_log, on_unreachable=_audit_unreachable)
async def _serve_refreshed(request: Request, call_next, new_session, provider: str) -> Response:
"""Serve the request under a just-rotated session and write the rotated cookies back. Writing
the ROTATED RT is mandatory: Portal runs reuse detection, so replaying the stale RT would
revoke the session."""
request.state.session = new_session
response = await call_next(request)
set_session_cookies(
response, access_token=new_session.access_token, refresh_token=new_session.refresh_token,
access_token_expires_in=_expires_in_seconds(new_session), use_https=detect_https(request),
prefix=prefix_from_request(request), provider=provider)
audit_log(AuditEvent.REFRESH_SUCCESS, provider=provider, user_id=new_session.user_id,
ip=_client_ip(request))
return response
def _session_expired_response(request: Request) -> Response:
"""Refresh failed (or no RT): structured 401/redirect and clear the dead cookies under the
active prefix so the deletion Path matches the set Path."""
audit_log(AuditEvent.SESSION_VERIFY_FAILURE, reason="no_provider_recognises",
ip=_client_ip(request))
response = _unauth_response(request, reason="invalid_or_expired_session")
clear_session_cookies(response, prefix=prefix_from_request(request))
return response
async def gated_auth_middleware(
request: Request, call_next: Callable[[Request], Awaitable[Response]]) -> Response:
"""Engaged only when ``app.state.auth_required is True``."""
if not getattr(request.app.state, "auth_required", False):
return await call_next(request)
# Already authenticated by the token-auth seam (service caller on a
# registered token route): not a cookie session, must not bounce to /login.
if getattr(request.state, "token_authenticated", False):
# Already authenticated by the token-auth seam (service caller on a registered token
# route): not a cookie session, must not bounce to /login.
if getattr(request.state, "token_authenticated", False) or _path_is_public(request.url.path):
return await call_next(request)
if _path_is_public(request.url.path):
return await call_next(request)
# RFC 8252 native-app bearer path: the same provider-minted access token
# the cookie flow stores, verified with the same provider stack, no cookie
# read or set. A presented-but-invalid bearer gets the structured 401 so
# the desktop refreshes/re-logs instead of following a cookie redirect.
# RFC 8252 native-app bearer path: the same provider-minted access token the cookie flow
# stores, verified with the same provider stack, no cookie read or set. A presented-but-
# invalid bearer gets the structured 401 so the desktop refreshes/re-logs instead of
# following a cookie redirect.
bearer = _extract_bearer(request)
if bearer:
try:
@@ -175,49 +175,25 @@ async def gated_auth_middleware(
# No session at all: try the silent portal bounce before /login.
auto = _auto_sso_response(request)
return auto if auto is not None else _unauth_response(request, reason="no_cookie")
# An absent AT with a present RT is the COMMON expiry case (the AT cookie's
# Max-Age tracks the token TTL, so the browser evicts it first) — skip
# straight to refresh rather than bouncing to /login.
# An absent AT with a present RT is the COMMON expiry case (the AT cookie's Max-Age tracks
# the token TTL, so the browser evicts it first) — skip straight to refresh.
session = None
if at:
try:
session = _verify_access_token(request, access_token=at, provider_hint=provider_hint)
except ProviderError as e:
return unreachable_response(str(e))
if session is None:
# Rotate via the refresh token before forcing re-login. On success the
# rotated cookies are re-set and the request served transparently.
# Rotate via the refresh token before forcing re-login; on success the request is
# served transparently with the rotated cookies re-set.
try:
refreshed = _attempt_refresh(request, refresh_token=_rt, provider_hint=provider_hint)
except ProviderError as e:
# Uncertain (provider unreachable), not rejected: keep the cookies.
return unreachable_response(str(e))
if refreshed is not None:
new_session, refreshing_provider = refreshed
request.state.session = new_session
response = await call_next(request)
# Writing the ROTATED RT back is mandatory: Portal runs reuse
# detection, so replaying the stale RT would revoke the session.
set_session_cookies(
response, access_token=new_session.access_token,
refresh_token=new_session.refresh_token,
access_token_expires_in=_expires_in_seconds(new_session),
use_https=detect_https(request), prefix=prefix_from_request(request),
provider=refreshing_provider)
audit_log(
AuditEvent.REFRESH_SUCCESS, provider=refreshing_provider,
user_id=new_session.user_id, ip=_client_ip(request))
return response
audit_log(AuditEvent.SESSION_VERIFY_FAILURE, reason="no_provider_recognises",
ip=_client_ip(request))
response = _unauth_response(request, reason="invalid_or_expired_session")
# Refresh failed (or no RT): clear the dead cookies under the active
# prefix so the deletion Path matches the set Path.
clear_session_cookies(response, prefix=prefix_from_request(request))
return response
if refreshed is None:
return _session_expired_response(request)
return await _serve_refreshed(request, call_next, *refreshed)
request.state.session = session
response = await call_next(request)
@@ -229,10 +205,10 @@ async def gated_auth_middleware(
def _attempt_refresh(request: Request, *, refresh_token, provider_hint: str | None = None):
"""Rotate an expired session via the refresh token; ``(Session, provider_name)``
or ``None``. ``RefreshExpiredError`` rejects that candidate only (Basic raises it
for foreign opaque tokens too); if none succeeds and any raised ``ProviderError``
it is re-raised so the caller returns 503 without clearing cookies."""
"""Rotate an expired session via the refresh token; ``(Session, provider_name)`` or ``None``.
``RefreshExpiredError`` rejects that candidate only (Basic raises it for foreign opaque tokens
too); if none succeeds and any raised ``ProviderError`` it is re-raised so the caller returns
503 without clearing cookies."""
if not refresh_token:
return None
hermes_cli/dashboard_auth/native_flow.py
+23 -28
View File
@@ -25,14 +25,11 @@ from typing import Dict, Optional
from hermes_cli.dashboard_auth.base import Session
# Pending authorization: the whole interactive login (mirrors the PKCE cookie).
_PENDING_TTL_SECONDS = 600
# Minted code: only the loopback redirect + immediate token POST.
_CODE_TTL_SECONDS = 120
# Global cap so a misbehaving client cannot grow the store unbounded.
_MAX_ENTRIES = 256
# Per-IP cap on PENDING entries: /auth/native/authorize is a public pre-auth
# route, so one spammer must not fill the global store and lock out logins.
_PENDING_TTL_SECONDS = 600 # whole interactive login (mirrors the PKCE cookie)
_CODE_TTL_SECONDS = 120 # loopback redirect + immediate token POST only
_MAX_ENTRIES = 256 # global cap so a misbehaving client cannot grow the store unbounded
# Per-IP cap on PENDING entries: /auth/native/authorize is a public pre-auth route, so one
# spammer must not fill the global store and lock out logins.
_MAX_PENDING_PER_IP = 8
_lock = threading.Lock()
@@ -41,7 +38,6 @@ _lock = threading.Lock()
@dataclass
class _Pending:
"""In-flight native authorization awaiting the upstream callback."""
code_challenge: str # the DESKTOP's S256 challenge (cc_d), base64url no-pad
redirect_uri: str # the desktop's loopback redirect (127.0.0.1:<port>/...)
client_state: str # the desktop's own ``state`` (echoed back on redirect)
@@ -52,7 +48,6 @@ class _Pending:
@dataclass
class _IssuedCode:
"""A minted one-time gateway authorization code bound to a Session."""
code_challenge: str # cc_d — verified against cv_d at redemption
session: Session
expires_at: int
@@ -95,12 +90,20 @@ def _now(now: Optional[int]) -> int:
return int(time.time()) if now is None else now
def _pop_pending_locked(broker_state: str, *, consume: bool) -> _Pending:
"""Look up (optionally consuming) a pending entry; :class:`PendingNotFound` if unknown."""
entry = (_pending.pop if consume else _pending.get)(broker_state, None)
if entry is None:
raise PendingNotFound("unknown or expired native authorization")
return entry
def register_pending(
*, code_challenge: str, redirect_uri: str, client_state: str, client_ip: str = "",
now: Optional[int] = None) -> str:
"""Stash a pending native authorization; return an opaque ``broker_state``.
``code_challenge`` is the DESKTOP's cc_d. Raises ``NativeFlowError`` (fail closed)
at store capacity or when ``client_ip`` holds ``_MAX_PENDING_PER_IP`` entries."""
"""Stash a pending native authorization; return an opaque ``broker_state``. ``code_challenge``
is the DESKTOP's cc_d. Raises ``NativeFlowError`` (fail closed) at store capacity or when
``client_ip`` holds ``_MAX_PENDING_PER_IP`` entries."""
now = _now(now)
broker_state = secrets.token_urlsafe(32)
with _lock:
@@ -117,26 +120,19 @@ def register_pending(
def get_pending(broker_state: str, *, now: Optional[int] = None) -> _Pending:
"""Peek (without consuming) the pending authorization; raises
:class:`PendingNotFound` if unknown or expired."""
"""Peek (without consuming) the pending authorization."""
with _lock:
_gc_locked(_now(now))
entry = _pending.get(broker_state)
if entry is None:
raise PendingNotFound("unknown or expired native authorization")
return entry
return _pop_pending_locked(broker_state, consume=False)
def complete_pending(broker_state: str, *, session: Session, now: Optional[int] = None) -> str:
"""Consume a pending authorization (single use) and mint a one-time gateway code
bound to the desktop's challenge + ``session``; :class:`PendingNotFound` if
unknown/expired."""
"""Consume a pending authorization (single use) and mint a one-time gateway code bound to the
desktop's challenge + ``session``."""
now = _now(now)
with _lock:
_gc_locked(now)
pending = _pending.pop(broker_state, None)
if pending is None:
raise PendingNotFound("unknown or expired native authorization")
pending = _pop_pending_locked(broker_state, consume=True)
if not _capacity_ok_locked():
raise NativeFlowError("native-flow code store at capacity")
gw_code = secrets.token_urlsafe(32)
@@ -147,9 +143,8 @@ def complete_pending(broker_state: str, *, session: Session, now: Optional[int]
def redeem_code(*, code: str, code_verifier: str, now: Optional[int] = None) -> Session:
"""Verify PKCE + consume a gateway code; return the bound :class:`Session`. The
entry is popped BEFORE the PKCE check so a wrong verifier cannot be retried
(no oracle, no replay). Raises :class:`CodeInvalid`."""
"""Verify PKCE + consume a gateway code; return the bound :class:`Session`. The entry is popped
BEFORE the PKCE check so a wrong verifier cannot be retried (no oracle, no replay)."""
now = _now(now)
with _lock:
_gc_locked(now)
hermes_cli/dashboard_auth/prefix.py
+31 -43
View File
@@ -16,57 +16,51 @@ from typing import Optional
_log = logging.getLogger(__name__)
# Home Assistant ingress prefixes are already 63 chars before a deployment adds
# its own sub-path; keep a bounded header budget with room for real mounts.
# Home Assistant ingress prefixes are already 63 chars before a deployment adds its own
# sub-path; keep a bounded header budget with room for real mounts.
_MAX_PREFIX_LENGTH = 256
# Presence of any of these in a public_url / prefix means a typo or a
# header-injection attempt: reject the whole value, never sanitise.
# Any of these in a public_url / prefix means a typo or a header-injection attempt: reject the
# whole value, never sanitise.
_REJECT_CHARS = frozenset(('"', "'", "<", ">", " ", "\n", "\r", "\t"))
# ``resolve_public_url`` runs on every authenticated request, so warnings are
# de-duplicated per distinct (source, value) — a changed value warns afresh.
# ``resolve_public_url`` runs on every authenticated request, so warnings are de-duplicated per
# distinct (source, value) — a changed value warns afresh.
_warned_malformed_public_urls: set = set()
_warned_malformed_prefixes: set = set()
def _warn_if_malformed(source: str, raw: str) -> None:
"""Warn once when a non-empty public-url value was rejected (almost always
a missing scheme; silently falling back to header reconstruction can yield
the wrong scheme behind a proxy)."""
cleaned = raw.strip() if raw else ""
if not cleaned or (source, cleaned) in _warned_malformed_public_urls:
def _warn_once(seen: set, key: tuple, cleaned: str, msg: str, *args) -> None:
if not cleaned or key in seen:
return
_warned_malformed_public_urls.add((source, cleaned))
_log.warning(
seen.add(key)
_log.warning(msg, *args)
def _warn_if_malformed(source: str, raw: str) -> None:
"""Warn once when a non-empty public-url value was rejected (almost always a missing scheme;
silently falling back to header reconstruction can yield the wrong scheme behind a proxy)."""
cleaned = raw.strip() if raw else ""
_warn_once(
_warned_malformed_public_urls, (source, cleaned), cleaned,
"%s is set to %r but was ignored because it is not a valid "
"absolute URL — it must include an http:// or https:// scheme "
"(e.g. https://%s). Falling back to reconstructing the OAuth "
"redirect URI from request headers, which may produce the wrong "
"scheme behind a reverse proxy.",
source,
cleaned,
cleaned.split("://")[-1] or "hermes.example.com")
source, cleaned, cleaned.split("://")[-1] or "hermes.example.com")
def _warn_if_malformed_prefix(raw: Optional[str], reason: str) -> None:
"""Warn once when a non-empty X-Forwarded-Prefix value is rejected."""
cleaned = raw.strip() if raw else ""
if not cleaned or (cleaned, reason) in _warned_malformed_prefixes:
return
_warned_malformed_prefixes.add((cleaned, reason))
_log.warning(
_warn_once(
_warned_malformed_prefixes, (cleaned, reason), cleaned,
"X-Forwarded-Prefix header %r was ignored because %s. "
"Dashboard URLs will be generated without a reverse-proxy path prefix.",
cleaned,
reason)
"Dashboard URLs will be generated without a reverse-proxy path prefix.", cleaned, reason)
def normalise_prefix(raw: Optional[str]) -> str:
"""Normalise an X-Forwarded-Prefix header to ``"/hermes"`` form (no trailing
slash) or ``""`` when unset/malformed. ``..``, ``//`` and injection
characters are rejected so a hostile proxy cannot smuggle HTML or traversal.
"""
"""``"/hermes"`` form (no trailing slash) or ``""`` when unset/malformed. ``..``, ``//`` and
injection characters are rejected so a hostile proxy cannot smuggle HTML or traversal."""
p = raw.strip() if raw else ""
if not p:
return ""
@@ -89,11 +83,9 @@ def prefix_from_request(request) -> str:
# --- HERMES_DASHBOARD_PUBLIC_URL / dashboard.public_url --------------------
def _normalise_public_url(raw: Optional[str]) -> str:
"""Cleaned ``scheme://netloc[/path]`` (trailing slash stripped) or ``""`` when
empty/malformed/injection-suspect; ``""`` means "fall back to request
reconstruction" (an explicit empty value equals an unset env var)."""
empty/malformed/injection-suspect (= fall back to request reconstruction)."""
url = raw.strip() if raw else ""
if not url or any(c in url for c in _REJECT_CHARS):
return ""
@@ -107,8 +99,7 @@ def _normalise_public_url(raw: Optional[str]) -> str:
def _load_dashboard_section() -> dict:
"""``dashboard`` block of config.yaml as a dict, or ``{}`` when the config
cannot be loaded or the block is absent/non-dict."""
"""``dashboard`` block of config.yaml as a dict, or ``{}`` when unloadable/absent/non-dict."""
try:
from hermes_cli.config import load_config
except Exception:
@@ -116,20 +107,17 @@ def _load_dashboard_section() -> dict:
try:
cfg = load_config()
except Exception as exc: # noqa: BLE001 — broad catch is intentional
_log.debug(
"dashboard-auth.prefix: load_config() raised %s; "
"falling back to env-only configuration",
exc)
_log.debug("dashboard-auth.prefix: load_config() raised %s; "
"falling back to env-only configuration", exc)
return {}
section = cfg.get("dashboard") if isinstance(cfg, dict) else None
return section if isinstance(section, dict) else {}
def resolve_public_url() -> str:
"""Operator-declared dashboard public URL, or ``""`` (reconstruct from request).
Precedence: ``HERMES_DASHBOARD_PUBLIC_URL`` env (blank counts as unset so a
provisioned-but-blank secret cannot shadow config.yaml), then
``dashboard.public_url``; a malformed value warns and falls through."""
"""Operator-declared dashboard public URL, or ``""`` (reconstruct from request). Precedence:
``HERMES_DASHBOARD_PUBLIC_URL`` env (blank counts as unset so a provisioned-but-blank secret
cannot shadow config.yaml), then ``dashboard.public_url``; malformed warns and falls through."""
env_raw = os.environ.get("HERMES_DASHBOARD_PUBLIC_URL", "")
env_clean = _normalise_public_url(env_raw)
if env_clean:
hermes_cli/dashboard_auth/public_paths.py
+7 -12
View File
@@ -1,13 +1,9 @@
"""Shared allowlist of ``/api/*`` paths that bypass dashboard auth.
Imported by BOTH gates — ``web_server.auth_middleware`` (loopback /
``--insecure``, ephemeral ``_SESSION_TOKEN``) and
``dashboard_auth.middleware.gated_auth_middleware`` (OAuth cookie) — so the
lists cannot drift again (a drift once 401'd ``/api/status`` under the OAuth
gate and broke the portal's cookie-less liveness probe). Keep minimal: every
entry must be safe for external uptime probes, the pre-login SPA, and anyone
who ``curl``s the hostname; otherwise gate it and bootstrap after login.
"""
"""Shared allowlist of ``/api/*`` paths that bypass dashboard auth. Imported by BOTH gates —
``web_server.auth_middleware`` (loopback / ``--insecure``) and
``dashboard_auth.middleware.gated_auth_middleware`` (OAuth cookie) — so the lists cannot drift
again (a drift once 401'd ``/api/status`` and broke the portal's cookie-less liveness probe).
Keep minimal: every entry must be safe for external uptime probes, the pre-login SPA, and anyone
who ``curl``s the hostname; otherwise gate it and bootstrap after login."""
from __future__ import annotations
PUBLIC_API_PATHS: frozenset[str] = frozenset({
@@ -28,5 +24,4 @@ PUBLIC_API_PATHS: frozenset[str] = frozenset({
# Chronos managed-cron fire webhook (NAS -> agent). NOT cookie-gated: it
# carries its own short-lived NAS-minted JWT (purpose=cron_fire), which the
# handler verifies — the JWT, not this allowlist, is the security boundary.
"/api/cron/fire",
})
"/api/cron/fire"})
hermes_cli/dashboard_auth/registry.py
+16 -31
View File
@@ -1,9 +1,6 @@
"""Module-level registry for DashboardAuthProvider instances.
Plugins call ``register_provider`` via the plugin context hook at startup.
The auth gate middleware iterates ``list_providers()`` and uses
``get_provider`` to dispatch on the session's ``provider`` field.
"""
"""Module-level registry for DashboardAuthProvider instances. Plugins call ``register_provider``
via the plugin context hook at startup; the auth gate iterates ``list_providers()`` and uses
``get_provider`` to dispatch on the session's ``provider`` field."""
from __future__ import annotations
import logging
@@ -37,12 +34,7 @@ def _log_registered(kind: str, provider: DashboardAuthProvider) -> None:
def register_provider(provider: DashboardAuthProvider, *, scope: Optional[str] = None) -> None:
"""Register a provider.
Raises:
TypeError: on protocol violation.
ValueError: if a provider with the same name is already registered.
"""
"""Raises ``TypeError`` on protocol violation, ``ValueError`` on a duplicate name."""
assert_protocol_compliance(type(provider))
with _lock:
target = _target(scope, create=True)
@@ -89,31 +81,24 @@ def list_providers(*, scope: Optional[str] = None) -> List[DashboardAuthProvider
def list_token_providers() -> List[DashboardAuthProvider]:
"""Providers with ``supports_token`` True, in registration order.
The ``token_auth`` seam consults only these, so OAuth/password-only
providers are never asked to ``verify_token``. Empty => a token-authable
route fails closed (401), never open.
"""
"""Providers with ``supports_token`` True, in registration order. The ``token_auth`` seam
consults only these, so OAuth/password-only providers are never asked to ``verify_token``;
empty => a token-authable route fails closed (401)."""
return [p for p in list_providers() if getattr(p, "supports_token", False)]
def list_session_providers() -> List[DashboardAuthProvider]:
"""Providers with ``supports_session`` True (interactive cookie sessions);
the login page, /auth/login and the gate's verify/refresh loops use only these."""
"""Providers with ``supports_session`` True (interactive cookie sessions); the login page,
/auth/login and the gate's verify/refresh loops use only these."""
return [p for p in list_providers() if getattr(p, "supports_session", True)]
def register_global_provider(provider: DashboardAuthProvider) -> None:
"""Register a host-owned provider in the process-global slot (upsert).
The registry is process-global and shared across every profile one
dashboard process serves, so these providers outlive any per-home plugin
manager. Always targets ``_providers`` (never a per-home overlay) and
*replaces* a same-name entry instead of raising, so a forced plugin
re-discovery (e.g. after a password change) rotates the provider in place.
Pairs with ``unregister_global_provider``.
"""
"""Register a host-owned provider in the process-global slot (upsert). The registry is shared
across every profile one dashboard process serves, so these outlive any per-home plugin
manager: always targets ``_providers`` (never a per-home overlay) and *replaces* a same-name
entry instead of raising, so a forced plugin re-discovery (e.g. after a password change)
rotates the provider in place. Pairs with ``unregister_global_provider``."""
assert_protocol_compliance(type(provider))
with _lock:
_providers[provider.name] = provider
@@ -121,8 +106,8 @@ def register_global_provider(provider: DashboardAuthProvider) -> None:
def unregister_global_provider(name: str, provider: DashboardAuthProvider) -> bool:
"""Remove a global registration if ``provider`` is still current (a stale
handle whose provider was already replaced never clears the live one)."""
"""Remove a global registration if ``provider`` is still current (a stale handle whose
provider was already replaced never clears the live one)."""
with _lock:
if _providers.get(name) is provider:
_providers.pop(name, None)
hermes_cli/dashboard_auth/request_utils.py
+12 -24
View File
@@ -11,18 +11,15 @@ from fastapi.responses import JSONResponse
from hermes_cli.dashboard_auth import list_session_providers
from hermes_cli.dashboard_auth.base import DashboardAuthProvider, ProviderError
# Paths a post-login redirect must never land on: the auth flow itself (would
# loop) and any ``/api/*`` target (renders raw JSON in the address bar and is
# indistinguishable from an attacker weaponising the redirect).
# Paths a post-login redirect must never land on: the auth flow itself (would loop) and any
# ``/api/*`` target (raw JSON in the address bar, indistinguishable from a weaponised redirect).
_NEXT_DENY_PREFIXES = ("/login", "/auth/", "/api/auth/")
def client_ip(request: Request) -> str:
"""First ``X-Forwarded-For`` hop, else the peer address."""
fwd = request.headers.get("x-forwarded-for", "")
if fwd:
return fwd.split(",")[0].strip()
return request.client.host if request.client else ""
return fwd.split(",")[0].strip() if fwd else (request.client.host if request.client else "")
def extract_bearer(request: Request) -> str:
@@ -34,11 +31,8 @@ def extract_bearer(request: Request) -> str:
def is_safe_next_path(path: str) -> bool:
"""True if ``path`` is a same-origin post-login target.
Rejects non-relative and protocol-relative (``//evil``) values, the auth
routes themselves, and every ``/api`` path.
"""
"""Same-origin post-login target: rejects non-relative and protocol-relative (``//evil``)
values, the auth routes themselves, and every ``/api`` path."""
if not path.startswith("/") or path.startswith("//"):
return False
if any(path == p or path.startswith(p) for p in _NEXT_DENY_PREFIXES):
@@ -57,23 +51,17 @@ def unreachable_response(provider_name: str) -> JSONResponse:
def scan_session_providers(
provider_hint: Optional[str],
call: Callable[[DashboardAuthProvider], object],
*,
phase: str,
log: logging.Logger,
swallow: tuple[type[BaseException], ...] = (),
provider_hint: Optional[str], call: Callable[[DashboardAuthProvider], object], *, phase: str,
log: logging.Logger, swallow: tuple[type[BaseException], ...] = (),
on_swallow: Optional[Callable[[DashboardAuthProvider], None]] = None,
on_unreachable: Optional[Callable[[DashboardAuthProvider], None]] = None):
"""Run ``call`` across the session providers; first non-``None`` result or ``None``.
The hinted provider goes first (stable sort; a stale/unknown hint leaves
registration order intact). ``swallow`` exceptions reject that candidate
only. A ``ProviderError`` (IDP/JWKS unreachable) must NOT abort the chain —
the credential may belong to a different, reachable provider; it is logged
under ``phase`` and, if nothing else succeeds, re-raised as
``ProviderError(name)`` so the caller answers 503 instead of forcing a
re-login through a possibly-unreachable path.
The hinted provider goes first (stable sort; a stale/unknown hint leaves registration order
intact). ``swallow`` exceptions reject that candidate only. A ``ProviderError`` (IDP/JWKS
unreachable) must NOT abort the chain — the credential may belong to a different, reachable
provider; it is logged under ``phase`` and, if nothing else succeeds, re-raised as
``ProviderError(name)`` so the caller answers 503 instead of forcing a re-login.
"""
providers = list_session_providers()
if provider_hint:
hermes_cli/dashboard_auth/routes.py
+106 -145
View File
@@ -59,14 +59,15 @@ def _prefix(request: Request) -> str:
return _prefix_mod.prefix_from_request(request)
def _redirect_uri(request: Request) -> str:
"""Absolute ``/auth/callback`` URL handed to the IDP.
def _audit(request: Request, event: AuditEvent, **fields) -> None:
audit_log(event, **fields, ip=_client_ip(request))
An operator-declared public URL is the complete authority (``X-Forwarded-Prefix``
ignored so a baked-in prefix is not doubled); otherwise ``url_for`` (honours
``X-Forwarded-Host/Proto`` under uvicorn ``proxy_headers``) with the prefix
prepended, which Starlette does not do.
"""
def _redirect_uri(request: Request) -> str:
"""Absolute ``/auth/callback`` URL handed to the IDP. An operator-declared public URL is the
complete authority (``X-Forwarded-Prefix`` ignored so a baked-in prefix is not doubled);
otherwise ``url_for`` (honours ``X-Forwarded-Host/Proto`` under uvicorn ``proxy_headers``)
with the prefix prepended, which Starlette does not do."""
public_url = _prefix_mod.resolve_public_url()
if public_url:
return f"{public_url}/auth/callback"
@@ -79,8 +80,8 @@ def _redirect_uri(request: Request) -> str:
def _provider_pkce_segments(cookie_payload: dict[str, str]) -> dict[str, str]:
"""Parse a provider's flat ``state=…;verifier=…`` PKCE string into a dict — the
ONE place the flat form is parsed; :func:`set_pkce_cookie` encodes the dict."""
"""Parse a provider's flat ``state=…;verifier=…`` PKCE string into a dict — the ONE place
the flat form is parsed; :func:`set_pkce_cookie` encodes the dict."""
flat = cookie_payload.get("hermes_session_pkce", "")
return dict(seg.split("=", 1) for seg in flat.split(";") if "=" in seg)
@@ -88,10 +89,8 @@ def _provider_pkce_segments(cookie_payload: dict[str, str]) -> dict[str, str]:
def _validate_post_login_target(raw: str) -> str:
"""``raw`` (URL-decoded) if it is a safe same-origin path, else ``""``. Re-validated
at every hop because a ``next=`` value can re-enter via a crafted URL."""
if not raw:
return ""
decoded = unquote(raw)
return decoded if is_safe_next_path(decoded) else ""
decoded = unquote(raw) if raw else ""
return decoded if decoded and is_safe_next_path(decoded) else ""
def _set_pkce(resp, request: Request, payload: dict[str, str]) -> None:
@@ -115,39 +114,46 @@ def _bearer_payload(session: Session) -> dict[str, Any]:
def _finish_native_login(
request: Request, *, broker_state: str, session: Session, provider: str) -> str:
"""Shared tail of ``/auth/callback`` and ``/auth/password-login``: mint the
one-time loopback code and return the desktop's ``redirect_uri?code=…&state=…``.
"""Mint the one-time loopback code and return the desktop's ``redirect_uri?code=…&state=…``.
No session cookies on the native path — the desktop redeems at ``/auth/native/token``."""
ip = _client_ip(request)
try:
pending = native_flow.get_pending(broker_state)
gw_code = native_flow.complete_pending(broker_state, session=session)
except native_flow.NativeFlowError:
audit_log(AuditEvent.NATIVE_TOKEN_FAILURE, provider=provider, reason="pending_not_found",
ip=ip)
_audit(request, AuditEvent.NATIVE_TOKEN_FAILURE, provider=provider,
reason="pending_not_found")
raise _http(400, _NATIVE_EXPIRED_DETAIL)
sep = "&" if "?" in pending.redirect_uri else "?"
query = urlencode({'code': gw_code, 'state': pending.client_state})
loopback = f"{pending.redirect_uri}{sep}{query}"
audit_log(AuditEvent.NATIVE_CODE_ISSUED, provider=provider, user_id=session.user_id, ip=ip)
return loopback
_audit(request, AuditEvent.NATIVE_CODE_ISSUED, provider=provider, user_id=session.user_id)
return f"{pending.redirect_uri}{sep}{query}"
def _login_failure(request: Request, provider: str, reason: str, **extra) -> None:
audit_log(AuditEvent.LOGIN_FAILURE, provider=provider, reason=reason, **extra,
ip=_client_ip(request))
_audit(request, AuditEvent.LOGIN_FAILURE, provider=provider, reason=reason, **extra)
def _login_success(request: Request, session: Session, provider: str) -> None:
audit_log(
AuditEvent.LOGIN_SUCCESS, provider=provider, user_id=session.user_id,
email=session.email, org_id=session.org_id, ip=_client_ip(request))
_audit(request, AuditEvent.LOGIN_SUCCESS, provider=provider, user_id=session.user_id,
email=session.email, org_id=session.org_id)
def _complete_login(request: Request, provider: str, session: Session, *, broker_state: str,
next_raw: str) -> tuple:
"""Shared tail of the callback + password routes after credentials verified: audit success,
then either the native loopback redirect (no cookies) or the landing path. Returns
``(target_url, native)``."""
_login_success(request, session, provider)
if broker_state:
return _finish_native_login(
request, broker_state=broker_state, session=session, provider=provider), True
return _validate_post_login_target(next_raw) or "/", False
def _start_upstream_login(request: Request, p, *, audit_failure: bool, extra_pkce: dict[str, str]):
"""Run ``start_login`` and 302 to the IDP with the PKCE cookie set. That cookie
is the only server-controlled channel surviving the round trip (IDPs echo back
only code+state), so it carries the provider name plus ``extra_pkce``."""
"""Run ``start_login`` and 302 to the IDP with the PKCE cookie set. That cookie is the only
server-controlled channel surviving the round trip (IDPs echo back only code+state), so it
carries the provider name plus ``extra_pkce``."""
try:
ls = p.start_login(redirect_uri=_redirect_uri(request))
except ProviderError as e:
@@ -164,7 +170,6 @@ def _start_upstream_login(request: Request, p, *, audit_failure: bool, extra_pkc
# --- Public: login page + provider list ------------------------------------
@router.get("/login", name="login_page")
async def login_page(request: Request) -> HTMLResponse:
# ``next=`` is set by the gate's redirect but /login is reachable directly.
@@ -186,7 +191,6 @@ async def api_auth_providers() -> Any:
# --- Public: OAuth round trip ----------------------------------------------
@router.get("/auth/login", name="auth_login")
async def auth_login(request: Request, provider: str, next: str = ""):
p = get_provider(provider)
@@ -202,17 +206,16 @@ async def auth_login(request: Request, provider: str, next: str = ""):
return RedirectResponse(url=login_url, status_code=302)
resp = _start_upstream_login(
request, p, audit_failure=True, extra_pkce={"next": safe_next} if safe_next else {})
audit_log(AuditEvent.LOGIN_START, provider=provider, ip=_client_ip(request))
_audit(request, AuditEvent.LOGIN_START, provider=provider)
return resp
# --- Public: RFC 8252 native-app authorization (system browser + loopback + PKCE)
def _validate_loopback_redirect_uri(raw: str) -> str:
"""Accept only ``http://127.0.0.1[:port]/…`` / ``http://[::1][:port]/…``. Security
boundary: the route is public, so a non-loopback host would make the callback
an open redirect leaking a live code. ``localhost`` is rejected (RFC 8252 §8.3)."""
"""Accept only ``http://127.0.0.1[:port]/…`` / ``http://[::1][:port]/…``. Security boundary:
the route is public, so a non-loopback host would make the callback an open redirect leaking
a live code. ``localhost`` is rejected (RFC 8252 §8.3)."""
if not raw:
raise _http(400, "redirect_uri required")
parsed = urlparse(raw)
@@ -224,75 +227,65 @@ def _validate_loopback_redirect_uri(raw: str) -> str:
def _select_native_provider(provider: str):
"""Resolve the provider for a native authorize request. An empty ``provider``
auto-selects the single brokerable (non-password) session provider — so an
OIDC+basic deployment does not fail with "Unknown provider"; with none, a lone
password provider is still returned so the caller emits a 400 rather than 404."""
"""Resolve the provider for a native authorize request. An empty ``provider`` auto-selects
the single brokerable (non-password) session provider — so an OIDC+basic deployment does not
fail with "Unknown provider"; with none, a lone password provider is still returned so the
caller emits a 400 rather than 404."""
if provider:
return get_provider(provider)
sess_providers = list_session_providers()
native_eligible = [pp for pp in sess_providers if not getattr(pp, "supports_password", False)]
if len(native_eligible) == 1:
return native_eligible[0]
if not native_eligible and len(sess_providers) == 1:
return sess_providers[0]
return None
candidates = native_eligible or sess_providers
return candidates[0] if len(candidates) == 1 else None
@router.get("/auth/native/authorize", name="auth_native_authorize")
async def auth_native_authorize(
request: Request, provider: str = "", code_challenge: str = "",
code_challenge_method: str = "", redirect_uri: str = "", state: str = ""):
"""Begin an RFC 8252 native-app login: stash a pending broker authorization keyed
by an opaque ``broker_state`` riding in the gateway's own PKCE cookie (the
desktop's challenge/state never touch it), then run the normal upstream round
trip. Password providers go to the ``/login`` form instead."""
"""Begin an RFC 8252 native-app login: stash a pending broker authorization keyed by an
opaque ``broker_state`` riding in the gateway's own PKCE cookie (the desktop's
challenge/state never touch it), then run the normal upstream round trip. Password providers
go to the ``/login`` form instead."""
if code_challenge_method.upper() != "S256":
raise _http(400, "code_challenge_method must be S256")
if not code_challenge:
raise _http(400, "code_challenge required")
_validate_loopback_redirect_uri(redirect_uri)
p = _select_native_provider(provider)
if p is None:
raise _http(404, f"Unknown provider: {provider!r}")
if not getattr(p, "supports_session", True):
raise _http(400, f"Provider does not support native login: {p.name!r}")
try:
broker_state = native_flow.register_pending(
code_challenge=code_challenge, redirect_uri=redirect_uri, client_state=state,
client_ip=_client_ip(request))
except native_flow.NativeFlowError as e:
raise _http(503, str(e))
if getattr(p, "supports_password", False):
audit_log(AuditEvent.NATIVE_AUTHORIZE_START, provider=p.name, ip=_client_ip(request))
_audit(request, AuditEvent.NATIVE_AUTHORIZE_START, provider=p.name)
resp = RedirectResponse(url=f"{_prefix(request)}/login", status_code=302)
_set_pkce(resp, request, {"provider": p.name, "broker": broker_state})
return resp
resp = _start_upstream_login(
request, p, audit_failure=False, extra_pkce={"broker": broker_state})
audit_log(AuditEvent.NATIVE_AUTHORIZE_START, provider=p.name, ip=_client_ip(request))
_audit(request, AuditEvent.NATIVE_AUTHORIZE_START, provider=p.name)
return resp
@router.get("/auth/callback", name="auth_callback")
async def auth_callback(
request: Request, code: str = "", state: str = "", error: str = "", error_description: str = "",
):
request: Request, code: str = "", state: str = "", error: str = "",
error_description: str = ""):
pkce_raw = read_pkce_cookie(request)
if not pkce_raw:
audit_log(AuditEvent.LOGIN_FAILURE, reason="missing_pkce_cookie", ip=_client_ip(request))
_audit(request, AuditEvent.LOGIN_FAILURE, reason="missing_pkce_cookie")
raise _http(400, "Missing PKCE state cookie")
# ``next`` and ``broker`` come from the server-set cookie ONLY: the IDP
# echoes back just code+state, so any such query param is attacker controlled.
parts = parse_pkce_payload(pkce_raw)
provider_name = parts.get("provider", "")
broker_state = parts.get("broker", "")
p = get_provider(provider_name)
if p is None:
raise _http(400, f"Unknown provider in cookie: {provider_name!r}")
@@ -302,7 +295,6 @@ async def auth_callback(
if not state or state != parts.get("state", ""):
_login_failure(request, provider_name, "state_mismatch")
raise _http(400, "OAuth state mismatch (CSRF check failed)")
try:
session = p.complete_login(
code=code, state=state, code_verifier=parts.get("verifier", ""),
@@ -313,32 +305,23 @@ async def auth_callback(
except ProviderError as e:
_login_failure(request, provider_name, "provider_unreachable")
raise _http(503, f"Provider unreachable: {e}")
_login_success(request, session, provider_name)
https = detect_https(request)
prefix = _prefix(request)
if broker_state:
loopback = _finish_native_login(
request, broker_state=broker_state, session=session, provider=provider_name)
resp = RedirectResponse(url=loopback, status_code=302)
else:
landing = _validate_post_login_target(parts.get("next", "")) or "/"
resp = RedirectResponse(url=landing, status_code=302)
target, native = _complete_login(
request, provider_name, session, broker_state=parts.get("broker", ""),
next_raw=parts.get("next", ""))
resp = RedirectResponse(url=target, status_code=302)
if not native:
_set_session(resp, request, session)
clear_pkce_cookie(resp, use_https=https, prefix=prefix)
# Clear the one-shot auto-SSO loop-guard so it never suppresses a future
# silent attempt after logout.
prefix = _prefix(request)
clear_pkce_cookie(resp, use_https=detect_https(request), prefix=prefix)
# Clear the one-shot auto-SSO loop-guard so it never suppresses a future silent attempt.
clear_sso_attempt_cookie(resp, prefix=prefix)
return resp
# --- Public: password (non-redirect) login ---------------------------------
#
# Brute-force throttle: a process-local sliding window per client IP. Best
# effort defence-in-depth on top of the provider's constant-time verify (resets
# on restart; behind a proxy the IP is the proxy's unless X-Forwarded-For).
# Brute-force throttle: a process-local sliding window per client IP. Best-effort
# defence-in-depth on top of the provider's constant-time verify (resets on restart; behind a
# proxy the IP is the proxy's unless X-Forwarded-For).
_PW_RATE_MAX_ATTEMPTS = 10
_PW_RATE_WINDOW_SEC = 60.0
_pw_attempts: Dict[str, Deque[float]] = defaultdict(deque)
@@ -346,8 +329,8 @@ _pw_attempts_lock = threading.Lock()
def _password_rate_limited(ip: str) -> bool:
"""True if ``ip`` exceeded the budget; records the attempt when allowed. An empty
IP shares one bucket — fail-safe toward throttling."""
"""True if ``ip`` exceeded the budget; records the attempt when allowed. An empty IP shares
one bucket — fail-safe toward throttling."""
now = time.monotonic()
cutoff = now - _PW_RATE_WINDOW_SEC
with _pw_attempts_lock:
@@ -377,35 +360,30 @@ class _PasswordLoginBody(BaseModel):
async def auth_password_login(request: Request, body: _PasswordLoginBody):
"""Authenticate a username/password against a password provider.
Returns ``{"ok": true, "next": <path>}`` (the form POSTs via fetch, which
follows a 302 opaquely) and sets the session cookies; with a native ``broker``
handle in the PKCE cookie, ``next`` is the desktop's loopback redirect and NO
cookies are set. Failures are deliberately generic (no username/provider
oracle): unknown/non-password provider 404, bad credentials 401, store
unreachable 503, rate limited 429.
Returns ``{"ok": true, "next": <path>}`` (the form POSTs via fetch, which follows a 302
opaquely) and sets the session cookies; with a native ``broker`` handle in the PKCE cookie,
``next`` is the desktop's loopback redirect and NO cookies are set. Failures are deliberately
generic (no username/provider oracle): unknown/non-password provider 404, bad credentials
401, store unreachable 503, rate limited 429.
"""
ip = _client_ip(request)
if _password_rate_limited(ip):
if _password_rate_limited(_client_ip(request)):
_login_failure(request, body.provider, "rate_limited")
raise _http(429, "Too many login attempts. Try again shortly.")
p = get_provider(body.provider)
if p is None or not getattr(p, "supports_password", False):
_login_failure(request, body.provider, "unknown_password_provider")
raise _http(404, "Unknown provider")
# The native broker handle also records WHICH provider the flow was started
# for. Enforce equality BEFORE verifying credentials so a flow started for
# provider A cannot be completed with provider B's credentials.
# The native broker handle also records WHICH provider the flow was started for. Enforce
# equality BEFORE verifying credentials so a flow started for provider A cannot be completed
# with provider B's credentials.
pkce_raw = read_pkce_cookie(request)
pkce_parts = parse_pkce_payload(pkce_raw) if pkce_raw else {}
broker_state = pkce_parts.get("broker", "")
if broker_state and pkce_parts.get("provider", "") != body.provider:
audit_log(AuditEvent.NATIVE_TOKEN_FAILURE, provider=body.provider,
reason="provider_mismatch", ip=ip)
_audit(request, AuditEvent.NATIVE_TOKEN_FAILURE, provider=body.provider,
reason="provider_mismatch")
raise _http(400, "This native sign-in was started for a different provider; "
"use that provider's form or restart sign-in.")
try:
session = p.complete_password_login(username=body.username, password=body.password)
except InvalidCredentialsError:
@@ -417,36 +395,28 @@ async def auth_password_login(request: Request, body: _PasswordLoginBody):
except ProviderError as e:
_login_failure(request, body.provider, "provider_unreachable")
raise _http(503, f"Provider unreachable: {e}")
_login_success(request, session, body.provider)
if broker_state:
loopback = _finish_native_login(
request, broker_state=broker_state, session=session, provider=body.provider)
resp = JSONResponse({"ok": True, "next": loopback})
target, native = _complete_login(
request, body.provider, session, broker_state=broker_state, next_raw=body.next)
resp = JSONResponse({"ok": True, "next": target})
if native:
clear_pkce_cookie(resp, use_https=detect_https(request), prefix=_prefix(request))
return resp
resp = JSONResponse({"ok": True, "next": _validate_post_login_target(body.next) or "/"})
_set_session(resp, request, session)
else:
_set_session(resp, request, session)
return resp
@router.post("/auth/logout", name="auth_logout")
async def auth_logout(request: Request):
_at, rt = read_session_cookies(request)
if rt:
# Best-effort revoke on every provider; failures logged, never raised.
for provider in list_providers():
try:
provider.revoke_session(refresh_token=rt)
except Exception as e: # noqa: BLE001 — best-effort
_log.warning("dashboard-auth: revoke on %r failed: %s", provider.name, e)
# Best-effort revoke on every provider; failures logged, never raised.
for provider in list_providers() if rt else ():
try:
provider.revoke_session(refresh_token=rt)
except Exception as e: # noqa: BLE001 — best-effort
_log.warning("dashboard-auth: revoke on %r failed: %s", provider.name, e)
sess = getattr(request.state, "session", None)
audit_log(
AuditEvent.LOGOUT, provider=(sess.provider if sess else "unknown"),
user_id=(sess.user_id if sess else ""), ip=_client_ip(request))
_audit(request, AuditEvent.LOGOUT, provider=(sess.provider if sess else "unknown"),
user_id=(sess.user_id if sess else ""))
prefix = _prefix(request)
resp = RedirectResponse(url=f"{prefix}/login", status_code=302)
clear_session_cookies(resp, prefix=prefix)
@@ -456,7 +426,6 @@ async def auth_logout(request: Request):
# --- Auth-required: identity probe + WS ticket for the SPA -----------------
def _require_session(request: Request):
sess = getattr(request.state, "session", None)
if sess is None:
@@ -479,16 +448,13 @@ async def api_auth_ws_ticket(request: Request):
``Authorization`` on the upgrade); one ticket per WS."""
sess = _require_session(request)
from hermes_cli.dashboard_auth.ws_tickets import TTL_SECONDS, mint_ticket
ticket = mint_ticket(user_id=sess.user_id, provider=sess.provider)
audit_log(AuditEvent.WS_TICKET_MINTED, provider=sess.provider, user_id=sess.user_id,
ip=_client_ip(request))
_audit(request, AuditEvent.WS_TICKET_MINTED, provider=sess.provider, user_id=sess.user_id)
return {"ticket": ticket, "ttl_seconds": TTL_SECONDS}
# --- Public: RFC 8252 native-app token exchange + refresh ------------------
class _NativeTokenBody(BaseModel):
code: str
code_verifier: str
@@ -496,18 +462,16 @@ class _NativeTokenBody(BaseModel):
@router.post("/auth/native/token", name="auth_native_token")
async def auth_native_token(request: Request, body: _NativeTokenBody):
"""Exchange a loopback gateway code + PKCE verifier for bearer tokens. The code
is consumed on every path (no verifier oracle, no replay); any failure is a
generic 400. Tokens go in the JSON body; no cookie is set."""
"""Exchange a loopback gateway code + PKCE verifier for bearer tokens. The code is consumed
on every path (no verifier oracle, no replay); any failure is a generic 400. Tokens go in
the JSON body; no cookie is set."""
try:
session = native_flow.redeem_code(code=body.code, code_verifier=body.code_verifier)
except native_flow.CodeInvalid:
audit_log(AuditEvent.NATIVE_TOKEN_FAILURE, reason="invalid_code_or_pkce",
ip=_client_ip(request))
_audit(request, AuditEvent.NATIVE_TOKEN_FAILURE, reason="invalid_code_or_pkce")
raise _http(400, "Invalid or expired authorization code.")
audit_log(
AuditEvent.NATIVE_TOKEN_SUCCESS, provider=session.provider, user_id=session.user_id,
ip=_client_ip(request))
_audit(request, AuditEvent.NATIVE_TOKEN_SUCCESS, provider=session.provider,
user_id=session.user_id)
return _bearer_payload(session)
@@ -518,9 +482,9 @@ class _NativeRefreshBody(BaseModel):
@router.post("/auth/native/refresh", name="auth_native_refresh")
async def auth_native_refresh(request: Request, body: _NativeRefreshBody):
"""Rotate a desktop-held refresh token (mirrors the gate's ``_attempt_refresh``):
every provider rejecting the RT -> 401 ``session_expired`` (desktop re-logs);
none rotated and one unreachable -> 503."""
"""Rotate a desktop-held refresh token (mirrors the gate's ``_attempt_refresh``): every
provider rejecting the RT -> 401 ``session_expired`` (desktop re-logs); none rotated and one
unreachable -> 503."""
if not body.refresh_token:
raise _http(400, "refresh_token required")
try:
@@ -530,13 +494,10 @@ async def auth_native_refresh(request: Request, body: _NativeRefreshBody):
except ProviderError as e:
raise _http(503, f"Auth provider {str(e)!r} unreachable")
if session is not None:
audit_log(
AuditEvent.REFRESH_SUCCESS, provider=session.provider, user_id=session.user_id,
ip=_client_ip(request))
_audit(request, AuditEvent.REFRESH_SUCCESS, provider=session.provider,
user_id=session.user_id)
return _bearer_payload(session)
audit_log(AuditEvent.REFRESH_FAILURE, reason="all_providers_rejected_rt",
ip=_client_ip(request))
_audit(request, AuditEvent.REFRESH_FAILURE, reason="all_providers_rejected_rt")
return JSONResponse(
{"error": "session_expired",
"detail": "Refresh token expired or invalid; start a new sign-in."},
status_code=401)
"detail": "Refresh token expired or invalid; start a new sign-in."}, status_code=401)
hermes_cli/dashboard_auth/token_auth.py
+12 -24
View File
@@ -1,16 +1,11 @@
"""Route-agnostic non-interactive (bearer-token) auth seam for the dashboard.
Any machine-credential provider plugs in here (the drain bearer-secret plugin is
the first consumer). A route opts in by registering its exact path via
:func:`register_token_route`; only registered paths are token-authable, so the
auth surface of existing routes never widens.
:func:`token_auth_middleware` runs OUTERMOST (installed last in web_server.py)
and fully owns the decision for a token route: a recognised token attaches
``request.state.token_principal`` + ``token_authenticated`` (the cookie gates
honour that flag and never bounce to /login); otherwise 401, or 503 when a
provider's backing store was unreachable. Fails closed: no provider, no token,
or an unrecognised token is always 401.
Any machine-credential provider plugs in here. A route opts in by registering its exact path via
:func:`register_token_route`; only registered paths are token-authable, so the auth surface of
existing routes never widens. :func:`token_auth_middleware` runs OUTERMOST (installed last) and
owns the decision for a token route: a recognised token attaches ``request.state.token_principal``
+ ``token_authenticated`` (the cookie gates honour that flag and never bounce to /login);
otherwise 401, or 503 when a provider's backing store was unreachable. Fails closed.
"""
from __future__ import annotations
@@ -34,8 +29,7 @@ _lock = threading.Lock()
def register_token_route(path: str) -> None:
"""Mark ``path`` (exact match) as token-authable. Idempotent; does NOT make
the route public — it authenticates by token instead of by cookie."""
"""Mark ``path`` (exact match) as token-authable. Idempotent; does NOT make the route public."""
with _lock:
_token_routes.add(path)
@@ -53,13 +47,9 @@ def clear_token_routes() -> None:
def authenticate_token(request: Request) -> Tuple[Optional[TokenPrincipal], Optional[str]]:
"""Try every token provider against the request's bearer token.
Returns ``(principal, None)`` on success; ``(None, None)`` for no token or
no recogniser (401); ``(None, name)`` when no provider accepted it AND at
least one was unreachable (caller surfaces 503, not "bad credentials").
Never raises — a buggy provider must not 500 the gate.
"""
"""Try every token provider against the request's bearer token. Returns ``(principal, None)``
on success; ``(None, None)`` for no token or no recogniser (401); ``(None, name)`` when no
provider accepted it AND at least one was unreachable (caller surfaces 503). Never raises."""
token = extract_bearer_token(request)
if not token:
return None, None
@@ -84,18 +74,16 @@ def authenticate_token(request: Request) -> Tuple[Optional[TokenPrincipal], Opti
async def token_auth_middleware(
request: Request, call_next: Callable[[Request], Awaitable[Response]]) -> Response:
"""Outermost auth seam: pass-through for unregistered paths; for a token
route, valid token -> attach principal + flag, unreachable -> 503, else 401."""
"""Pass-through for unregistered paths; for a token route, valid token -> attach principal +
flag, unreachable -> 503, else 401."""
path = request.url.path
if not is_token_route(path):
return await call_next(request)
principal, unreachable = authenticate_token(request)
if principal is not None:
request.state.token_principal = principal
request.state.token_authenticated = True
return await call_next(request)
if unreachable:
audit_log(
AuditEvent.TOKEN_AUTH_FAILURE, provider=unreachable, reason="provider_unreachable",
hermes_cli/dashboard_auth/ws_tickets.py
+20 -39
View File
@@ -1,23 +1,15 @@
"""WS-upgrade auth credentials for gated mode.
Browsers cannot set ``Authorization`` on a WebSocket upgrade, and gated mode
has no token injected into the SPA, so two credential shapes exist:
1. **Single-use browser tickets** (``mint_ticket`` / ``consume_ticket``): the
SPA fetches one via authenticated ``POST /api/auth/ws-ticket`` and passes it
as ``?ticket=`` on the upgrade. Single-use, 30 s TTL — a leak is uninteresting.
2. **A process-lifetime internal credential** (``internal_ws_credential`` /
``consume_internal_credential``) for *server-spawned* WS clients (the
embedded-TUI PTY child on ``/api/ws`` + ``/api/pub``), which read their attach
URL once and reuse it on every reconnect, possibly >30 s after a cold boot.
Minted once, never expires, multi-use, and never injected into any HTML/SPA:
it leaves the process only via the child's environment, so browser XSS cannot
read it; a leak grants no more than a ticket does (same two endpoints, same
Origin/host guards).
In-memory (single process). Functional API so tests can patch ``time.time``.
Browsers cannot set ``Authorization`` on a WebSocket upgrade, and gated mode has no token
injected into the SPA, so two credential shapes exist: (1) single-use browser tickets
(``mint_ticket`` / ``consume_ticket``) fetched via authenticated ``POST /api/auth/ws-ticket`` and
passed as ``?ticket=`` on the upgrade — 30 s TTL, a leak is uninteresting; (2) a process-lifetime
internal credential (``internal_ws_credential`` / ``consume_internal_credential``) for
*server-spawned* WS clients (the embedded-TUI PTY child on ``/api/ws`` + ``/api/pub``), which
reuse their attach URL on every reconnect, possibly >30 s after boot — minted once, never expires,
multi-use, never injected into any HTML/SPA (leaves the process only via the child's environment,
so browser XSS cannot read it; grants no more than a ticket). In-memory; ``time.time`` patchable.
"""
from __future__ import annotations
import secrets
@@ -25,18 +17,14 @@ import threading
import time
from typing import Any, Dict, Optional, Tuple
#: Ticket TTL: long enough for ``getWsTicket()`` -> open WS, short enough that
#: a leaked ticket is uninteresting.
#: Long enough for ``getWsTicket()`` -> open WS, short enough that a leaked ticket is uninteresting.
TTL_SECONDS = 30
_lock = threading.Lock()
_tickets: Dict[str, Tuple[int, Dict[str, Any]]] = {} # ticket -> (expires_at, info)
_internal_credential: Optional[str] = None # lazily minted; guarded by ``_lock``
#: Process-lifetime internal credential, lazily minted; guarded by ``_lock``.
_internal_credential: Optional[str] = None
#: Identity recorded for internal-credential connections so audit logs
#: distinguish them from browser-initiated tickets.
#: Identity recorded for internal-credential connections (audit logs distinguish them from tickets).
INTERNAL_USER_ID = "server-internal"
INTERNAL_PROVIDER = "server-internal"
@@ -46,8 +34,8 @@ class TicketInvalid(Exception):
def mint_ticket(*, user_id: str, provider: str) -> str:
"""Generate a one-shot base64url ticket (32 random bytes) bound to this
identity; ``consume_ticket`` hands the ``info`` dict back to the WS handler."""
"""One-shot base64url ticket (32 random bytes) bound to this identity; ``consume_ticket``
hands the ``info`` dict back to the WS handler."""
ticket = secrets.token_urlsafe(32)
info = {"user_id": user_id, "provider": provider, "minted_at": int(time.time())}
with _lock:
@@ -57,8 +45,7 @@ def mint_ticket(*, user_id: str, provider: str) -> str:
def consume_ticket(ticket: str) -> Dict[str, Any]:
"""Validate and consume (single-use: a second call raises). Raises
:class:`TicketInvalid` on missing/expired/used."""
"""Validate and consume (single-use). Raises :class:`TicketInvalid` on missing/expired/used."""
now = int(time.time())
with _lock:
entry = _tickets.pop(ticket, None)
@@ -80,11 +67,8 @@ def _gc_expired_locked() -> None:
def internal_ws_credential() -> str:
"""Return the process-lifetime internal WS credential, minting it once.
Stable, multi-use, never expires; never injected into the SPA or returned
over REST — only passed to a spawned child via its environment.
"""
"""Process-lifetime internal WS credential, minted once. Never injected into the SPA or
returned over REST — only passed to a spawned child via its environment."""
global _internal_credential
with _lock:
if _internal_credential is None:
@@ -93,12 +77,9 @@ def internal_ws_credential() -> str:
def consume_internal_credential(value: str) -> Dict[str, Any]:
"""Validate an internal credential (NOT single-use). Raises :class:`TicketInvalid`.
Returns the fixed server-internal ``{user_id, provider}`` info dict, mirroring
``consume_ticket``. Constant-time compare; any value is rejected until a
credential has been minted.
"""
"""Validate an internal credential (NOT single-use); returns the fixed server-internal
``{user_id, provider}`` info dict, mirroring ``consume_ticket``. Constant-time compare; any
value is rejected until a credential has been minted."""
with _lock:
expected = _internal_credential
if not value or expected is None: