Release exp-v0.52.43: trusted-header auth / reverse-proxy SSO (#5568, #3351) (#5956)

* feat(#3351): add trusted-header auth core

* fix(#3351): reconcile trusted sessions per request

* fix(#3351): preserve valid trusted group mappings

* fix(#3351): defer trusted logout reconciliation until after csrf

* fix(auth): clear _pending_set_cookies in reset_trusted_auth_request_state (keep-alive cookie cross-request leak)

Codex gate CORE finding: a Set-Cookie queued in request N but not flushed
survived onto the reused HTTP/1.1 handler and could be emitted by request N+1
— after trusted-identity rotation on logout this could overwrite a valid login
cookie and 401 the user. Reset the queue at the per-request boundary + add a
keep-alive regression test.

* docs(.env.example): document trusted-header auth / reverse-proxy SSO env vars

HERMES_WEBUI_TRUSTED_AUTH_HEADER + TRUSTED_PROXY_CIDRS (with exact-proxy-IP
security warning) + optional groups-header/JSON group-profile-map/logout-url.

* Release exp-v0.52.43: trusted-header auth / reverse-proxy SSO (#5568, #3351) + keep-alive cookie fix + docs

---------

Co-authored-by: Rod Boev <rod.boev@gmail.com>
Co-authored-by: nesquena-hermes <agent@nesquena-hermes>
This commit is contained in:
nesquena-hermes
2026-07-11 16:34:14 -07:00
committed by GitHub
parent 18793a1ba4
commit b7b6b3a4c8
8 changed files with 1271 additions and 46 deletions
+21
View File
@@ -77,6 +77,27 @@
# HERMES_WEBUI_TRUST_FORWARDED_FOR=1
# HERMES_WEBUI_TRUST_FORWARDED_PROTO=1
# ── Trusted-header auth / reverse-proxy SSO (opt-in; OFF by default) ──────────
# Delegate authentication to a reverse proxy you run (Authelia, oauth2-proxy,
# corporate SSO, …) that authenticates the user and forwards their identity in a
# header. When HERMES_WEBUI_TRUSTED_AUTH_HEADER is set, the WebUI treats a request
# as that identity — but ONLY when the request's un-spoofable raw socket peer is
# an allowlisted proxy (HERMES_WEBUI_TRUSTED_PROXY_CIDRS below). With no header
# set, this path is disabled and normal password/passkey auth applies.
#
# ⚠️ SECURITY: HERMES_WEBUI_TRUSTED_PROXY_CIDRS MUST list the EXACT IP(s) of your
# proxy — e.g. 10.0.0.5/32 or 127.0.0.1/32. NEVER use 0.0.0.0/0 or a broad range:
# any client whose peer IP falls in the range could forge the identity header and
# log in as anyone. If the allowlist is empty/malformed, the header is ignored
# (fails closed). Loopback is trusted by default. Set the CIDR to your proxy only.
# HERMES_WEBUI_TRUSTED_AUTH_HEADER=Remote-User
# HERMES_WEBUI_TRUSTED_PROXY_CIDRS=10.0.0.5/32
# Optional: map a forwarded groups header to WebUI profiles (JSON object), and a
# logout URL to redirect to after clearing the local session (SSO single-logout).
# HERMES_WEBUI_TRUSTED_GROUPS_HEADER=Remote-Groups
# HERMES_WEBUI_GROUP_PROFILE_MAP={"admins":"default","team-a":"projecta"}
# HERMES_WEBUI_TRUSTED_AUTH_LOGOUT_URL=https://sso.example.com/logout
# ── Uploads & limits ─────────────────────────────────────────────────────────
# Where chat attachments are stored (default: <state dir>/attachments).
+2
View File
@@ -5,6 +5,8 @@
### Added
- **Reverse-proxy SSO via trusted identity headers (opt-in, off by default).** You can now put the WebUI behind a reverse proxy you run (Authelia, oauth2-proxy, corporate SSO, …) that authenticates the user and forwards their identity in a header. Set `HERMES_WEBUI_TRUSTED_AUTH_HEADER` (e.g. `Remote-User`) plus `HERMES_WEBUI_TRUSTED_PROXY_CIDRS` with the **exact** IP(s) of your proxy — the header is honored *only* when the request's un-spoofable socket peer is an allowlisted proxy, so a direct client can't forge it. Optional group→profile mapping (`HERMES_WEBUI_GROUP_PROFILE_MAP`, JSON) and SSO single-logout URL are supported. Fails closed on empty/malformed config; existing password/passkey auth is unchanged when this is not configured. See `.env.example` for the full setup and the security warning. Thanks @rodboev. (#5568, #3351)
- **Deep-link straight to a profile with `?profile=<name>`.** Opening the WebUI with a `?profile=NAME` query parameter now switches to that profile on boot. The name is validated against the existing profile allowlist (a nonexistent profile boots normally, path-traversal / cross-profile attempts are rejected), the switch goes through the same access-enforced `switchToProfile`, and boot fails safe on anything invalid. Thanks @rodboev. (#5730, #5682)
### Fixed
+359 -29
View File
@@ -101,9 +101,33 @@ def _warn_auth_persistence_failure(prefix: str, artifact: Path, exc: Exception,
_SESSIONS_FILE = STATE_DIR / '.sessions.json'
_TRUSTED_AUTH_HEADER_ENV = 'HERMES_WEBUI_TRUSTED_AUTH_HEADER'
_TRUSTED_GROUPS_HEADER_ENV = 'HERMES_WEBUI_TRUSTED_GROUPS_HEADER'
_TRUSTED_GROUP_PROFILE_MAP_ENV = 'HERMES_WEBUI_GROUP_PROFILE_MAP'
_TRUSTED_AUTH_LOGOUT_URL_ENV = 'HERMES_WEBUI_TRUSTED_AUTH_LOGOUT_URL'
_TRUSTED_AUTH_WARNINGS_EMITTED: set[str] = set()
def _load_sessions() -> dict[str, float]:
def _warn_trusted_auth_once(key: str, message: str, *args) -> None:
if key in _TRUSTED_AUTH_WARNINGS_EMITTED:
return
_TRUSTED_AUTH_WARNINGS_EMITTED.add(key)
logger.warning(message, *args)
def _session_expiry(record) -> float | None:
if isinstance(record, dict):
expiry = record.get('expiry', record.get('expires_at'))
else:
expiry = record
try:
expiry_f = float(expiry)
except (TypeError, ValueError):
return None
return expiry_f
def _load_sessions() -> dict[str, float | dict]:
"""Load persisted sessions from STATE_DIR, pruning expired entries.
Returns an empty dict on any read or parse error so startup is never
@@ -141,11 +165,23 @@ def _load_sessions() -> dict[str, float]:
)
return {}
now = time.time()
return {t: exp for t, exp in data.items()
if isinstance(t, str) and isinstance(exp, (int, float)) and exp > now}
sessions: dict[str, float | dict] = {}
for token, record in data.items():
if not isinstance(token, str) or not token:
continue
expiry = _session_expiry(record)
if expiry is None or expiry <= now:
continue
if isinstance(record, dict):
normalized = dict(record)
normalized['expiry'] = expiry
sessions[token] = normalized
else:
sessions[token] = expiry
return sessions
def _save_sessions(sessions: dict[str, float]) -> None:
def _save_sessions(sessions: dict[str, float | dict]) -> None:
"""Atomically persist sessions to STATE_DIR/.sessions.json (0600).
Uses a temp file + os.replace() so a crash mid-write never leaves a
@@ -500,11 +536,12 @@ def get_oidc_startup_warning() -> str | None:
def is_auth_enabled() -> bool:
"""True if password auth, passkeys, or OIDC login is configured."""
"""True if password auth, passkeys, OIDC login, or trusted-header auth is configured."""
return (
is_password_auth_enabled()
or are_passkeys_enabled()
or is_oidc_auth_enabled()
or is_trusted_auth_enabled()
)
@@ -539,11 +576,22 @@ def verify_password(plain: str) -> bool:
return False
def create_session() -> str:
def create_session(*, auth_type: str | None = None, username: str | None = None, bound_profile: str | None = None) -> str:
"""Create a new auth session. Returns signed cookie value."""
token = secrets.token_hex(32)
expiry = time.time() + _resolve_session_ttl()
record: float | dict
if any(value is not None for value in (auth_type, username, bound_profile)):
record = {
'expiry': expiry,
'auth_type': auth_type,
'username': username,
'bound_profile': bound_profile,
}
else:
record = expiry
with _SESSIONS_LOCK:
_sessions[token] = time.time() + _resolve_session_ttl()
_sessions[token] = record
_save_sessions(_sessions)
sig = hmac.new(_signing_key(), token.encode(), hashlib.sha256).hexdigest()
return f"{token}.{sig}"
@@ -553,7 +601,7 @@ def _prune_expired_sessions():
"""Remove all expired session entries to prevent unbounded memory growth."""
now = time.time()
with _SESSIONS_LOCK:
expired = [t for t, exp in _sessions.items() if now > exp]
expired = [t for t, record in _sessions.items() if (expiry := _session_expiry(record)) is None or now > expiry]
if expired:
for token in expired:
_sessions.pop(token, None)
@@ -576,14 +624,287 @@ def verify_session(cookie_value: str) -> bool:
if not valid:
return False
with _SESSIONS_LOCK:
expiry = _sessions.get(token)
if not expiry or time.time() > expiry:
expiry = _session_expiry(_sessions.get(token))
if expiry is None or time.time() > expiry:
_sessions.pop(token, None)
_save_sessions(_sessions)
return False
return True
def _trusted_auth_header_name() -> str | None:
name = os.getenv(_TRUSTED_AUTH_HEADER_ENV, '').strip()
if not name:
return None
if not _COOKIE_NAME_RE.match(name):
_warn_trusted_auth_once(
'trusted-auth-header',
'Ignoring invalid %s=%r; trusted-header auth rejects every request',
_TRUSTED_AUTH_HEADER_ENV,
name,
)
return None
return name
def _trusted_auth_header_configured() -> bool:
return bool(os.getenv(_TRUSTED_AUTH_HEADER_ENV, '').strip())
def _trusted_group_profile_map() -> dict[str, str] | None:
raw = os.getenv(_TRUSTED_GROUP_PROFILE_MAP_ENV, '').strip()
if not raw:
return None
try:
data = json.loads(raw)
except json.JSONDecodeError:
_warn_trusted_auth_once(
'trusted-group-map',
'Ignoring invalid %s JSON; trusted-header auth falls back to default profile binding',
_TRUSTED_GROUP_PROFILE_MAP_ENV,
)
return {}
if not isinstance(data, dict):
_warn_trusted_auth_once(
'trusted-group-map-type',
'Ignoring non-dict %s; trusted-header auth falls back to default profile binding',
_TRUSTED_GROUP_PROFILE_MAP_ENV,
)
return {}
mapping: dict[str, str] = {}
for group, profile in data.items():
group_name = str(group or '').strip()
profile_name = str(profile or '').strip()
if not group_name or not profile_name:
_warn_trusted_auth_once(
'trusted-group-map-entry',
'Ignoring invalid entry in %s; trusted-header auth falls back to default profile binding',
_TRUSTED_GROUP_PROFILE_MAP_ENV,
)
continue
mapping[group_name] = profile_name
return mapping
def _trusted_groups_header_value(handler) -> list[str]:
header_name = os.getenv(_TRUSTED_GROUPS_HEADER_ENV, '').strip()
if not header_name:
return []
try:
raw = handler.headers.get(header_name, '')
except Exception:
return []
if not raw:
return []
values = []
for part in str(raw).replace('\n', ',').split(','):
part = part.strip()
if part:
values.append(part)
return values
def _trusted_auth_username(handler) -> str | None:
header_name = _trusted_auth_header_name()
if not header_name:
return None
try:
raw = handler.headers.get(header_name, '')
except Exception:
return None
username = str(raw or '').strip()
return username or None
def _trusted_auth_bound_profile(handler) -> str | None:
mapping = _trusted_group_profile_map()
if mapping is None:
return None
groups = set(_trusted_groups_header_value(handler))
for group, profile in mapping.items():
if group in groups:
return profile
return 'default'
def _queue_pending_cookie(handler, cookie_header: str) -> None:
if not cookie_header:
return
pending = getattr(handler, '_pending_set_cookies', None)
if pending is None:
pending = []
handler._pending_set_cookies = pending
pending.append(cookie_header)
def _auth_cookie_header(cookie_value, handler=None) -> str:
cookie = http.cookies.SimpleCookie()
name = _resolve_cookie_name()
cookie[name] = cookie_value
cookie[name]['httponly'] = True
cookie[name]['samesite'] = 'Lax'
cookie[name]['path'] = '/'
cookie[name]['max-age'] = str(_resolve_session_ttl())
if _is_secure_context(handler):
cookie[name]['secure'] = True
return cookie[name].OutputString()
def _clear_auth_cookie_header() -> str:
cookie = http.cookies.SimpleCookie()
name = _resolve_cookie_name()
cookie[name] = ''
cookie[name]['httponly'] = True
cookie[name]['path'] = '/'
cookie[name]['samesite'] = 'Lax'
cookie[name]['max-age'] = '0'
return cookie[name].OutputString()
def _build_profile_cookie_header(name: str, session_cookie_value: str | None) -> str:
from api.helpers import build_profile_cookie
return build_profile_cookie(name, session_cookie_value=session_cookie_value)
def _request_profile_matches_bound(bound_profile: str | None) -> bool:
if not bound_profile:
return True
try:
from api.profiles import get_active_profile_name, _profiles_match
return _profiles_match(bound_profile, get_active_profile_name())
except Exception:
return False
def get_session_info(cookie_value: str) -> dict | None:
if not verify_session(cookie_value):
return None
token = _session_token_from_cookie_value(cookie_value)
if not token:
return None
with _SESSIONS_LOCK:
record = _sessions.get(token)
expiry = _session_expiry(record)
if expiry is None:
return None
info: dict[str, object] = {'token': token, 'expiry': expiry}
if isinstance(record, dict):
info.update({k: v for k, v in record.items() if k != 'expiry'})
if 'bound_profile' not in info and isinstance(info.get('profile'), str):
info['bound_profile'] = info.get('profile')
info.setdefault('auth_type', None)
info.setdefault('username', None)
info.setdefault('bound_profile', None)
return info
def session_bound_profile(cookie_value: str) -> str | None:
info = get_session_info(cookie_value)
if not info:
return None
bound_profile = info.get('bound_profile')
bound_profile = str(bound_profile or '').strip()
return bound_profile or None
def is_trusted_auth_enabled() -> bool:
return _trusted_auth_header_configured()
def get_trusted_auth_logout_url() -> str | None:
value = os.getenv(_TRUSTED_AUTH_LOGOUT_URL_ENV, '').strip()
return value or None
def _remember_trusted_auth_session(handler, info: dict | None, cookie_value: str | None = None) -> dict | None:
handler._trusted_auth_session_reconciled = info
if info and info.get('auth_type') == 'trusted':
handler._trusted_auth_session_info = info
handler._trusted_auth_session_cookie_value = cookie_value
return info
def reset_trusted_auth_request_state(handler) -> None:
for name in (
'_trusted_auth_session_reconciled',
'_trusted_auth_session_rejected',
'_trusted_auth_session_info',
'_trusted_auth_session_cookie_value',
# Clear any auth cookie queued by a prior request but not yet flushed.
# The handler is reused across HTTP/1.1 keep-alive requests, so a stale
# queued Set-Cookie would otherwise cross the request boundary and be
# emitted by a later response — e.g. after trusted-identity rotation on
# logout it could overwrite a subsequent valid login cookie and 401 the
# user. Reset it at the per-request boundary (server.py do_GET/do_POST).
'_pending_set_cookies',
):
try:
delattr(handler, name)
except AttributeError:
pass
def _apply_trusted_session_profile(handler, bound_profile: str | None, cookie_value: str) -> None:
if bound_profile is None:
return
from api.helpers import get_profile_cookie
from api.profiles import set_request_profile
set_request_profile(bound_profile)
if get_profile_cookie(handler) != bound_profile:
_queue_pending_cookie(handler, _build_profile_cookie_header(bound_profile, cookie_value))
def ensure_trusted_auth_session(handler) -> dict | None:
if hasattr(handler, '_trusted_auth_session_reconciled'):
return handler._trusted_auth_session_reconciled
cookie_value = parse_cookie(handler)
info = get_session_info(cookie_value) if cookie_value and verify_session(cookie_value) else None
if info and info.get('auth_type') != 'trusted':
return _remember_trusted_auth_session(handler, info)
if not is_trusted_auth_enabled():
if info:
invalidate_session(cookie_value)
handler._trusted_auth_session_rejected = True
return _remember_trusted_auth_session(handler, None)
from api.routes import _raw_peer_is_trusted_proxy
if not _raw_peer_is_trusted_proxy(handler):
if info:
invalidate_session(cookie_value)
handler._trusted_auth_session_rejected = True
return _remember_trusted_auth_session(handler, None)
username = _trusted_auth_username(handler)
if not username:
if info:
invalidate_session(cookie_value)
handler._trusted_auth_session_rejected = True
return _remember_trusted_auth_session(handler, None)
bound_profile = _trusted_auth_bound_profile(handler)
if info and info.get('username') == username and info.get('bound_profile') == bound_profile:
_apply_trusted_session_profile(handler, bound_profile, cookie_value)
return _remember_trusted_auth_session(handler, info, cookie_value)
if info:
invalidate_session(cookie_value)
cookie_value = create_session(
auth_type='trusted',
username=username,
bound_profile=bound_profile,
)
_queue_pending_cookie(handler, _auth_cookie_header(cookie_value, handler))
_apply_trusted_session_profile(handler, bound_profile, cookie_value)
info = get_session_info(cookie_value)
return _remember_trusted_auth_session(handler, info, cookie_value)
def trusted_session_allows_active_profile(info: dict | None) -> bool:
if not info:
return True
return _request_profile_matches_bound(str(info.get('bound_profile') or '') or None)
def _session_token_from_cookie_value(cookie_value: str) -> str | None:
"""Return the raw server-side session token from a signed cookie value."""
if not cookie_value or '.' not in cookie_value:
@@ -743,9 +1064,33 @@ def check_auth(handler, parsed) -> bool:
or parsed.path.startswith('/session/static/')
):
return True
# Check session cookie
cookie_val = parse_cookie(handler)
if cookie_val and verify_session(cookie_val):
has_session = bool(cookie_val and verify_session(cookie_val))
if parsed.path == '/api/auth/logout':
if has_session:
return True
body = b'{"error":"Authentication required"}'
handler.send_response(401)
handler.send_header('Content-Type', 'application/json')
handler.send_header('Content-Length', str(len(body)))
handler.end_headers()
handler.wfile.write(body)
return False
session_info = ensure_trusted_auth_session(handler)
if session_info:
if not trusted_session_allows_active_profile(session_info):
if parsed.path.startswith('/api/'):
body = b'{"error":"Profile access forbidden"}'
handler.send_response(403)
handler.send_header('Content-Type', 'application/json')
else:
body = b'Profile access forbidden'
handler.send_response(403)
handler.send_header('Content-Type', 'text/plain; charset=utf-8')
handler.send_header('Content-Length', str(len(body)))
handler.end_headers()
handler.wfile.write(body)
return False
return True
# Not authorized
if parsed.path.startswith('/api/'):
@@ -866,24 +1211,9 @@ def _is_secure_context(handler=None) -> bool:
def set_auth_cookie(handler, cookie_value) -> None:
"""Set the auth cookie on the response."""
cookie = http.cookies.SimpleCookie()
name = _resolve_cookie_name()
cookie[name] = cookie_value
cookie[name]['httponly'] = True
cookie[name]['samesite'] = 'Lax'
cookie[name]['path'] = '/'
cookie[name]['max-age'] = str(_resolve_session_ttl())
if _is_secure_context(handler):
cookie[name]['secure'] = True
handler.send_header('Set-Cookie', cookie[name].OutputString())
handler.send_header('Set-Cookie', _auth_cookie_header(cookie_value, handler))
def clear_auth_cookie(handler) -> None:
"""Clear the auth cookie on the response."""
cookie = http.cookies.SimpleCookie()
name = _resolve_cookie_name()
cookie[name] = ''
cookie[name]['httponly'] = True
cookie[name]['path'] = '/'
cookie[name]['max-age'] = '0'
handler.send_header('Set-Cookie', cookie[name].OutputString())
handler.send_header('Set-Cookie', _clear_auth_cookie_header())
+35 -3
View File
@@ -188,6 +188,15 @@ def _security_headers(handler):
)
def flush_pending_auth_cookies(handler) -> None:
pending = getattr(handler, '_pending_set_cookies', None)
if not pending:
return
handler._pending_set_cookies = []
for cookie in pending:
handler.send_header('Set-Cookie', cookie)
def _accepts_gzip(handler) -> bool:
"""Check if the client accepts gzip encoding."""
headers = getattr(handler, 'headers', None)
@@ -250,6 +259,7 @@ def j(handler, payload, status: int=200, extra_headers: dict=None, *, pretty: bo
handler.send_header('Content-Length', str(len(body)))
handler.send_header('Cache-Control', 'no-store')
_security_headers(handler)
flush_pending_auth_cookies(handler)
if extra_headers:
for k, v in extra_headers.items():
handler.send_header(k, v)
@@ -273,6 +283,7 @@ def t(
if extra_headers:
for k, v in extra_headers.items():
handler.send_header(k, v)
flush_pending_auth_cookies(handler)
_safe_write(handler, body)
@@ -720,7 +731,7 @@ def get_profile_cookie(handler) -> str | None:
return raw_val if _valid_profile_name(raw_val) else None
def build_profile_cookie(name: str, handler=None) -> str:
def build_profile_cookie(name: str, handler=None, *, session_cookie_value: str | None = None) -> str:
"""Build a Set-Cookie header value for the active-profile cookie.
Always persist the selected profile in the cookie, including 'default'.
@@ -746,8 +757,16 @@ def build_profile_cookie(name: str, handler=None) -> str:
except Exception:
_auth_on = False
if _auth_on and handler is None:
raise RuntimeError("build_profile_cookie requires a request handler when auth is enabled (to bind the profile cookie to the session)")
if handler is not None:
if session_cookie_value is None:
raise RuntimeError("build_profile_cookie requires a request handler when auth is enabled (to bind the profile cookie to the session)")
if session_cookie_value is not None:
try:
from api.auth import sign_profile_cookie_value
value = sign_profile_cookie_value(name, session_cookie_value)
except Exception as exc:
logger.warning("Failed to sign active profile cookie", exc_info=True)
raise RuntimeError("could not sign active profile cookie") from exc
elif handler is not None:
try:
from api.auth import is_auth_enabled, parse_cookie, sign_profile_cookie_value
if is_auth_enabled():
@@ -760,3 +779,16 @@ def build_profile_cookie(name: str, handler=None) -> str:
cookie[cookie_name]['httponly'] = True
cookie[cookie_name]['samesite'] = 'Lax'
return cookie[cookie_name].OutputString()
def clear_profile_cookie(handler) -> None:
import http.cookies as _hc
cookie = _hc.SimpleCookie()
cookie_name = get_profile_cookie_name()
cookie[cookie_name] = ''
cookie[cookie_name]['path'] = '/'
cookie[cookie_name]['httponly'] = True
cookie[cookie_name]['samesite'] = 'Lax'
cookie[cookie_name]['max-age'] = '0'
handler.send_header('Set-Cookie', cookie[cookie_name].OutputString())
+48 -9
View File
@@ -11711,6 +11711,8 @@ def handle_get(handler, parsed) -> bool:
if is_auth_enabled():
cookie_val = parse_cookie(handler)
if not cookie_val:
cookie_val = getattr(handler, "_trusted_auth_session_cookie_value", None)
if cookie_val and verify_session(cookie_val):
csrf_token = csrf_token_for_session(cookie_val) or ""
except Exception:
@@ -11827,19 +11829,27 @@ def handle_get(handler, parsed) -> bool:
return True
if parsed.path == "/api/auth/status":
from api.auth import _passkey_feature_flag_enabled, get_password_hash, is_auth_enabled, is_oidc_auth_enabled, parse_cookie, verify_session
from api.auth import (
_passkey_feature_flag_enabled,
ensure_trusted_auth_session,
get_password_hash,
is_auth_enabled,
is_oidc_auth_enabled,
is_trusted_auth_enabled,
)
from api.passkeys import registered_credentials
logged_in = False
session_info = None
auth_enabled = is_auth_enabled()
oidc_enabled = is_oidc_auth_enabled()
if auth_enabled:
cv = parse_cookie(handler)
logged_in = bool(cv and verify_session(cv))
session_info = ensure_trusted_auth_session(handler)
logged_in = bool(session_info)
passkey_flag = _passkey_feature_flag_enabled()
passkeys = registered_credentials() if passkey_flag else []
password_auth_enabled = get_password_hash() is not None
return j(handler, {
payload = {
"auth_enabled": auth_enabled,
"logged_in": logged_in,
"oidc_enabled": oidc_enabled,
@@ -11849,7 +11859,14 @@ def handle_get(handler, parsed) -> bool:
"passkeys_count": len(passkeys),
"passkey_feature_flag": passkey_flag,
"auth_disabled_acknowledged": bool(load_settings().get("auth_disabled_acknowledged")) if not auth_enabled else False,
})
}
if is_trusted_auth_enabled() or (session_info and session_info.get("auth_type") == "trusted"):
payload["trusted_auth_enabled"] = True
if session_info and session_info.get("auth_type") == "trusted":
payload["auth_type"] = session_info.get("auth_type")
payload["user"] = session_info.get("username")
payload["bound_profile"] = session_info.get("bound_profile")
return j(handler, payload)
if parsed.path.startswith("/api/share/"):
token = parsed.path[len("/api/share/"):].strip()
@@ -15133,10 +15150,17 @@ def handle_post(handler, parsed) -> bool:
if not name:
return bad(handler, "name is required")
try:
from api.auth import ensure_trusted_auth_session
from api.profiles import switch_profile, _validate_profile_name
from api.helpers import build_profile_cookie
if name != 'default':
_validate_profile_name(name)
session_info = ensure_trusted_auth_session(handler)
if getattr(handler, '_trusted_auth_session_rejected', False):
return bad(handler, 'Authentication required', 401)
bound_profile = str((session_info or {}).get("bound_profile") or "").strip() or None
if bound_profile and name != bound_profile:
return bad(handler, "Profile is bound to the current session", 403)
# process_wide=False: don't mutate the process-global _active_profile.
# Per-client profile is managed via cookie + thread-local (#798).
result = switch_profile(name, process_wide=False)
@@ -15150,8 +15174,15 @@ def handle_post(handler, parsed) -> bool:
restart_watcher_for_profile(name)
except Exception as exc:
logger.warning("Failed to restart gateway watcher for profile %s: %s", name, exc)
session_cookie_value = getattr(handler, '_trusted_auth_session_cookie_value', None)
if session_cookie_value:
if bound_profile and name == bound_profile:
return j(handler, result)
extra_header = build_profile_cookie(name, session_cookie_value=session_cookie_value)
else:
extra_header = build_profile_cookie(name, handler)
return j(handler, result, extra_headers={
'Set-Cookie': build_profile_cookie(name, handler),
'Set-Cookie': extra_header,
})
except PermissionError as e:
return bad(handler, _sanitize_error(e), 403)
@@ -16087,18 +16118,26 @@ def handle_post(handler, parsed) -> bool:
return j(handler, {"credentials": registered_credentials()})
if parsed.path == "/api/auth/logout":
from api.auth import clear_auth_cookie, invalidate_session, parse_cookie
from api.auth import clear_auth_cookie, ensure_trusted_auth_session, get_trusted_auth_logout_url, invalidate_session, parse_cookie
from api.helpers import clear_profile_cookie
cookie_val = parse_cookie(handler)
session_info = ensure_trusted_auth_session(handler)
cookie_val = getattr(handler, '_trusted_auth_session_cookie_value', None) or parse_cookie(handler)
if cookie_val:
invalidate_session(cookie_val)
body = json.dumps({"ok": True}).encode()
payload = {"ok": True}
if session_info and session_info.get("auth_type") == "trusted":
logout_url = get_trusted_auth_logout_url()
if logout_url:
payload["trusted_logout_url"] = logout_url
body = json.dumps(payload).encode()
handler.send_response(200)
handler.send_header("Content-Type", "application/json")
handler.send_header("Content-Length", str(len(body)))
handler.send_header("Cache-Control", "no-store")
_security_headers(handler)
clear_auth_cookie(handler)
clear_profile_cookie(handler)
handler.end_headers()
handler.wfile.write(body)
return True
+3 -3
View File
@@ -100,7 +100,7 @@ from urllib.parse import urlparse
logger = logging.getLogger(__name__)
from api.auth import check_auth
from api.auth import check_auth, reset_trusted_auth_request_state
from api.config import HOST, PORT, STATE_DIR, SESSION_DIR, DEFAULT_WORKSPACE
from api.helpers import (
j,
@@ -373,7 +373,7 @@ class Handler(BaseHTTPRequestHandler):
self._safe_webui_print(f'[webui] {record}')
def do_GET(self) -> None:
self._req_t0 = time.time()
self._req_t0 = time.time(); reset_trusted_auth_request_state(self)
cookie_profile = get_profile_cookie(self)
if cookie_profile:
set_request_profile(cookie_profile)
@@ -398,7 +398,7 @@ class Handler(BaseHTTPRequestHandler):
clear_request_profile()
def _handle_write(self, route_func) -> None:
self._req_t0 = time.time()
self._req_t0 = time.time(); reset_trusted_auth_request_state(self)
cookie_profile = get_profile_cookie(self)
if cookie_profile:
set_request_profile(cookie_profile)
+2 -2
View File
@@ -12539,8 +12539,8 @@ async function saveSettings(andClose){
async function signOut(){
try{
await api('/api/auth/logout',{method:'POST',body:'{}'});
window.location.href='login';
const response=await api('/api/auth/logout',{method:'POST',body:'{}'});
window.location.href=response.trusted_logout_url||'login';
}catch(e){
showToast(t('sign_out_failed')+e.message);
}
+801
View File
@@ -0,0 +1,801 @@
from __future__ import annotations
import io
import json
import shutil
import subprocess
import time
from pathlib import Path
from types import SimpleNamespace
import pytest
import api.auth as auth
import api.routes as routes
import api.profiles as profiles
from tests.js_source_extract import extract_function
PANELS_JS = (Path(__file__).resolve().parents[1] / "static" / "panels.js").read_text(encoding="utf-8")
NODE = shutil.which("node")
class _Handler:
def __init__(self, *, headers=None, client_address=("127.0.0.1", 12345)):
self.headers = dict(headers or {})
self.client_address = client_address
self.command = "GET"
self.path = "/"
self.request = SimpleNamespace()
self.rfile = io.BytesIO(b"")
self.wfile = io.BytesIO()
self.status = None
self.sent_headers = []
def send_response(self, status):
self.status = status
def send_header(self, name, value):
self.sent_headers.append((name, value))
def end_headers(self):
pass
def body_bytes(self):
return self.wfile.getvalue()
def body_text(self):
return self.body_bytes().decode("utf-8")
def json_body(self):
return json.loads(self.body_text())
def header_values(self, name):
return [value for key, value in self.sent_headers if key == name]
@pytest.fixture(autouse=True)
def isolated_auth_state(monkeypatch, tmp_path):
monkeypatch.setattr(auth, "STATE_DIR", tmp_path)
monkeypatch.setattr(auth, "_SESSIONS_FILE", tmp_path / ".sessions.json")
monkeypatch.setattr(auth, "is_password_auth_enabled", lambda: False)
monkeypatch.setattr(auth, "are_passkeys_enabled", lambda: False)
monkeypatch.setattr(auth, "is_oidc_auth_enabled", lambda: False)
auth._sessions.clear()
auth._TRUSTED_AUTH_WARNINGS_EMITTED.clear()
profiles.clear_request_profile()
yield
auth._sessions.clear()
auth._TRUSTED_AUTH_WARNINGS_EMITTED.clear()
profiles.clear_request_profile()
def _trusted_env(
monkeypatch,
*,
header="Remote-User",
groups_header=None,
group_map=None,
proxy_cidrs=None,
logout_url=None,
):
for key in (
"HERMES_WEBUI_TRUSTED_AUTH_HEADER",
"HERMES_WEBUI_TRUSTED_GROUPS_HEADER",
"HERMES_WEBUI_GROUP_PROFILE_MAP",
"HERMES_WEBUI_TRUSTED_PROXY_CIDRS",
"HERMES_WEBUI_TRUSTED_AUTH_LOGOUT_URL",
):
monkeypatch.delenv(key, raising=False)
if header is not None:
monkeypatch.setenv("HERMES_WEBUI_TRUSTED_AUTH_HEADER", header)
if groups_header is not None:
monkeypatch.setenv("HERMES_WEBUI_TRUSTED_GROUPS_HEADER", groups_header)
if group_map is not None:
monkeypatch.setenv("HERMES_WEBUI_GROUP_PROFILE_MAP", json.dumps(group_map))
if proxy_cidrs is not None:
monkeypatch.setenv("HERMES_WEBUI_TRUSTED_PROXY_CIDRS", proxy_cidrs)
if logout_url is not None:
monkeypatch.setenv("HERMES_WEBUI_TRUSTED_AUTH_LOGOUT_URL", logout_url)
def test_trusted_header_only_enables_auth_gate(monkeypatch):
_trusted_env(monkeypatch)
assert auth.is_trusted_auth_enabled() is True
assert auth.is_auth_enabled() is True
def test_untrusted_peer_header_does_not_create_session(monkeypatch):
_trusted_env(monkeypatch)
handler = _Handler(
headers={"Remote-User": "alice"},
client_address=("10.0.0.5", 12345),
)
result = auth.check_auth(handler, SimpleNamespace(path="/api/sessions", query=""))
assert result is False
assert handler.status == 401
assert getattr(handler, "_pending_set_cookies", []) == []
def test_malformed_trusted_proxy_cidr_rejects_non_loopback_peer(monkeypatch):
_trusted_env(monkeypatch, proxy_cidrs="bad-cidr")
handler = _Handler(headers={"Remote-User": "alice"}, client_address=("10.0.0.5", 12345))
result = auth.check_auth(handler, SimpleNamespace(path="/api/sessions", query=""))
assert auth.is_trusted_auth_enabled() is True
assert auth.is_auth_enabled() is True
assert result is False
assert handler.status == 401
assert getattr(handler, "_pending_set_cookies", []) == []
def test_malformed_trusted_proxy_cidr_rejects_existing_trusted_session(monkeypatch):
_trusted_env(monkeypatch, proxy_cidrs="bad-cidr")
cookie = auth.create_session(auth_type="trusted", username="alice")
handler = _Handler(
headers={"Cookie": f"hermes_session={cookie}", "Remote-User": "alice"},
client_address=("10.0.0.5", 12345),
)
result = auth.check_auth(handler, SimpleNamespace(path="/api/sessions", query=""))
assert auth.is_trusted_auth_enabled() is True
assert auth.is_auth_enabled() is True
assert result is False
assert handler.status == 401
assert handler.body_text() == '{"error":"Authentication required"}'
assert auth.verify_session(cookie) is False
def test_invalid_trusted_header_name_fails_closed(monkeypatch):
_trusted_env(monkeypatch, header="Bad Header")
handler = _Handler(headers={"Bad Header": "alice"})
result = auth.check_auth(handler, SimpleNamespace(path="/api/sessions", query=""))
assert auth.is_trusted_auth_enabled() is True
assert auth.is_auth_enabled() is True
assert result is False
assert handler.status == 401
def test_allowlisted_peer_header_creates_trusted_session(monkeypatch):
_trusted_env(monkeypatch)
handler = _Handler(headers={"Remote-User": "alice"})
result = auth.check_auth(handler, SimpleNamespace(path="/api/sessions", query=""))
assert result is True
assert handler.status is None
pending = getattr(handler, "_pending_set_cookies", [])
assert any(cookie.startswith("hermes_session=") for cookie in pending)
assert not any(cookie.startswith("hermes_profile=") for cookie in pending)
def test_group_map_binds_profile(monkeypatch):
_trusted_env(
monkeypatch,
groups_header="Remote-Groups",
group_map={"hermes_devops": "devops"},
)
handler = _Handler(
headers={
"Remote-User": "alice",
"Remote-Groups": "hermes_devops,ai_users",
}
)
info = auth.ensure_trusted_auth_session(handler)
assert info["auth_type"] == "trusted"
assert info["username"] == "alice"
assert info["bound_profile"] == "devops"
cookie_value = handler._trusted_auth_session_cookie_value
assert auth.session_bound_profile(cookie_value) == "devops"
assert any(cookie.startswith("hermes_profile=") for cookie in handler._pending_set_cookies)
def test_group_map_prefers_mapping_order_over_header_order(monkeypatch):
_trusted_env(
monkeypatch,
groups_header="Remote-Groups",
group_map={"admins": "ops", "devs": "sandbox"},
)
first = _Handler(headers={"Remote-User": "alice", "Remote-Groups": "devs,admins"})
second = _Handler(headers={"Remote-User": "bob", "Remote-Groups": "admins,devs"})
assert auth.ensure_trusted_auth_session(first)["bound_profile"] == "ops"
assert auth.ensure_trusted_auth_session(second)["bound_profile"] == "ops"
@pytest.mark.parametrize(
"group_map",
[
{"ops": "ops_profile", "": "admin"},
{"": "admin", "ops": "ops_profile"},
],
)
def test_group_map_ignores_invalid_entry_without_discarding_valid_mappings(monkeypatch, group_map):
_trusted_env(
monkeypatch,
groups_header="Remote-Groups",
group_map=group_map,
)
handler = _Handler(headers={"Remote-User": "alice", "Remote-Groups": "ops"})
assert auth.ensure_trusted_auth_session(handler)["bound_profile"] == "ops_profile"
def test_group_map_without_match_binds_default(monkeypatch):
_trusted_env(
monkeypatch,
groups_header="Remote-Groups",
group_map={"hermes_devops": "devops"},
)
handler = _Handler(
headers={
"Remote-User": "alice",
"Remote-Groups": "ai_users,ops",
}
)
info = auth.ensure_trusted_auth_session(handler)
assert info["bound_profile"] == "default"
assert auth.session_bound_profile(handler._trusted_auth_session_cookie_value) == "default"
def test_bound_profile_mismatch_rejected(monkeypatch):
_trusted_env(monkeypatch, groups_header="Remote-Groups", group_map={"hermes_devops": "devops"})
cookie = auth.create_session(
auth_type="trusted",
username="alice",
bound_profile="devops",
)
handler = _Handler(headers={"Cookie": f"hermes_session={cookie}", "Remote-User": "alice", "Remote-Groups": "hermes_devops"})
monkeypatch.setattr("api.profiles.get_active_profile_name", lambda: "coworkers")
result = auth.check_auth(handler, SimpleNamespace(path="/api/sessions", query=""))
assert result is False
assert handler.status == 403
assert handler.body_text() == '{"error":"Profile access forbidden"}'
def test_bound_profile_match_allowed(monkeypatch):
_trusted_env(monkeypatch, groups_header="Remote-Groups", group_map={"hermes_devops": "devops"})
cookie = auth.create_session(
auth_type="trusted",
username="alice",
bound_profile="devops",
)
handler = _Handler(headers={"Cookie": f"hermes_session={cookie}", "Remote-User": "alice", "Remote-Groups": "hermes_devops"})
monkeypatch.setattr("api.profiles.get_active_profile_name", lambda: "devops")
result = auth.check_auth(handler, SimpleNamespace(path="/api/sessions", query=""))
assert result is True
assert handler.status is None
def test_profile_switch_rejects_other_bound_profile(monkeypatch):
cookie = auth.create_session(
auth_type="trusted",
username="alice",
bound_profile="devops",
)
_trusted_env(monkeypatch, groups_header="Remote-Groups", group_map={"hermes_devops": "devops"})
handler = _Handler(headers={"Cookie": f"hermes_session={cookie}", "Remote-User": "alice", "Remote-Groups": "hermes_devops"})
handler.command = "POST"
monkeypatch.setattr("api.profiles.get_active_profile_name", lambda: "devops")
monkeypatch.setattr(routes, "_check_csrf", lambda _handler: True)
monkeypatch.setattr(routes, "read_body", lambda _handler: {"name": "coworkers"})
monkeypatch.setattr("api.profiles.switch_profile", lambda *_, **__: (_ for _ in ()).throw(AssertionError("should not switch")))
routes.handle_post(handler, SimpleNamespace(path="/api/profile/switch", query=""))
assert handler.status == 403
assert handler.json_body()["error"] == "Profile is bound to the current session"
def test_first_trusted_profile_switch_rejection_keeps_session_cookies(monkeypatch):
_trusted_env(
monkeypatch,
groups_header="Remote-Groups",
group_map={"hermes_devops": "devops"},
)
handler = _Handler(
headers={
"Remote-User": "alice",
"Remote-Groups": "hermes_devops",
}
)
handler.command = "POST"
monkeypatch.setattr("api.profiles.get_active_profile_name", lambda: "devops")
monkeypatch.setattr(routes, "_check_csrf", lambda _handler: True)
monkeypatch.setattr(routes, "read_body", lambda _handler: {"name": "coworkers"})
monkeypatch.setattr("api.profiles.switch_profile", lambda *_, **__: (_ for _ in ()).throw(AssertionError("should not switch")))
assert auth.check_auth(handler, SimpleNamespace(path="/api/profile/switch", query="")) is True
routes.handle_post(handler, SimpleNamespace(path="/api/profile/switch", query=""))
set_cookies = handler.header_values("Set-Cookie")
assert handler.status == 403
assert handler.json_body()["error"] == "Profile is bound to the current session"
assert any(cookie.startswith("hermes_session=") for cookie in set_cookies)
assert any(cookie.startswith("hermes_profile=") for cookie in set_cookies)
assert len([cookie for cookie in set_cookies if cookie.startswith("hermes_session=")]) == 1
def test_profile_switch_accepts_bound_profile(monkeypatch):
_trusted_env(monkeypatch, groups_header="Remote-Groups", group_map={"hermes_devops": "devops"})
cookie = auth.create_session(
auth_type="trusted",
username="alice",
bound_profile="devops",
)
handler = _Handler(headers={"Cookie": f"hermes_session={cookie}", "Remote-User": "alice", "Remote-Groups": "hermes_devops"})
handler.command = "POST"
monkeypatch.setattr("api.profiles.get_active_profile_name", lambda: "devops")
monkeypatch.setattr(routes, "_check_csrf", lambda _handler: True)
monkeypatch.setattr(routes, "read_body", lambda _handler: {"name": "devops"})
monkeypatch.setattr("api.profiles.switch_profile", lambda name, process_wide=False: {"ok": True, "profile": name})
monkeypatch.setattr("api.config.invalidate_models_cache", lambda: None)
monkeypatch.setattr("api.gateway_watcher.restart_watcher_for_profile", lambda _name: None)
routes.handle_post(handler, SimpleNamespace(path="/api/profile/switch", query=""))
assert handler.status == 200
assert handler.json_body()["profile"] == "devops"
assert any(value.startswith("hermes_profile=") for value in handler.header_values("Set-Cookie"))
def test_auth_status_reports_trusted_session_fields(monkeypatch):
_trusted_env(monkeypatch, groups_header="Remote-Groups", group_map={"hermes_devops": "devops"})
cookie = auth.create_session(
auth_type="trusted",
username="alice",
bound_profile="devops",
)
handler = _Handler(headers={"Cookie": f"hermes_session={cookie}", "Remote-User": "alice", "Remote-Groups": "hermes_devops"})
monkeypatch.setattr(auth, "_passkey_feature_flag_enabled", lambda: False)
monkeypatch.setattr("api.passkeys.registered_credentials", lambda: [])
routes.handle_get(handler, SimpleNamespace(path="/api/auth/status", query=""))
payload = handler.json_body()
assert payload["auth_enabled"] is True
assert payload["logged_in"] is True
assert payload["trusted_auth_enabled"] is True
assert payload["auth_type"] == "trusted"
assert payload["user"] == "alice"
assert payload["bound_profile"] == "devops"
def test_auth_status_rejects_trusted_cookie_when_proxy_cidr_is_malformed(monkeypatch):
_trusted_env(monkeypatch, proxy_cidrs="bad-cidr")
cookie = auth.create_session(
auth_type="trusted",
username="alice",
bound_profile="devops",
)
handler = _Handler(
headers={"Cookie": f"hermes_session={cookie}", "Remote-User": "alice"},
client_address=("10.0.0.5", 12345),
)
monkeypatch.setattr(auth, "_passkey_feature_flag_enabled", lambda: False)
monkeypatch.setattr("api.passkeys.registered_credentials", lambda: [])
routes.handle_get(handler, SimpleNamespace(path="/api/auth/status", query=""))
payload = handler.json_body()
assert payload["auth_enabled"] is True
assert payload["logged_in"] is False
assert payload["trusted_auth_enabled"] is True
assert "auth_type" not in payload
assert "user" not in payload
assert "bound_profile" not in payload
assert auth.verify_session(cookie) is False
def test_malformed_trusted_proxy_cidr_keeps_loopback_trusted(monkeypatch):
_trusted_env(monkeypatch, proxy_cidrs="bad-cidr")
handler = _Handler(headers={"Remote-User": "alice"})
assert auth.check_auth(handler, SimpleNamespace(path="/api/sessions", query="")) is True
def test_mapped_ipv6_peer_matches_canonical_trusted_proxy_cidr(monkeypatch):
_trusted_env(monkeypatch, proxy_cidrs="10.0.0.0/8")
handler = _Handler(
headers={"Remote-User": "alice"},
client_address=("::ffff:10.0.0.5", 12345),
)
assert auth.check_auth(handler, SimpleNamespace(path="/api/sessions", query="")) is True
assert handler._trusted_auth_session_info["username"] == "alice"
def test_existing_trusted_session_rotates_for_current_identity(monkeypatch):
_trusted_env(
monkeypatch,
groups_header="Remote-Groups",
group_map={"alice-group": "alice", "bob-group": "bob"},
)
cookie = auth.create_session(
auth_type="trusted",
username="alice",
bound_profile="alice",
)
handler = _Handler(
headers={
"Cookie": f"hermes_session={cookie}",
"Remote-User": "bob",
"Remote-Groups": "bob-group",
}
)
assert auth.check_auth(handler, SimpleNamespace(path="/api/sessions", query="")) is True
assert auth.verify_session(cookie) is False
assert handler._trusted_auth_session_info["username"] == "bob"
assert handler._trusted_auth_session_info["bound_profile"] == "bob"
assert handler._trusted_auth_session_cookie_value != cookie
assert profiles.get_active_profile_name() == "bob"
def test_trusted_reconciliation_cache_resets_between_requests(monkeypatch):
_trusted_env(monkeypatch)
handler = _Handler(headers={"Remote-User": "alice"})
assert auth.check_auth(handler, SimpleNamespace(path="/api/sessions", query="")) is True
alice_cookie = handler._trusted_auth_session_cookie_value
handler.headers = {"Cookie": f"hermes_session={alice_cookie}", "Remote-User": "bob"}
handler._pending_set_cookies = []
auth.reset_trusted_auth_request_state(handler)
assert auth.check_auth(handler, SimpleNamespace(path="/api/sessions", query="")) is True
assert auth.verify_session(alice_cookie) is False
assert handler._trusted_auth_session_info["username"] == "bob"
def test_server_resets_trusted_auth_request_state_per_request():
server_source = (Path(__file__).resolve().parents[1] / "server.py").read_text(encoding="utf-8")
assert server_source.count("reset_trusted_auth_request_state(self)") == 2
def test_reset_clears_pending_cookies_across_keepalive_requests(monkeypatch):
# A queued-but-unflushed Set-Cookie must NOT survive onto the next request on
# a reused HTTP/1.1 keep-alive handler (regression: a stale trusted-auth
# cookie leaking across the request boundary could overwrite a later valid
# login cookie and 401 the user). reset_trusted_auth_request_state() runs at
# the per-request entry (server.py do_GET/do_POST), so it must drop the queue.
_trusted_env(monkeypatch)
handler = _Handler()
# Request N queues an auth cookie but the response is never flushed.
auth._queue_pending_cookie(handler, "hermes_session=stale-value; Path=/")
assert handler._pending_set_cookies == ["hermes_session=stale-value; Path=/"]
# Request N+1 begins on the same reused handler.
auth.reset_trusted_auth_request_state(handler)
# The stale queued cookie must be gone — nothing to flush into N+1's response.
assert getattr(handler, "_pending_set_cookies", []) == []
from api.helpers import flush_pending_auth_cookies
flush_pending_auth_cookies(handler)
assert handler.header_values("Set-Cookie") == []
def test_auth_status_reports_reconciled_trusted_identity(monkeypatch):
_trusted_env(
monkeypatch,
groups_header="Remote-Groups",
group_map={"alice-group": "alice", "bob-group": "bob"},
)
cookie = auth.create_session(
auth_type="trusted",
username="alice",
bound_profile="alice",
)
handler = _Handler(
headers={
"Cookie": f"hermes_session={cookie}",
"Remote-User": "bob",
"Remote-Groups": "bob-group",
}
)
monkeypatch.setattr(auth, "_passkey_feature_flag_enabled", lambda: False)
monkeypatch.setattr("api.passkeys.registered_credentials", lambda: [])
routes.handle_get(handler, SimpleNamespace(path="/api/auth/status", query=""))
payload = handler.json_body()
assert payload["logged_in"] is True
assert payload["user"] == "bob"
assert payload["bound_profile"] == "bob"
assert auth.verify_session(cookie) is False
def test_existing_trusted_session_without_header_is_invalidated(monkeypatch):
_trusted_env(monkeypatch)
cookie = auth.create_session(auth_type="trusted", username="alice")
handler = _Handler(headers={"Cookie": f"hermes_session={cookie}"})
assert auth.check_auth(handler, SimpleNamespace(path="/api/sessions", query="")) is False
assert handler.status == 401
assert auth.verify_session(cookie) is False
def test_untrusted_existing_trusted_session_is_rejected_by_all_consumers(monkeypatch):
_trusted_env(monkeypatch)
headers = {"Remote-User": "alice"}
protected_cookie = auth.create_session(auth_type="trusted", username="alice")
protected = _Handler(
headers={**headers, "Cookie": f"hermes_session={protected_cookie}"},
client_address=("10.0.0.5", 12345),
)
assert auth.check_auth(protected, SimpleNamespace(path="/api/sessions", query="")) is False
assert protected.status == 401
assert auth.verify_session(protected_cookie) is False
status_cookie = auth.create_session(auth_type="trusted", username="alice")
status = _Handler(
headers={**headers, "Cookie": f"hermes_session={status_cookie}"},
client_address=("10.0.0.5", 12345),
)
monkeypatch.setattr(auth, "_passkey_feature_flag_enabled", lambda: False)
monkeypatch.setattr("api.passkeys.registered_credentials", lambda: [])
routes.handle_get(status, SimpleNamespace(path="/api/auth/status", query=""))
assert status.json_body()["logged_in"] is False
assert auth.verify_session(status_cookie) is False
switch_cookie = auth.create_session(auth_type="trusted", username="alice")
switch = _Handler(
headers={**headers, "Cookie": f"hermes_session={switch_cookie}"},
client_address=("10.0.0.5", 12345),
)
switch.command = "POST"
monkeypatch.setattr(routes, "_check_csrf", lambda _handler: True)
monkeypatch.setattr(routes, "read_body", lambda _handler: {"name": "default"})
monkeypatch.setattr(
"api.profiles.switch_profile",
lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("should not switch")),
)
routes.handle_post(switch, SimpleNamespace(path="/api/profile/switch", query=""))
assert switch.status == 401
assert auth.verify_session(switch_cookie) is False
def test_trusted_session_rehydrates_bound_profile_cookie(monkeypatch):
_trusted_env(monkeypatch, groups_header="Remote-Groups", group_map={"hermes_devops": "devops"})
cookie = auth.create_session(
auth_type="trusted",
username="alice",
bound_profile="devops",
)
handler = _Handler(headers={"Cookie": f"hermes_session={cookie}", "Remote-User": "alice", "Remote-Groups": "hermes_devops"})
assert auth.check_auth(handler, SimpleNamespace(path="/api/sessions", query="")) is True
assert profiles.get_active_profile_name() == "devops"
profile_cookie = next(
cookie_header for cookie_header in handler._pending_set_cookies if cookie_header.startswith("hermes_profile=")
)
profile_value = profile_cookie.split("=", 1)[1].split(";", 1)[0]
assert auth.verify_profile_cookie_value(profile_value, cookie) == "devops"
def test_first_trusted_shell_response_includes_csrf_token(monkeypatch):
_trusted_env(monkeypatch)
handler = _Handler(headers={"Remote-User": "alice"})
monkeypatch.setattr(routes, "_render_index_shell_base", lambda: "csrfToken:__CSRF_TOKEN_JSON__")
monkeypatch.setattr("api.extensions.inject_extension_tags", lambda html: html)
assert auth.check_auth(handler, SimpleNamespace(path="/", query="")) is True
routes.handle_get(handler, SimpleNamespace(path="/", query=""))
cookie_value = handler._trusted_auth_session_cookie_value
assert any(cookie.startswith("hermes_session=") for cookie in handler.header_values("Set-Cookie"))
assert handler.body_text() == f"csrfToken:{json.dumps(auth.csrf_token_for_session(cookie_value))}"
def test_logout_clears_auth_and_profile_cookies(monkeypatch):
_trusted_env(
monkeypatch,
groups_header="Remote-Groups",
group_map={"hermes_devops": "devops"},
logout_url="https://auth.example.com/logout",
)
cookie = auth.create_session(
auth_type="trusted",
username="alice",
bound_profile="devops",
)
handler = _Handler(headers={"Cookie": f"hermes_session={cookie}", "Remote-User": "alice", "Remote-Groups": "hermes_devops"})
handler.command = "POST"
monkeypatch.setattr("api.profiles.get_active_profile_name", lambda: "devops")
monkeypatch.setattr(routes, "_check_csrf", lambda _handler: True)
monkeypatch.setattr(routes, "read_body", lambda _handler: {})
routes.handle_post(handler, SimpleNamespace(path="/api/auth/logout", query=""))
payload = handler.json_body()
assert payload["ok"] is True
assert payload["trusted_logout_url"] == "https://auth.example.com/logout"
set_cookies = handler.header_values("Set-Cookie")
assert any(cookie.startswith("hermes_session=") and "Max-Age=0" in cookie for cookie in set_cookies)
assert any(cookie.startswith("hermes_profile=") and "Max-Age=0" in cookie and "SameSite=Lax" in cookie for cookie in set_cookies)
assert auth.verify_session(cookie) is False
def test_logout_identity_rotation_preserves_csrf_validation(monkeypatch):
_trusted_env(
monkeypatch,
groups_header="Remote-Groups",
group_map={"alice-group": "alice", "bob-group": "bob"},
logout_url="https://auth.example.com/logout",
)
cookie = auth.create_session(
auth_type="trusted",
username="alice",
bound_profile="alice",
)
handler = _Handler(
headers={
"Cookie": f"hermes_session={cookie}",
"Remote-User": "bob",
"Remote-Groups": "bob-group",
auth.CSRF_HEADER_NAME: auth.csrf_token_for_session(cookie),
}
)
handler.command = "POST"
monkeypatch.setattr(routes, "read_body", lambda _handler: {})
assert auth.check_auth(handler, SimpleNamespace(path="/api/auth/logout", query="")) is True
routes.handle_post(handler, SimpleNamespace(path="/api/auth/logout", query=""))
payload = handler.json_body()
assert payload["ok"] is True
assert payload["trusted_logout_url"] == "https://auth.example.com/logout"
assert handler._trusted_auth_session_info["username"] == "bob"
assert handler._trusted_auth_session_cookie_value != cookie
assert auth.verify_session(cookie) is False
assert auth.verify_session(handler._trusted_auth_session_cookie_value) is False
set_cookies = handler.header_values("Set-Cookie")
assert any(cookie_header.startswith("hermes_session=") and "Max-Age=0" in cookie_header for cookie_header in set_cookies)
assert any(cookie_header.startswith("hermes_profile=") and "Max-Age=0" in cookie_header for cookie_header in set_cookies)
def test_unconfigured_remote_user_header_is_ordinary_header(monkeypatch):
_trusted_env(monkeypatch, header=None)
handler = _Handler(headers={"Remote-User": "alice"})
assert auth.is_trusted_auth_enabled() is False
assert auth.ensure_trusted_auth_session(handler) is None
assert auth.check_auth(handler, SimpleNamespace(path="/api/sessions", query="")) is True
def test_trusted_auth_owner_contract(monkeypatch):
_trusted_env(
monkeypatch,
groups_header="Remote-Groups",
group_map={"hermes_devops": "devops"},
)
handler = _Handler(
headers={
"Remote-User": "alice",
"Remote-Groups": "hermes_devops,ai_users",
}
)
info = auth.ensure_trusted_auth_session(handler)
assert info["auth_type"] == "trusted"
assert info["bound_profile"] == "devops"
assert auth.is_trusted_auth_enabled() is True
token = "deadbeef" * 8
auth._sessions[token] = time.time() + 3600
legacy_sig = auth.hmac.new(
auth._signing_key(),
token.encode(),
auth.hashlib.sha256,
).hexdigest()
legacy_cookie = f"{token}.{legacy_sig}"
legacy_info = auth.get_session_info(legacy_cookie)
assert legacy_info["auth_type"] is None
assert legacy_info["bound_profile"] is None
assert legacy_info["expiry"] > time.time()
def test_consumers_route_through_auth_owner(monkeypatch):
_trusted_env(
monkeypatch,
groups_header="Remote-Groups",
group_map={"hermes_devops": "devops"},
logout_url="https://auth.example.com/logout",
)
cookie = auth.create_session(
auth_type="trusted",
username="alice",
bound_profile="devops",
)
handler = _Handler(headers={"Cookie": f"hermes_session={cookie}"})
handler.command = "POST"
calls = []
def _get_session_info(cookie_value):
calls.append(("get_session_info", cookie_value))
return {
"token": cookie_value.split(".", 1)[0],
"expiry": time.time() + 3600,
"auth_type": "trusted",
"username": "alice",
"bound_profile": "devops",
}
monkeypatch.setattr(auth, "get_session_info", _get_session_info)
monkeypatch.setattr(auth, "ensure_trusted_auth_session", lambda _handler: calls.append(("ensure", None)) or {
"token": "token",
"expiry": time.time() + 3600,
"auth_type": "trusted",
"username": "alice",
"bound_profile": "devops",
})
monkeypatch.setattr(auth, "parse_cookie", lambda _handler: cookie)
monkeypatch.setattr(auth, "verify_session", lambda _cookie: True)
monkeypatch.setattr(auth, "is_auth_enabled", lambda: True)
monkeypatch.setattr(auth, "_passkey_feature_flag_enabled", lambda: False)
monkeypatch.setattr("api.passkeys.registered_credentials", lambda: [])
monkeypatch.setattr("api.profiles.get_active_profile_name", lambda: "devops")
monkeypatch.setattr(routes, "_check_csrf", lambda _handler: True)
monkeypatch.setattr(routes, "read_body", lambda _handler: {"name": "devops"})
monkeypatch.setattr("api.profiles.switch_profile", lambda name, process_wide=False: {"ok": True, "profile": name})
monkeypatch.setattr("api.config.invalidate_models_cache", lambda: None)
monkeypatch.setattr("api.gateway_watcher.restart_watcher_for_profile", lambda _name: None)
routes.handle_get(handler, SimpleNamespace(path="/api/auth/status", query=""))
assert calls and calls[0][0] == "ensure"
assert handler.json_body()["bound_profile"] == "devops"
calls.clear()
routes.handle_post(handler, SimpleNamespace(path="/api/profile/switch", query=""))
assert calls and calls[0][0] == "ensure"
assert handler.status == 200
def test_sign_out_uses_trusted_logout_url_with_login_fallback():
sign_out = extract_function(PANELS_JS, "signOut", prefix="async function")
assert "const response=await api('/api/auth/logout',{method:'POST',body:'{}'});" in sign_out
assert "window.location.href=response.trusted_logout_url||'login';" in sign_out
assert NODE is not None
def run_sign_out(logout_url):
script = f"""
const signOut = (0, eval)("(" + {json.dumps(sign_out)} + ")");
globalThis.api = async () => ({{trusted_logout_url: {json.dumps(logout_url)}}});
globalThis.window = {{location: {{href: null}}}};
globalThis.showToast = () => {{}};
globalThis.t = (key) => key;
signOut().then(() => process.stdout.write(JSON.stringify(window.location.href)));
"""
creationflags = getattr(subprocess, "CREATE_NO_WINDOW", 0)
result = subprocess.run(
[NODE, "-e", script],
capture_output=True,
text=True,
timeout=30,
creationflags=creationflags,
)
assert result.returncode == 0, result.stderr
return json.loads(result.stdout)
assert run_sign_out("https://auth.example.com/logout") == "https://auth.example.com/logout"
assert run_sign_out(None) == "login"