fix(security): isolate interactive sudo password cache per session

This commit is contained in:
hharry11
2026-04-28 09:28:19 +03:00
committed by Teknium
parent efb7d27609
commit de03a332f7
3 changed files with 102 additions and 9 deletions

View File

@@ -145,8 +145,14 @@ def _check_disk_usage_warning():
return False
# Session-cached sudo password (persists until CLI exits)
_cached_sudo_password: str = ""
# Interactive sudo password cache.
#
# Scope the cache to the active session when a session key is available, then
# fall back to callback identity (ACP / CLI interactive callbacks), then the
# current thread. This prevents one interactive session from reusing another
# session's cached sudo password inside the same long-lived process.
_sudo_password_cache: dict[str, str] = {}
_sudo_password_cache_lock = threading.Lock()
# Optional UI callbacks for interactive prompts. When set, these are called
# instead of the default /dev/tty or input() readers. The CLI registers these
@@ -190,6 +196,54 @@ def set_approval_callback(cb):
"""
_callback_tls.approval = cb
def _get_sudo_password_cache_scope() -> str:
"""Return the cache scope for interactive sudo passwords."""
try:
from gateway.session_context import get_session_env
session_key = get_session_env("HERMES_SESSION_KEY", "")
except Exception:
session_key = os.getenv("HERMES_SESSION_KEY", "")
if session_key:
return f"session:{session_key}"
callback = _get_sudo_password_callback()
if callback is not None:
owner = getattr(callback, "__self__", None)
func = getattr(callback, "__func__", None)
if owner is not None and func is not None:
return f"callback-owner:{id(owner)}:{id(func)}"
return f"callback:{id(callback)}"
return f"thread:{threading.get_ident()}"
def _get_cached_sudo_password() -> str:
"""Return the cached sudo password for the current scope."""
scope = _get_sudo_password_cache_scope()
with _sudo_password_cache_lock:
return _sudo_password_cache.get(scope, "")
def _set_cached_sudo_password(password: str) -> None:
"""Persist a sudo password for the current scope."""
scope = _get_sudo_password_cache_scope()
with _sudo_password_cache_lock:
if password:
_sudo_password_cache[scope] = password
else:
_sudo_password_cache.pop(scope, None)
def _reset_cached_sudo_passwords() -> None:
"""Clear all cached sudo passwords.
Internal helper for tests and process teardown paths.
"""
with _sudo_password_cache_lock:
_sudo_password_cache.clear()
# =============================================================================
# Dangerous Command Approval System
# =============================================================================
@@ -700,8 +754,6 @@ def _transform_sudo_command(command: str | None) -> tuple[str | None, str | None
If SUDO_PASSWORD is not set and NOT interactive:
Command runs as-is (fails gracefully with "sudo: a password is required").
"""
global _cached_sudo_password
if command is None:
return None, None
transformed, has_real_sudo = _rewrite_real_sudo_invocations(command)
@@ -709,12 +761,16 @@ def _transform_sudo_command(command: str | None) -> tuple[str | None, str | None
return command, None
has_configured_password = "SUDO_PASSWORD" in os.environ
sudo_password = os.environ.get("SUDO_PASSWORD", "") if has_configured_password else _cached_sudo_password
sudo_password = (
os.environ.get("SUDO_PASSWORD", "")
if has_configured_password
else _get_cached_sudo_password()
)
if not has_configured_password and not sudo_password and os.getenv("HERMES_INTERACTIVE"):
sudo_password = _prompt_for_sudo_password(timeout_seconds=45)
if sudo_password:
_cached_sudo_password = sudo_password
_set_cached_sudo_password(sudo_password)
if has_configured_password or sudo_password:
# Trailing newline is required: sudo -S reads one line for the password.