"""Hermes Web UI -- optional authentication.Off by default. Enable by setting HERMES_WEBUI_PASSWORD, configuring apassword in Settings, registering passkeys, or configuring native OIDC SSO."""import hashlibimport hmacimport http.cookiesimport jsonimport loggingimport osimport reimport secretsimport tempfileimport threadingimport timefrom pathlib import Path from api.config import STATE_DIR, get_config, load_settings logger = logging.getLogger(__name__)  # Default session TTL — 30 days. Kept as a module-level constant for backwards# compatibility with downstream code and regression tests that import it.# At runtime, prefer ``_resolve_session_ttl()`` which honours the env var and# settings.json overrides; this constant is the floor / fallback.SESSION_TTL = 86400 * 30  # 30 days  def _resolve_session_ttl() -> int:    """Resolve session TTL from env > settings > default.     Priority mirrors get_password_hash(): HERMES_WEBUI_SESSION_TTL env var    first, then settings.json, falling back to ``SESSION_TTL`` (30 days).    Clamped to [60s, 1 year] to prevent runaway cookies or self-lockout.    """    env_v = os.getenv('HERMES_WEBUI_SESSION_TTL', '').strip()    if env_v.isdigit():        val = int(env_v)        if 60 <= val <= 86400 * 365:            return val    s = load_settings()    v = s.get('session_ttl_seconds')    if isinstance(v, int) and 60 <= v <= 86400 * 365:        return v    return SESSION_TTL  # ── Public paths (no auth required) ─────────────────────────────────────────PUBLIC_PATHS = frozenset({    '/login', '/health', '/favicon.ico', '/sw.js',    '/api/auth/login', '/api/auth/status',    '/api/auth/oidc/start', '/api/auth/oidc/callback',    '/api/auth/passkey/options', '/api/auth/passkey/login',    '/share',    '/manifest.json', '/manifest.webmanifest',    '/session/manifest.json', '/session/manifest.webmanifest',}) COOKIE_NAME = 'hermes_session'CSRF_HEADER_NAME = 'X-Hermes-CSRF-Token'  # RFC 6265 cookie-name token: a non-empty run of token chars# (no controls, whitespace, or separators such as ';', '=', ',')._COOKIE_NAME_RE = re.compile(r"^[-!#$%&'*+.^_`|~0-9A-Za-z]+$")  def _resolve_cookie_name() -> str:    """Resolve the auth session cookie name from env > default.     Honours ``HERMES_WEBUI_COOKIE_NAME`` so multiple WebUI instances sharing a    hostname (different ports) can use distinct cookie names instead of    trampling each other's session — browsers scope cookies by host, not    host+port (RFC 6265). Falls back to ``COOKIE_NAME`` when the env var is    unset, empty, or not a valid RFC 6265 token.    """    name = os.getenv('HERMES_WEBUI_COOKIE_NAME', '').strip()    if not name:        return COOKIE_NAME    if _COOKIE_NAME_RE.match(name):        return name    logger.warning(        'Ignoring invalid HERMES_WEBUI_COOKIE_NAME=%r; falling back to %r '        '(name must be a valid RFC 6265 token)', name, COOKIE_NAME,    )    return COOKIE_NAME  def _warn_auth_persistence_failure(prefix: str, artifact: Path, exc: Exception, consequence: str) -> None:    logger.warning(        '%s at %s (STATE_DIR=%s): %s: %s; %s',        prefix,        artifact,        STATE_DIR,        exc.__class__.__name__,        exc,        consequence,    )  _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 _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    blocked by a corrupt or missing sessions file.    """    try:        if not _SESSIONS_FILE.exists():            return {}        raw = _SESSIONS_FILE.read_text(encoding='utf-8')        data = json.loads(raw)        if not isinstance(data, dict):            raise ValueError('malformed sessions file: expected dict')    except OSError as e:        _warn_auth_persistence_failure(            'Auth session store read failed',            _SESSIONS_FILE,            e,            'starting fresh with an empty session table',        )        return {}    except (UnicodeDecodeError, json.JSONDecodeError) as e:        _warn_auth_persistence_failure(            'Ignoring malformed auth session store',            _SESSIONS_FILE,            e,            'starting fresh with an empty session table',        )        return {}    except Exception as e:        _warn_auth_persistence_failure(            'Ignoring malformed auth session store',            _SESSIONS_FILE,            e,            'starting fresh with an empty session table',        )        return {}    now = time.time()    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 | 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    truncated file.  Mirrors the same pattern as .signing_key persistence.    """    try:        STATE_DIR.mkdir(parents=True, exist_ok=True)        fd, tmp = tempfile.mkstemp(dir=STATE_DIR, suffix='.sessions.tmp')        try:            with os.fdopen(fd, 'w', encoding='utf-8') as f:                json.dump(sessions, f)            os.chmod(tmp, 0o600)            os.replace(tmp, _SESSIONS_FILE)        except Exception:            try:                os.unlink(tmp)            except OSError:                pass            raise    except Exception as e:        _warn_auth_persistence_failure(            'Auth session persistence failed',            _SESSIONS_FILE,            e,            'keeping the in-process session table available',        )  # Active sessions: token -> expiry timestamp (persisted across restarts via STATE_DIR)_sessions = _load_sessions()_SESSIONS_LOCK = threading.Lock() # ── Login rate limiter ──────────────────────────────────────────────────────_LOGIN_ATTEMPTS_FILE = STATE_DIR / '.login_attempts.json'_LOGIN_MAX_ATTEMPTS = 5_LOGIN_WINDOW = 60  # seconds  def _load_login_attempts() -> dict[str, list[float]]:    """Load persisted login attempts from STATE_DIR, pruning expired entries."""    try:        if _LOGIN_ATTEMPTS_FILE.exists():            data = json.loads(_LOGIN_ATTEMPTS_FILE.read_text(encoding='utf-8'))            if not isinstance(data, dict):                raise ValueError('malformed login-attempts file — expected dict')            now = time.time()            attempts: dict[str, list[float]] = {}            for ip, raw_times in data.items():                if not isinstance(ip, str) or not isinstance(raw_times, list):                    continue                fresh = [                    float(t)                    for t in raw_times                    if isinstance(t, (int, float)) and now - float(t) < _LOGIN_WINDOW                ]                if fresh:                    attempts[ip] = fresh            return attempts    except Exception as e:        logger.debug("Failed to load login attempts file, starting fresh: %s", e)    return {}  def _save_login_attempts(attempts: dict[str, list[float]]) -> None:    """Atomically persist login attempts to STATE_DIR/.login_attempts.json (0600)."""    try:        _LOGIN_ATTEMPTS_FILE.parent.mkdir(parents=True, exist_ok=True)        fd, tmp = tempfile.mkstemp(dir=_LOGIN_ATTEMPTS_FILE.parent, suffix='.login_attempts.tmp')        try:            with os.fdopen(fd, 'w', encoding='utf-8') as f:                json.dump(attempts, f)            os.chmod(tmp, 0o600)            os.replace(tmp, _LOGIN_ATTEMPTS_FILE)        except Exception:            try:                os.unlink(tmp)            except OSError:                pass            raise    except Exception as e:        logger.debug("Failed to persist login attempts: %s", e)  _login_attempts = _load_login_attempts()  # ip -> [timestamp, ...]_LOGIN_ATTEMPTS_LOCK = threading.Lock()  def _check_login_rate(ip: str) -> bool:    """Return True if the IP is allowed to attempt login (thread-safe)."""    with _LOGIN_ATTEMPTS_LOCK:        now = time.time()        attempts = _login_attempts.get(ip, [])        # Prune old attempts        attempts = [t for t in attempts if now - t < _LOGIN_WINDOW]        if attempts:            _login_attempts[ip] = attempts        else:            _login_attempts.pop(ip, None)        _save_login_attempts(_login_attempts)        return len(attempts) < _LOGIN_MAX_ATTEMPTS  def _record_login_attempt(ip: str) -> None:    """Record a login attempt for rate limiting (thread-safe)."""    with _LOGIN_ATTEMPTS_LOCK:        now = time.time()        attempts = _login_attempts.get(ip, [])        attempts.append(now)        _login_attempts[ip] = attempts        _save_login_attempts(_login_attempts)  def _clear_login_attempts(ip: str) -> None:    """Clear failed login attempts after a successful login (thread-safe)."""    with _LOGIN_ATTEMPTS_LOCK:        if ip in _login_attempts:            _login_attempts.pop(ip, None)            _save_login_attempts(_login_attempts)  def _load_key(filename: str) -> bytes:    """Load a 32-byte key from STATE_DIR, generating and persisting one if missing."""    key_file = STATE_DIR / filename    try:        if key_file.exists():            raw = key_file.read_bytes()            if len(raw) >= 32:                return raw[:32]    except OSError as e:        _warn_auth_persistence_failure(            'Auth key read failed',            key_file,            e,            'generating a new key and continuing',        )    except Exception as e:        _warn_auth_persistence_failure(            'Auth key read failed',            key_file,            e,            'generating a new key and continuing',        )    key = secrets.token_bytes(32)    try:        STATE_DIR.mkdir(parents=True, exist_ok=True)        key_file.write_bytes(key)        key_file.chmod(0o600)    except OSError as e:        _warn_auth_persistence_failure(            'Auth key persistence failed',            key_file,            e,            'returning the generated key so startup can continue',        )    except Exception as e:        _warn_auth_persistence_failure(            'Auth key persistence failed',            key_file,            e,            'returning the generated key so startup can continue',        )    return key  _PBKDF2_KEY_CACHE: bytes | None = None_SIGNING_KEY_CACHE: bytes | None = None  def _pbkdf2_key() -> bytes:    global _PBKDF2_KEY_CACHE    if _PBKDF2_KEY_CACHE is None:        _PBKDF2_KEY_CACHE = _load_key('.pbkdf2_key')    return _PBKDF2_KEY_CACHE  def _signing_key() -> bytes:    global _SIGNING_KEY_CACHE    if _SIGNING_KEY_CACHE is None:        _SIGNING_KEY_CACHE = _load_key('.signing_key')    return _SIGNING_KEY_CACHE  def _hash_password(password, *, salt: bytes | None = None) -> str:    """PBKDF2-SHA256 with 600k iterations (OWASP recommendation).    Salt is the persisted PBKDF2 key, which is secret and unique per    installation. This keeps the stored hash format a plain hex string    (no format change to settings.json) while replacing the predictable    STATE_DIR-derived salt from the original implementation.     The *salt* parameter exists solely to support transparent migration    of password hashes that were computed with a different key (e.g. the    old `.signing_key`). Normal callers should never pass it.    """    if salt is None:        salt = _pbkdf2_key()    dk = hashlib.pbkdf2_hmac('sha256', password.encode(), salt, 600_000)    return dk.hex()  _AUTH_HASH_LOCK = threading.Lock()_AUTH_HASH_COMPUTED: bool = False_AUTH_HASH_CACHE: str | None = None  def _invalidate_password_hash_cache() -> None:    """Invalidate the in-process password hash cache so the next call to    get_password_hash() re-reads from settings.json or the env var."""    global _AUTH_HASH_COMPUTED, _AUTH_HASH_CACHE    with _AUTH_HASH_LOCK:        _AUTH_HASH_COMPUTED = False        _AUTH_HASH_CACHE = None  def get_password_hash() -> str | None:    """Return the active password hash, or None if auth is disabled.    Priority: env var > settings.json.     The hash is computed once and cached for the lifetime of the process.    PBKDF2-600k takes ~1 s and is called on nearly every HTTP request via    check_auth → is_auth_enabled, so caching avoids wasting a full second    of CPU per request after the first one.     Thread-safe: double-checked locking ensures that under a burst of    concurrent requests only one thread computes PBKDF2, while the fast    path (after initialisation) requires zero locks.    """    global _AUTH_HASH_COMPUTED, _AUTH_HASH_CACHE     # Fast path — no lock needed once cache is populated.    if _AUTH_HASH_COMPUTED:        return _AUTH_HASH_CACHE     with _AUTH_HASH_LOCK:        # Re-check inside lock — another thread may have populated while        # we were waiting to acquire.        if _AUTH_HASH_COMPUTED:            return _AUTH_HASH_CACHE         env_pw = os.getenv('HERMES_WEBUI_PASSWORD', '').strip()        if env_pw:            result = _hash_password(env_pw)        else:            result = load_settings().get('password_hash') or None         _AUTH_HASH_CACHE = result        _AUTH_HASH_COMPUTED = True        return result  def is_password_auth_enabled() -> bool:    """True if a password is configured (env var or settings)."""    return get_password_hash() is not None  def _passkey_feature_flag_enabled() -> bool:    """Return True if the passkey/WebAuthn surface is enabled for this deployment.     Passkey support is opt-in default-off behind a feature flag so deployments    that don't want the WebAuthn surface (or whose RP-ID setup isn't ready for    non-localhost hosts) can disable it entirely with no UI surface, no    endpoints, no credential storage. To enable:       - Set ``HERMES_WEBUI_PASSKEY=1`` in the environment, OR      - Set ``webui_passkey_enabled: true`` in the per-profile config.yaml     With the flag off, ``are_passkeys_enabled()`` always returns False even if    credentials were registered in the past, and ``/login`` shows password-only.    """    env_value = os.getenv("HERMES_WEBUI_PASSKEY", "")    if env_value:        return env_value.strip().lower() in {"1", "true", "yes", "on"}    try:        from api.config import get_config         cfg = get_config()        if isinstance(cfg, dict):            raw = cfg.get("webui_passkey_enabled")            if isinstance(raw, bool):                return raw            if isinstance(raw, str):                return raw.strip().lower() in {"1", "true", "yes", "on"}    except Exception:        pass    return False  def are_passkeys_enabled() -> bool:    """True if the passkey feature flag is on AND at least one local passkey credential is registered."""    if not _passkey_feature_flag_enabled():        return False    try:        from api.passkeys import passkeys_available         return passkeys_available()    except Exception as exc:        logger.debug("Failed to inspect passkey availability: %s", exc)        return False  def is_oidc_auth_enabled() -> bool:    """True if native OIDC login is configured for WebUI sessions."""    try:        from api.auth_oidc import is_oidc_enabled         return is_oidc_enabled()    except Exception as exc:        logger.debug("Failed to inspect OIDC availability: %s", exc)        return False  def get_oidc_startup_warning() -> str | None:    """Return a startup warning when OIDC auth is only partially configured."""    try:        cfg = get_config()        raw = cfg.get("webui_oidc") if isinstance(cfg, dict) else {}        if not isinstance(raw, dict):

Lines 1–500 of 1219. Load more