diff --git a/agent/monitoring/gateway_health.py b/agent/monitoring/gateway_health.py index 21419a4f25..6e6e016dc1 100644 --- a/agent/monitoring/gateway_health.py +++ b/agent/monitoring/gateway_health.py @@ -35,12 +35,8 @@ class GatewayHealthSnapshot: _RUNNING_PLATFORM_STATES = {"running", "connected", "ok", "ready"} _FATAL_PLATFORM_STATES = {"fatal", "degraded", "error", "failed"} -_KNOWN_GATEWAY_STATES = { - "starting", "draining", "stopping", "stopped", "startup_failed", "unknown" -} | _RUNNING_PLATFORM_STATES | _FATAL_PLATFORM_STATES -_KNOWN_PLATFORM_STATES = _RUNNING_PLATFORM_STATES | _FATAL_PLATFORM_STATES | { - "connecting", "disconnected", "disabled", "paused", "retrying", "unknown" -} +_KNOWN_GATEWAY_STATES = _RUNNING_PLATFORM_STATES | _FATAL_PLATFORM_STATES | {"starting", "draining", "stopping", "stopped", "startup_failed", "unknown"} +_KNOWN_PLATFORM_STATES = _RUNNING_PLATFORM_STATES | _FATAL_PLATFORM_STATES | {"connecting", "disconnected", "disabled", "paused", "retrying", "unknown"} _SUPERVISION_MODES = {"systemd", "s6", "container", "launchd", "manual", "unknown"} _SOURCE_LOGGER_RE = re.compile(r"^gateway(?:\.[A-Za-z_][A-Za-z0-9_]*)*$") @@ -124,6 +120,14 @@ def platform_for_subsystem(subsystem: str) -> Optional[str]: return (subsystem.split(".", 1)[1] or None) if subsystem.startswith("platform.") else None +def _coerce_pid(raw: Any) -> Optional[int]: + try: + pid = int(raw) + except (TypeError, ValueError): + return None + return pid if pid > 0 else None + + def _gateway_status(name: str, fallback: Callable[[], Any], /, **kwargs: Any) -> Any: """Prefer ``gateway.status.`` (the runtime-status contract); fall back to the local approximation.""" try: @@ -133,52 +137,22 @@ def _gateway_status(name: str, fallback: Callable[[], Any], /, **kwargs: Any) -> return fallback() -def _int_or_zero(raw: Any) -> int: - try: - return max(0, int(raw)) - except (TypeError, ValueError): - return 0 - - def _parse_active_agents(raw: Any) -> int: - return _gateway_status("parse_active_agents", lambda: _int_or_zero(raw), raw=raw) + def fallback() -> int: + try: + return max(0, int(raw)) + except (TypeError, ValueError): + return 0 + + return _gateway_status("parse_active_agents", fallback, raw=raw) -def _derive_busy(gateway_running: bool, gateway_state: Any, active_agents: Any) -> bool: - fallback = lambda: bool(gateway_running and gateway_state == "running" and _parse_active_agents(active_agents) > 0) # noqa: E731 - return _gateway_status("derive_gateway_busy", fallback, gateway_running=gateway_running, gateway_state=gateway_state, active_agents=active_agents) - - -def _derive_drainable(gateway_running: bool, gateway_state: Any) -> bool: - fallback = lambda: bool(gateway_running and gateway_state == "running") # noqa: E731 - return _gateway_status("derive_gateway_drainable", fallback, gateway_running=gateway_running, gateway_state=gateway_state) - - -def _base_attrs(*, install_id: str, version: str, supervision_mode: str) -> Dict[str, str]: - mode = str(supervision_mode or "unknown").lower() - return { - "service.instance.id": _safe_instance_id(install_id), - "service.version": _safe_metric_value(version, limit=64), - "hermes.supervision_mode": mode if mode in _SUPERVISION_MODES else "unknown", - } - - -def _metric(name: str, value: int | float, attrs: Dict[str, str], **extra: str) -> GatewayMetric: - out = dict(attrs) - for key, val in extra.items(): - if val is not None: - out[key] = _safe_metric_value(val) - return GatewayMetric(name=name, value=value, attributes=out) - - -def _platforms_of(runtime: Optional[dict[str, Any]]) -> dict[str, Any]: - raw = (runtime or {}).get("platforms") +def _dict_or_empty(raw: Any) -> dict[str, Any]: return raw if isinstance(raw, dict) else {} -def _platform_error_code(pdata: dict[str, Any]) -> str: - # classify_* is idempotent on its own labels, so error_class == error_code downstream. - return classify_gateway_error(pdata.get("error_code") or pdata.get("error_message")) +def _platforms_of(runtime: Optional[dict[str, Any]]) -> dict[str, Any]: + return _dict_or_empty((runtime or {}).get("platforms")) def build_gateway_health_snapshot( @@ -189,32 +163,49 @@ def build_gateway_health_snapshot( runtime = runtime or {} gateway_state = _bounded_state(runtime.get("gateway_state"), allowed=_KNOWN_GATEWAY_STATES) active_agents = _parse_active_agents(runtime.get("active_agents", 0)) - busy = _derive_busy(gateway_running, gateway_state, active_agents) - drainable = _derive_drainable(gateway_running, gateway_state) + running = gateway_running and gateway_state == "running" + busy = _gateway_status( + "derive_gateway_busy", lambda: bool(running and _parse_active_agents(active_agents) > 0), + gateway_running=gateway_running, gateway_state=gateway_state, active_agents=active_agents, + ) + drainable = _gateway_status( + "derive_gateway_drainable", lambda: bool(running), gateway_running=gateway_running, gateway_state=gateway_state + ) platforms = _platforms_of(runtime) - base = _base_attrs(install_id=install_id, version=version, supervision_mode=supervision_mode) + mode = str(supervision_mode or "unknown").lower() + base = { + "service.instance.id": _safe_instance_id(install_id), + "service.version": _safe_metric_value(version, limit=64), + "hermes.supervision_mode": mode if mode in _SUPERVISION_MODES else "unknown", + } + + def metric(name: str, value: int | float, **extra: str) -> GatewayMetric: + attrs = dict(base) + for key, val in extra.items(): + if val is not None: + attrs[key] = _safe_metric_value(val) + return GatewayMetric(name=name, value=value, attributes=attrs) + metrics: list[GatewayMetric] = [ - _metric("hermes.gateway.up", 1 if gateway_running else 0, base), - _metric("hermes.gateway.active_agents", active_agents, base), - _metric("hermes.gateway.busy", 1 if busy else 0, base), - _metric("hermes.gateway.drainable", 1 if drainable else 0, base), - _metric("hermes.gateway.restart_requested", 1 if runtime.get("restart_requested") else 0, base), - _metric("hermes.gateway.state", 1, base, **{"hermes.gateway.state": gateway_state}), + metric("hermes.gateway.up", int(bool(gateway_running))), + metric("hermes.gateway.active_agents", active_agents), + metric("hermes.gateway.busy", int(bool(busy))), + metric("hermes.gateway.drainable", int(bool(drainable))), + metric("hermes.gateway.restart_requested", int(bool(runtime.get("restart_requested")))), + metric("hermes.gateway.state", 1, **{"hermes.gateway.state": gateway_state}), ] fatal_count = 0 events: list[GatewayHealthEvent | GatewayDiagnosticEvent] = [] + # classify_* is idempotent on its own labels, so error_class == error_code downstream. for platform, pdata in platforms.items(): - pdata = pdata if isinstance(pdata, dict) else {} + pdata = _dict_or_empty(pdata) state = _bounded_state(pdata.get("state"), allowed=_KNOWN_PLATFORM_STATES) - error_code = _platform_error_code(pdata) + error_code = classify_gateway_error(pdata.get("error_code") or pdata.get("error_message")) is_degraded = state in _FATAL_PLATFORM_STATES - if is_degraded: - fatal_count += 1 + fatal_count += is_degraded pattrs = {"hermes.platform": str(platform), "hermes.platform.state": state} - metrics.append(_metric("hermes.platform.up", 1 if state in _RUNNING_PLATFORM_STATES else 0, base, **pattrs)) - metrics.append(_metric( - "hermes.platform.degraded", 1 if is_degraded else 0, base, **pattrs, **{"hermes.error_code": error_code} - )) + metrics.append(metric("hermes.platform.up", int(state in _RUNNING_PLATFORM_STATES), **pattrs)) + metrics.append(metric("hermes.platform.degraded", int(is_degraded), **pattrs, **{"hermes.error_code": error_code})) if is_degraded: events.append(GatewayDiagnosticEvent( name="platform.fatal", subsystem=f"platform.{platform}", platform=str(platform), @@ -282,14 +273,13 @@ def _platform_events( old_platforms = _platforms_of(previous) out: list[GatewayDiagnosticEvent] = [] for platform, pdata in _platforms_of(current).items(): - pdata = pdata if isinstance(pdata, dict) else {} - prev_raw = old_platforms.get(platform, {}) - prev = prev_raw if isinstance(prev_raw, dict) else {} + pdata = _dict_or_empty(pdata) + prev = _dict_or_empty(old_platforms.get(platform, {})) old_state = _optional_state(prev.get("state"), allowed=_KNOWN_PLATFORM_STATES) new_state = _optional_state(pdata.get("state"), allowed=_KNOWN_PLATFORM_STATES) if old_state == new_state or not new_state: continue - error_code = _platform_error_code(pdata) + error_code = classify_gateway_error(pdata.get("error_code") or pdata.get("error_message")) common: dict[str, Any] = dict( subsystem=f"platform.{platform}", platform=str(platform), error_code=error_code, error_class=error_code, profile=profile, version=version, severity="error" if new_state in {"fatal", "failed", "error"} else "warning", @@ -311,14 +301,6 @@ def emit_runtime_status_transition(previous: Optional[dict[str, Any]], current: logger.debug("gateway runtime status transition emit failed", exc_info=True) -def _coerce_pid(raw: Any) -> Optional[int]: - try: - pid = int(raw) - except (TypeError, ValueError): - return None - return pid if pid > 0 else None - - class GatewayDiagnosticLogHandler(logging.Handler): """Allowlisted warning/error bridge for gateway-owned diagnostics.""" diff --git a/agent/monitoring/gateway_health_export.py b/agent/monitoring/gateway_health_export.py index 68f69bccc9..616d1e5d13 100644 --- a/agent/monitoring/gateway_health_export.py +++ b/agent/monitoring/gateway_health_export.py @@ -39,7 +39,6 @@ from agent.monitoring.redaction import redact_bounded logger = logging.getLogger(__name__) -_DEFAULT_DIAGNOSTIC_SCOPE = "hermes.gateway.diagnostics" _METRICS_SDK = ( "OTLPLogExporter", "OTLPMetricExporter", "Observation", "LogRecord", "LoggerProvider", "INVALID_SPAN_ID", "INVALID_TRACE_ID", "TraceFlags", "SeverityNumber", @@ -90,7 +89,7 @@ class GatewayHealthExportRuntime: bus.unsubscribe(sub) # Network flush/close runs under one bounded daemon-thread deadline so it can # never delay gateway teardown indefinitely. - closeables = subscribers + ([self.metric_provider] if self.metric_provider is not None else []) + closeables = [item for item in (self.streamer, self.log_streamer, self.metric_provider) if item is not None] def _close() -> None: for item in closeables: @@ -112,23 +111,9 @@ def _enabled(config: Dict[str, Any]) -> bool: return bool(_gateway_health_config(config).get("enabled") and otlp_exporter.is_enabled(config)) -def _require_metrics_sdk(*, auto_install: bool = True, prompt: bool = False) -> Dict[str, Any]: - try: - return otlp_exporter._require_sdk(_METRICS_SDK, auto_install=auto_install, prompt=prompt) - except Exception as exc: - raise RuntimeError(f"OTLP metrics SDK unavailable: {exc}") from exc - - def _exporter_kwargs(config: Dict[str, Any], signal: str) -> Dict[str, Any]: otlp = _otlp_config(config) - return { - "endpoint": _signal_endpoint(str(otlp.get("endpoint")), signal), - "headers": _resolve_headers(otlp.get("headers_env")) or None, - } - - -def _resource(config: Dict[str, Any], sdk: Dict[str, Any], telemetry_scope: str) -> Any: - return sdk["Resource"].create(_runtime_resource_attributes(config, telemetry_scope=telemetry_scope)) + return {"endpoint": _signal_endpoint(str(otlp.get("endpoint")), signal), "headers": _resolve_headers(otlp.get("headers_env")) or None} # Ordered detection: systemd > s6 > container > launchd > manual (first match wins). @@ -175,10 +160,8 @@ def _read_background_work_count() -> int: runs) deliberately does NOT include: backgrounded ``delegate_task`` subagents, ``terminal(background=true)`` processes, kanban workers. TASK-granular: a fan-out batch of N contributes N (real concurrent load), unlike the pool's one-slot-per-batch accounting.""" - return ( - _count("background-work async-delegation count failed", "tools.async_delegation", lambda m: m.active_task_count()) - + _count("background-work process-registry count failed", "tools.process_registry", - lambda m: m.process_registry.count_running()) + return _count("background-work async-delegation count failed", "tools.async_delegation", lambda m: m.active_task_count()) + _count( + "background-work process-registry count failed", "tools.process_registry", lambda m: m.process_registry.count_running() ) @@ -203,14 +186,12 @@ def _read_runtime_snapshot(config: Dict[str, Any]): logger.warning("background-work snapshot unavailable; metric not exported (error_type=%s)", type(exc).__name__) logger.debug("background-work snapshot traceback", exc_info=True) try: - cron_snapshot = _read_cron_snapshot() + gateway_snapshot.metrics.extend(_read_cron_snapshot().metrics) except Exception as exc: # Cron telemetry silently dropping out is a release-relevant regression: WARN with only # the exception *type* (the message could carry paths); exc_info stays on DEBUG. logger.warning("cron health snapshot unavailable; cron telemetry not exported (error_type=%s)", type(exc).__name__) logger.debug("cron health snapshot traceback", exc_info=True) - return gateway_snapshot - gateway_snapshot.metrics.extend(cron_snapshot.metrics) return gateway_snapshot @@ -225,11 +206,11 @@ def _emit_snapshot_events(config: Dict[str, Any]) -> None: def _start_metric_provider(config: Dict[str, Any], sdk: Dict[str, Any]) -> Any: - gh = _gateway_health_config(config) exporter = sdk["OTLPMetricExporter"](**_exporter_kwargs(config, "metrics")) - interval_ms = max(5, int(gh.get("export_interval_seconds", 60))) * 1000 + interval_ms = max(5, int(_gateway_health_config(config).get("export_interval_seconds", 60))) * 1000 reader = sdk["PeriodicExportingMetricReader"](exporter, export_interval_millis=interval_ms) - provider = sdk["MeterProvider"](metric_readers=[reader], resource=_resource(config, sdk, "gateway_health")) + resource = sdk["Resource"].create(_runtime_resource_attributes(config, telemetry_scope="gateway_health")) + provider = sdk["MeterProvider"](metric_readers=[reader], resource=resource) meter = provider.get_meter("hermes.gateway.health") Observation = sdk["Observation"] @@ -248,24 +229,22 @@ def _start_metric_provider(config: Dict[str, Any], sdk: Dict[str, Any]) -> Any: return provider -_SEVERITY_NAMES = { - "critical": "FATAL", "fatal": "FATAL", "error": "ERROR", "info": "INFO", "information": "INFO", "debug": "DEBUG", -} +_SEVERITY_NAMES = {"critical": "FATAL", "fatal": "FATAL", "error": "ERROR", "info": "INFO", "information": "INFO", "debug": "DEBUG"} def _severity_number(sdk: Dict[str, Any], severity: Any) -> Any: - sev = str(severity or "warning").lower() - return getattr(sdk["SeverityNumber"], _SEVERITY_NAMES.get(sev, "WARN")) + return getattr(sdk["SeverityNumber"], _SEVERITY_NAMES.get(str(severity or "warning").lower(), "WARN")) class GatewayDiagnosticLogStreamer(EmitterStreamer): """Emitter subscriber that sends gateway diagnostic events as OTLP logs.""" def __init__(self, config: Dict[str, Any], sdk: Dict[str, Any]): - self._provider = sdk["LoggerProvider"](resource=_resource(config, sdk, "gateway_diagnostics")) + resource = sdk["Resource"].create(_runtime_resource_attributes(config, telemetry_scope="gateway_diagnostics")) + self._provider = sdk["LoggerProvider"](resource=resource) self._processor = sdk["BatchLogRecordProcessor"](sdk["OTLPLogExporter"](**_exporter_kwargs(config, "logs"))) self._provider.add_log_record_processor(self._processor) - self._logger = self._provider.get_logger(_DEFAULT_DIAGNOSTIC_SCOPE) + self._logger = self._provider.get_logger("hermes.gateway.diagnostics") self._sdk = sdk self.exported = 0 @@ -288,33 +267,16 @@ class GatewayDiagnosticLogStreamer(EmitterStreamer): self.exported += 1 -def _start_snapshot_thread(config: Dict[str, Any], stop_event: threading.Event) -> threading.Thread: - interval = max(5, int(_gateway_health_config(config).get("logs_export_interval_seconds", 5))) - - def _run() -> None: - while not stop_event.wait(interval): - _emit_snapshot_events(config) - - thread = threading.Thread(target=_run, name="hermes-gateway-health-export", daemon=True) - thread.start() - return thread - - -def _attach_log_handler(config: Dict[str, Any]) -> Any: - gh = _gateway_health_config(config) - if not gh.get("diagnostic_events_enabled", True) or not gh.get("warning_error_events_enabled", True): - return None - handler = GatewayDiagnosticLogHandler(profile=_profile(), version=_version()) - root = logging.getLogger() - if handler not in root.handlers: - root.addHandler(handler) - return handler - - def _gateway_health_event(ev: Dict[str, Any]) -> bool: return ev.get("event") in {"gateway_health", "cron_execution"} +def _fail(runtime: GatewayHealthExportRuntime, log: Callable[..., None], msg: str, reason: str) -> GatewayHealthExportRuntime: + log(msg, exc_info=True) + runtime.shutdown() + return GatewayHealthExportRuntime(enabled=False, reason=reason) + + def start_gateway_health_export(config: Dict[str, Any]) -> GatewayHealthExportRuntime: """Start P0 gateway health export if configured. Never raises.""" if not _enabled(config): @@ -326,7 +288,7 @@ def start_gateway_health_export(config: Dict[str, Any]) -> GatewayHealthExportRu sdk: Optional[Dict[str, Any]] = None if metrics_on or diagnostics_on: try: - sdk = _require_metrics_sdk(prompt=False) + sdk = otlp_exporter._require_sdk(_METRICS_SDK, auto_install=True, prompt=False) except Exception: logger.warning("monitoring.gateway_health_export.enabled but OTLP SDK is unavailable; install 'hermes-agent[otlp]'", exc_info=True) return GatewayHealthExportRuntime(enabled=False, reason="otlp_unavailable") @@ -334,9 +296,7 @@ def start_gateway_health_export(config: Dict[str, Any]) -> GatewayHealthExportRu try: runtime.metric_provider = _start_metric_provider(config, sdk) except Exception: - logger.warning("gateway health OTLP metrics failed to start", exc_info=True) - runtime.shutdown() - return GatewayHealthExportRuntime(enabled=False, reason="metrics_start_failed") + return _fail(runtime, logger.warning, "gateway health OTLP metrics failed to start", "metrics_start_failed") if diagnostics_on and sdk is not None: try: runtime.streamer = otlp_exporter.start_streaming(config, event_filter=_gateway_health_event) @@ -346,18 +306,29 @@ def start_gateway_health_export(config: Dict[str, Any]) -> GatewayHealthExportRu emitter.get_emitter().subscribe(log_streamer) runtime.log_streamer = log_streamer except Exception: - logger.debug("gateway diagnostic OTLP export failed to start", exc_info=True) - runtime.shutdown() - return GatewayHealthExportRuntime(enabled=False, reason="diagnostics_start_failed") + return _fail(runtime, logger.debug, "gateway diagnostic OTLP export failed to start", "diagnostics_start_failed") try: - runtime.log_handler = _attach_log_handler(config) + if diagnostics_on and gh.get("warning_error_events_enabled", True): + handler = GatewayDiagnosticLogHandler(profile=_profile(), version=_version()) + root = logging.getLogger() + if handler not in root.handlers: + root.addHandler(handler) + runtime.log_handler = handler except Exception: logger.debug("gateway diagnostic log handler failed to attach", exc_info=True) if diagnostics_on: try: _emit_snapshot_events(config) - runtime.stop_event = threading.Event() - runtime.thread = _start_snapshot_thread(config, runtime.stop_event) + stop_event = runtime.stop_event = threading.Event() + interval = max(5, int(gh.get("logs_export_interval_seconds", 5))) + + def _run() -> None: + while not stop_event.wait(interval): + _emit_snapshot_events(config) + + thread = threading.Thread(target=_run, name="hermes-gateway-health-export", daemon=True) + thread.start() + runtime.thread = thread except Exception: logger.debug("gateway health snapshot thread failed to start", exc_info=True) return runtime diff --git a/agent/outbound_webhooks.py b/agent/outbound_webhooks.py index d9a81af1c3..61c8aa0e1e 100644 --- a/agent/outbound_webhooks.py +++ b/agent/outbound_webhooks.py @@ -69,8 +69,6 @@ class WebhookTarget(_ToolMatcherMixin): return self.name or self.url -# --- Public API ----------------------------------------------------------------- - def register_from_config(cfg: Optional[Dict[str, Any]]) -> List[WebhookTarget]: """Register every configured outbound webhook on the plugin manager. Malformed ``hooks.outbound`` means zero targets — never raises. Returns the targets that ended up wired (deduplicated).""" @@ -117,8 +115,7 @@ def iter_configured_targets(cfg: Optional[Dict[str, Any]]) -> List[WebhookTarget if not isinstance(raw, list): logger.warning("hooks.outbound must be a list of webhook targets; got %s", type(raw).__name__) return [] - targets = (_parse_single_target(i, entry) for i, entry in enumerate(raw)) - return [t for t in targets if t is not None] + return [t for t in (_parse_single_target(i, entry) for i, entry in enumerate(raw)) if t is not None] def flush(timeout: float = 5.0) -> bool: @@ -154,8 +151,6 @@ def reset_for_tests() -> None: pass -# --- Config parsing ------------------------------------------------------------- - def _parse_single_target(index: int, raw: Any) -> Optional[WebhookTarget]: from hermes_cli.plugins import VALID_HOOKS @@ -201,46 +196,34 @@ def _parse_single_target(index: int, raw: Any) -> Optional[WebhookTarget]: warn(".timeout must be an int (got %r); using default %ds", timeout_raw, DEFAULT_TIMEOUT_SECONDS) timeout = DEFAULT_TIMEOUT_SECONDS name = raw.get("name") + # ``secret_env`` (env var name, preferred) wins over inline ``secret``. + secret_env = raw.get("secret_env") + if isinstance(secret_env, str) and secret_env.strip(): + secret = os.environ.get(secret_env.strip(), "") or None + if secret is None: + warn(".secret_env=%r is not set in the environment — deliveries will be UNSIGNED", secret_env.strip()) + else: + secret = raw.get("secret") + secret = secret if isinstance(secret, str) and secret else None return WebhookTarget( - url=url, events=events, name=name.strip() if isinstance(name, str) else "", secret=_resolve_secret(index, raw), + url=url, events=events, name=name.strip() if isinstance(name, str) else "", secret=secret, matcher=matcher, timeout=max(1, min(timeout, MAX_TIMEOUT_SECONDS)), ) -def _resolve_secret(index: int, raw: Dict[str, Any]) -> Optional[str]: - """``secret_env`` (env var name, preferred) wins over inline ``secret``.""" - secret_env = raw.get("secret_env") - if isinstance(secret_env, str) and secret_env.strip(): - value = os.environ.get(secret_env.strip(), "") - if value: - return value - logger.warning( - "hooks.outbound[%d].secret_env=%r is not set in the environment — deliveries will be UNSIGNED", - index, secret_env.strip(), - ) - return None - secret = raw.get("secret") - return secret if isinstance(secret, str) and secret else None - - -# --- Callback + delivery -------------------------------------------------------- - def _make_callback(event: str, target: WebhookTarget): """Build the notify-only closure ``invoke_hook()`` calls per firing.""" def _callback(**kwargs: Any) -> None: if event in _TOOL_SCOPED_EVENTS and not target.matches_tool(kwargs.get("tool_name")): - return None + return delivery_id = uuid.uuid4().hex try: body = _serialize_payload(event, kwargs, delivery_id) except Exception: # a bad payload must not hurt the loop - logger.warning( - "outbound webhook payload serialization failed (event=%s target=%s)", event, target.label, exc_info=True, - ) - return None + logger.warning("outbound webhook payload serialization failed (event=%s target=%s)", event, target.label, exc_info=True) + return _enqueue(_build_delivery(event, target, body, delivery_id)) - return None _callback.__name__ = f"outbound_webhook[{event}:{target.label}]" _callback.__qualname__ = _callback.__name__ @@ -268,14 +251,19 @@ def _build_delivery(event: str, target: WebhookTarget, body: bytes, delivery_id: if target.secret: digest = hmac.new(target.secret.encode("utf-8"), body, hashlib.sha256).hexdigest() headers["X-Hermes-Signature-256"] = f"sha256={digest}" - return { - "url": target.url, "label": target.label, "event": event, - "body": body, "headers": headers, "timeout": target.timeout, - } + return {"url": target.url, "label": target.label, "event": event, "body": body, "headers": headers, "timeout": target.timeout} def _enqueue(delivery: Dict[str, Any]) -> None: - _ensure_worker() + global _worker + if _worker is None or not _worker.is_alive(): + with _worker_lock: + if _worker is None or not _worker.is_alive(): + _worker = threading.Thread(target=_worker_loop, name="outbound-webhooks", daemon=True) + _worker.start() + # Daemon worker: a short-lived process could exit right after enqueuing on_session_end. + # Drain at interpreter shutdown, bounded so a dead endpoint can only delay exit, never hang it. + atexit.register(flush, timeout=5.0) try: _delivery_queue.put_nowait(delivery) except queue.Full: @@ -285,20 +273,6 @@ def _enqueue(delivery: Dict[str, Any]) -> None: ) -def _ensure_worker() -> None: - global _worker - if _worker is not None and _worker.is_alive(): - return - with _worker_lock: - if _worker is not None and _worker.is_alive(): - return - _worker = threading.Thread(target=_worker_loop, name="outbound-webhooks", daemon=True) - _worker.start() - # Daemon worker: a short-lived process could exit right after enqueuing on_session_end. - # Drain at interpreter shutdown, bounded so a dead endpoint can only delay exit, never hang it. - atexit.register(flush, timeout=5.0) - - def _worker_loop() -> None: while True: delivery = _delivery_queue.get() @@ -306,11 +280,8 @@ def _worker_loop() -> None: if delivery is not None: _deliver(delivery) except Exception: # pragma: no cover — defensive - logger.warning( - "outbound webhook delivery crashed (target=%s)", - delivery.get("label") if isinstance(delivery, dict) else "?", - exc_info=True, - ) + label = delivery.get("label") if isinstance(delivery, dict) else "?" + logger.warning("outbound webhook delivery crashed (target=%s)", label, exc_info=True) finally: _delivery_queue.task_done() diff --git a/agent/proxy_sources/iron_proxy.py b/agent/proxy_sources/iron_proxy.py index 904c523025..f0b4c9f342 100644 --- a/agent/proxy_sources/iron_proxy.py +++ b/agent/proxy_sources/iron_proxy.py @@ -54,9 +54,8 @@ _DEFAULT_TUNNEL_PORT = 9090 # Hosts allowed by default for AI inference traffic. Anything else is 403'd. _DEFAULT_ALLOWED_HOSTS: Tuple[str, ...] = ( - "openrouter.ai", "*.openrouter.ai", "api.openai.com", "api.anthropic.com", - "generativelanguage.googleapis.com", "api.x.ai", "api.mistral.ai", "api.groq.com", - "api.together.xyz", "api.deepseek.com", "inference.nousresearch.com", + "openrouter.ai", "*.openrouter.ai", "api.openai.com", "api.anthropic.com", "generativelanguage.googleapis.com", + "api.x.ai", "api.mistral.ai", "api.groq.com", "api.together.xyz", "api.deepseek.com", "inference.nousresearch.com", ) # Provider env-var name -> upstream hosts on which the Authorization Bearer token is swapped. @@ -74,10 +73,8 @@ _BEARER_PROVIDERS: Dict[str, Tuple[str, ...]] = { # Gemini's ``?key=`` style is covered by match_query. _HEADER_AUTH_PROVIDERS: Dict[str, Dict[str, Tuple[str, ...]]] = { "ANTHROPIC_API_KEY": {"hosts": ("api.anthropic.com",), "match_headers": ("x-api-key", "Authorization"), "aliases": ()}, - "AZURE_OPENAI_API_KEY": { - "hosts": ("*.openai.azure.com", "*.cognitiveservices.azure.com", "*.services.ai.azure.com"), - "match_headers": ("api-key", "Authorization"), "aliases": (), - }, + "AZURE_OPENAI_API_KEY": {"hosts": ("*.openai.azure.com", "*.cognitiveservices.azure.com", "*.services.ai.azure.com"), + "match_headers": ("api-key", "Authorization"), "aliases": ()}, "GEMINI_API_KEY": {"hosts": ("generativelanguage.googleapis.com",), "match_headers": ("x-goog-api-key",), "aliases": ("GOOGLE_API_KEY",)}, } @@ -95,16 +92,14 @@ _DEFAULT_UPSTREAM_DENY_CIDRS: Tuple[str, ...] = ( "198.18.0.0/15", # RFC2544 benchmark range ) -# Minimal daemon env; everything else is stripped so /proc//environ never exposes operator secrets. +# Minimal daemon env (SYSTEMROOT/USERPROFILE are Windows); everything else is stripped so +# /proc//environ never exposes operator secrets. _PROXY_SUBPROCESS_ENV_ALLOWLIST: Tuple[str, ...] = ( - "PATH", "HOME", "TMPDIR", "TZ", "LANG", "LC_ALL", "LC_CTYPE", "NO_COLOR", "SSL_CERT_DIR", "SSL_CERT_FILE", - "SYSTEMROOT", "USERPROFILE", # Windows + "PATH", "HOME", "TMPDIR", "TZ", "LANG", "LC_ALL", "LC_CTYPE", "NO_COLOR", "SSL_CERT_DIR", "SSL_CERT_FILE", "SYSTEMROOT", "USERPROFILE", ) # Always stripped — these would recurse the proxy through itself or a corporate proxy. -_PROXY_SUBPROCESS_ENV_STRIP: Tuple[str, ...] = ( - "HTTPS_PROXY", "https_proxy", "HTTP_PROXY", "http_proxy", "ALL_PROXY", "all_proxy", "NO_PROXY", "no_proxy", -) +_PROXY_SUBPROCESS_ENV_STRIP: Tuple[str, ...] = ("HTTPS_PROXY", "https_proxy", "HTTP_PROXY", "http_proxy", "ALL_PROXY", "all_proxy", "NO_PROXY", "no_proxy") # SIGKILL doesn't exist on Windows; SIGTERM there is TerminateProcess() — same semantics. _KILL_SIGNAL = getattr(signal, "SIGKILL", signal.SIGTERM) @@ -122,7 +117,6 @@ _proxy_nonce: Optional[str] = None @dataclass class ProxyStatus: - """Snapshot of the iron-proxy installation + runtime state.""" enabled: bool = False binary_path: Optional[Path] = None binary_version: Optional[str] = None @@ -144,8 +138,8 @@ class ProxyStatus: @dataclass class TokenMapping: - """Sandbox-visible proxy token -> upstream credential lookup. ``real_env_name`` is read from - iron-proxy's OWN env at egress; ``alias_env_names`` are extra SANDBOX names for the same token.""" + """Sandbox-visible proxy token -> upstream credential lookup. ``real_env_name`` is read from iron-proxy's + OWN env at egress; ``alias_env_names`` are extra SANDBOX names for the same token.""" proxy_token: str real_env_name: str upstream_hosts: Tuple[str, ...] @@ -158,17 +152,14 @@ def _hermes_bin_dir() -> Path: return get_hermes_home() / "bin" -def _proxy_state_dir_ro() -> Path: - """Proxy state dir without creating it (status probes, pidfile reads).""" +def _proxy_state_dir_ro() -> Path: # without creating it (status probes, pidfile reads) from hermes_constants import get_hermes_home return get_hermes_home() / "proxy" def _proxy_state_dir() -> Path: - """Proxy state dir (CA key, pidfile, logs), created 0o700; chmod is unconditional so a - pre-existing slack-umask dir gets tightened.""" - d = _proxy_state_dir_ro() - d.mkdir(parents=True, exist_ok=True) + """Proxy state dir (CA key, pidfile, logs), created 0o700; unconditional chmod tightens a pre-existing slack-umask dir.""" + (d := _proxy_state_dir_ro()).mkdir(parents=True, exist_ok=True) with suppress(OSError): # Windows no-op / shared fs we don't own; files still get explicit perms d.chmod(0o700) return d @@ -180,16 +171,12 @@ def _platform_binary_name() -> str: def _platform_asset_name() -> str: """Map (uname, arch) -> ``iron-proxy___.tar.gz``; no Windows builds upstream.""" - system = platform.system() - machine = platform.machine().lower() - os_name = {"Linux": "linux", "Darwin": "darwin"}.get(system) - if os_name: + system, machine = platform.system(), platform.machine().lower() + if os_name := {"Linux": "linux", "Darwin": "darwin"}.get(system): arch = "arm64" if machine in ("arm64", "aarch64") else "amd64" return f"iron-proxy_{_IRON_PROXY_VERSION}_{os_name}_{arch}.tar.gz" if system == "Windows": - raise RuntimeError( - f"iron-proxy does not ship native Windows binaries as of v{_IRON_PROXY_VERSION}. Run the proxy on a Linux/macOS host, or inside WSL." - ) + raise RuntimeError(f"iron-proxy does not ship native Windows binaries as of v{_IRON_PROXY_VERSION}. Run the proxy on a Linux/macOS host, or inside WSL.") raise RuntimeError(f"Unsupported platform for iron-proxy auto-install: {system} {machine}") @@ -198,8 +185,7 @@ def find_iron_proxy(*, install_if_missing: bool = False) -> Optional[Path]: managed = _hermes_bin_dir() / _platform_binary_name() if managed.exists() and os.access(managed, os.X_OK): return managed - system = shutil.which("iron-proxy") - if system: + if system := shutil.which("iron-proxy"): return Path(system) if not install_if_missing: return None @@ -212,23 +198,19 @@ def find_iron_proxy(*, install_if_missing: bool = False) -> Optional[Path]: def install_iron_proxy(*, force: bool = False) -> Path: """Download, verify, and install the pinned binary; raises on any failure.""" - bin_dir = _hermes_bin_dir() - bin_dir.mkdir(parents=True, exist_ok=True) + (bin_dir := _hermes_bin_dir()).mkdir(parents=True, exist_ok=True) target = bin_dir / _platform_binary_name() if target.exists() and not force: return target asset_name = _platform_asset_name() with tempfile.TemporaryDirectory(prefix="hermes-iron-proxy-") as tmpdir: - tmp = Path(tmpdir) - archive_path = tmp / asset_name - checksum_path = tmp / _IRON_PROXY_CHECKSUM_NAME + archive_path, checksum_path = (tmp := Path(tmpdir)) / asset_name, tmp / _IRON_PROXY_CHECKSUM_NAME logger.info("Downloading %s", f"{_IRON_PROXY_RELEASE_BASE}/{asset_name}") _release_asset(asset_name, archive_path) _release_asset(_IRON_PROXY_CHECKSUM_NAME, checksum_path) # Best-effort GPG check of checksums.txt closes the release-channel tamper gap. _verify_checksums_signature(tmp, checksum_path) - expected = _expected_sha256(checksum_path, asset_name) - actual = _sha256_file(archive_path) + expected, actual = _expected_sha256(checksum_path, asset_name), _sha256_file(archive_path) if expected.lower() != actual.lower(): raise RuntimeError(f"Checksum mismatch for {asset_name}: expected {expected}, got {actual}") with tarfile.open(archive_path, "r:gz") as tf: @@ -254,8 +236,8 @@ def install_iron_proxy(*, force: bool = False) -> Path: def _release_asset(name: str, dest: Path) -> None: """Download one pinned-release asset to ``dest``; RuntimeError on any URL error.""" url = f"{_IRON_PROXY_RELEASE_BASE}/{name}" - req = urllib.request.Request(url, headers={"User-Agent": "hermes-agent"}) try: + req = urllib.request.Request(url, headers={"User-Agent": "hermes-agent"}) with urllib.request.urlopen(req, timeout=_DOWNLOAD_TIMEOUT) as resp, open(dest, "wb") as f: # noqa: S310 shutil.copyfileobj(resp, f) except urllib.error.URLError as exc: @@ -263,11 +245,9 @@ def _release_asset(name: str, dest: Path) -> None: def _verify_checksums_signature(tmp: Path, checksum_path: Path) -> bool: - """Best-effort GPG check of checksums.txt in an ephemeral keyring. False (with a warning) when - gpg or the signature assets are unavailable — SHA-256 stays enforced, gpg is never a hard - dependency. Raises ONLY on a present-but-bad signature (tamper signal).""" - gpg = shutil.which("gpg") - if not gpg: + """Best-effort GPG check of checksums.txt in an ephemeral keyring. False (with a warning) when gpg or the signature + assets are unavailable — SHA-256 stays enforced, gpg is never a hard dependency. Raises ONLY on a present-but-bad signature.""" + if not (gpg := shutil.which("gpg")): logger.warning("gpg not found on PATH — skipping iron-proxy release-signature verification (SHA-256 checksum check still enforced).") return False sig_path, pubkey_path = tmp / _IRON_PROXY_CHECKSUM_SIG_NAME, tmp / _IRON_PROXY_PUBKEY_NAME @@ -275,25 +255,16 @@ def _verify_checksums_signature(tmp: Path, checksum_path: Path) -> bool: _release_asset(_IRON_PROXY_CHECKSUM_SIG_NAME, sig_path) _release_asset(_IRON_PROXY_PUBKEY_NAME, pubkey_path) except RuntimeError as exc: - logger.warning( - "iron-proxy release signature assets unavailable (%s) — skipping GPG verification (SHA-256 checksum check still enforced).", exc, - ) + logger.warning("iron-proxy release signature assets unavailable (%s) — skipping GPG verification (SHA-256 checksum check still enforced).", exc) return False - gnupg_home = tmp / "gnupg" - gnupg_home.mkdir(mode=0o700, exist_ok=True) + (gnupg_home := tmp / "gnupg").mkdir(mode=0o700, exist_ok=True) gpg_base = [gpg, "--homedir", str(gnupg_home), "--batch", "--no-tty"] - imp = _run([*gpg_base, "--import", str(pubkey_path)], timeout=60) - if imp.returncode != 0: - logger.warning( - "Could not import iron-proxy signing key — skipping GPG verification (SHA-256 still enforced): %s", - imp.stderr.decode("utf-8", "replace")[:200], - ) + if (imp := _run([*gpg_base, "--import", str(pubkey_path)], timeout=60)).returncode != 0: + logger.warning("Could not import iron-proxy signing key — skipping GPG verification (SHA-256 still enforced): %s", imp.stderr.decode("utf-8", "replace")[:200]) return False - verify = _run([*gpg_base, "--verify", str(sig_path), str(checksum_path)], timeout=60) - if verify.returncode != 0: + if (verify := _run([*gpg_base, "--verify", str(sig_path), str(checksum_path)], timeout=60)).returncode != 0: raise RuntimeError( - "iron-proxy checksums.txt failed GPG signature verification — refusing to install (possible release-channel " - f"tampering). gpg: {verify.stderr.decode('utf-8', 'replace')[:300]}" + f"iron-proxy checksums.txt failed GPG signature verification — refusing to install (possible release-channel tampering). gpg: {verify.stderr.decode('utf-8', 'replace')[:300]}" ) logger.info("Verified iron-proxy checksums.txt GPG signature.") return True @@ -309,18 +280,14 @@ def _expected_sha256(checksum_file: Path, asset_name: str) -> str: def _sha256_file(path: Path) -> str: - h = hashlib.sha256() with open(path, "rb") as f: - for chunk in iter(lambda: f.read(65536), b""): - h.update(chunk) - return h.hexdigest() + return hashlib.file_digest(f, "sha256").hexdigest() def _pick_tar_member(tf: tarfile.TarFile, binary_name: str) -> tarfile.TarInfo: """Find the binary in the archive (flat or one dir deep); reject abs paths and ``..``.""" candidates = [ - m for m in tf.getmembers() - if m.isfile() and not m.name.startswith("/") and ".." not in Path(m.name).parts and Path(m.name).name == binary_name + m for m in tf.getmembers() if m.isfile() and not m.name.startswith("/") and ".." not in Path(m.name).parts and Path(m.name).name == binary_name ] if not candidates: raise RuntimeError(f"Could not find {binary_name} inside downloaded archive (members: {[m.name for m in tf.getmembers()[:5]]}...)") @@ -333,7 +300,6 @@ def _allowlisted_env() -> Dict[str, str]: def _run(argv: List[str], *, timeout: int, text: bool = False, **kwargs) -> "subprocess.CompletedProcess": - """``subprocess.run`` with captured output; argv[0] is always a trusted PATH/system binary.""" if text: kwargs.update(text=True, encoding="utf-8", errors="replace") return subprocess.run(argv, capture_output=True, timeout=timeout, stdin=subprocess.DEVNULL, **kwargs) # noqa: S603 @@ -341,23 +307,21 @@ def _run(argv: List[str], *, timeout: int, text: bool = False, **kwargs) -> "sub def iron_proxy_version(binary: Path) -> str: """``iron-proxy --version`` output, stripped and cached by path. Empty on failure.""" - key = str(binary) - if key in _VERSION_CACHE: + if (key := str(binary)) in _VERSION_CACHE: return _VERSION_CACHE[key] try: # Scrubbed env: a PATH-resolved binary must not see the host's API keys. res = _run([str(binary), "--version"], timeout=_RUN_TIMEOUT, text=True, env=_allowlisted_env()) except (OSError, subprocess.TimeoutExpired): return "" - out = (res.stdout or res.stderr or "").strip() - if out: # never cache empty output — it would poison status for the process lifetime + if out := (res.stdout or res.stderr or "").strip(): # never cache empty output — it would poison status for the process lifetime _VERSION_CACHE[key] = out return out def _write_private_file(path: Path, data: bytes) -> None: - """Create/truncate ``path`` 0o600 from the first byte (no chmod-after TOCTOU), O_NOFOLLOW - against a planted symlink, fchmod to tighten a pre-existing file.""" + """Create/truncate ``path`` 0o600 from the first byte (no chmod-after TOCTOU), O_NOFOLLOW against a planted + symlink, fchmod to tighten a pre-existing file.""" fd = os.open(str(path), os.O_WRONLY | os.O_CREAT | os.O_TRUNC | _O_NOFOLLOW, 0o600) try: with suppress(OSError, AttributeError): @@ -387,11 +351,8 @@ def ensure_ca_cert(*, force: bool = False) -> Tuple[Path, Path]: with tempfile.TemporaryDirectory(prefix="hermes-proxy-ca-") as tmpdir: tmp_key, tmp_crt = Path(tmpdir) / "ca.key", Path(tmpdir) / "ca.crt" _run(["openssl", "genrsa", "-out", str(tmp_key), "4096"], timeout=60, check=True) - _run([ - "openssl", "req", "-x509", "-new", "-nodes", "-key", str(tmp_key), "-sha256", "-days", "3650", - "-subj", "/CN=hermes iron-proxy CA", "-addext", "basicConstraints=critical,CA:TRUE", - "-addext", "keyUsage=critical,keyCertSign", "-out", str(tmp_crt), - ], timeout=60, check=True) + _run(["openssl", "req", "-x509", "-new", "-nodes", "-key", str(tmp_key), "-sha256", "-days", "3650", "-subj", "/CN=hermes iron-proxy CA", + "-addext", "basicConstraints=critical,CA:TRUE", "-addext", "keyUsage=critical,keyCertSign", "-out", str(tmp_crt)], timeout=60, check=True) # Key: stage 0o600 against a fresh inode, then atomically rename into place. key_staged = ca_key.with_suffix(ca_key.suffix + ".staged") key_staged.unlink(missing_ok=True) @@ -420,8 +381,7 @@ def _read_text_or_none(p: Path) -> Optional[str]: def ensure_management_token(*, force: bool = False) -> str: """Return the management-API bearer key (0600 at /management.token), minting on first call.""" p = _proxy_state_dir() / "management.token" - existing = None if force else _read_text_or_none(p) - if existing: + if not force and (existing := _read_text_or_none(p)): return existing token = mint_proxy_token(prefix="hermes-mgmt") _write_private_file(p, token.encode("utf-8")) @@ -437,17 +397,6 @@ def _yaml(): return None -def _load_proxy_yaml(cfg: Path): - """Parsed proxy.yaml, or ``{}`` when missing/unreadable/PyYAML absent.""" - yaml = _yaml() - if yaml is None: - return {} - try: - return yaml.safe_load(cfg.read_text(encoding="utf-8")) or {} - except (OSError, yaml.YAMLError): - return {} - - def _parse_listen(listen) -> Optional[Tuple[str, int]]: """``"host:port"`` -> ``(host, port)``; empty host means loopback.""" if not isinstance(listen, str) or ":" not in listen: @@ -461,8 +410,12 @@ def _parse_listen(listen) -> Optional[Tuple[str, int]]: def _config_listen(section: str, *keys: str, config_path: Optional[Path] = None) -> Optional[Tuple[str, int]]: - """``(host, port)`` from the first truthy ``proxy.yaml[section][key]``, or None.""" - block = _load_proxy_yaml(config_path or (_proxy_state_dir_ro() / "proxy.yaml")).get(section) or {} + """``(host, port)`` from the first truthy ``proxy.yaml[section][key]``, or None (also when file/PyYAML is missing).""" + yaml, data = _yaml(), {} + if yaml is not None: + with suppress(OSError, yaml.YAMLError): + data = yaml.safe_load((config_path or (_proxy_state_dir_ro() / "proxy.yaml")).read_text(encoding="utf-8")) or {} + block = data.get(section) or {} return _parse_listen(next((block[k] for k in keys if block.get(k)), "")) @@ -470,41 +423,28 @@ def _read_management_listen_from_config(config_path: Optional[Path] = None) -> O return _config_listen("management", "listen", config_path=config_path) -def _read_http_listen_from_config() -> Optional[Tuple[str, int]]: - """Sandbox-facing listener: ``tunnel_listen`` (CONNECT/MITM), falling back to ``http_listen`` - for configs written before the listener-role split.""" - return _config_listen("proxy", "tunnel_listen", "http_listen") - - def _probe_target() -> Tuple[str, int]: - """Configured bind host/port to probe — on Linux that's the docker bridge, where a loopback - connect would report a healthy daemon as down.""" - return _read_http_listen_from_config() or ("127.0.0.1", _DEFAULT_TUNNEL_PORT) + """Configured bind host/port to probe — on Linux the docker bridge, where a loopback connect would report a healthy + daemon as down. ``tunnel_listen`` (CONNECT/MITM) falls back to ``http_listen`` for pre-listener-role-split configs.""" + return _config_listen("proxy", "tunnel_listen", "http_listen") or ("127.0.0.1", _DEFAULT_TUNNEL_PORT) -# Management-API error status -> operator message (422 = validation rejected, running ruleset -# unchanged; 401 = daemon started with a different management.token). +# Management-API error status -> operator message (422 = validation rejected, ruleset unchanged; 401 = daemon started with another management.token). _RELOAD_HTTP_ERRORS = { 422: "iron-proxy rejected the new config (validation failed; the running ruleset is unchanged): {body}", - 401: "management API rejected our key (401). The running daemon was started with a different management.token — " - "run `hermes egress restart`.", + 401: "management API rejected our key (401). The running daemon was started with a different management.token — run `hermes egress restart`.", } def reload_proxy() -> bool: - """``POST /v1/reload`` (validation failures leave the running config untouched); RuntimeError - with an actionable message on any failure.""" - pid = _read_pid() - if not pid or not _pid_alive(pid): + """``POST /v1/reload`` (validation failures leave the running config untouched); actionable RuntimeError on any failure.""" + if not (pid := _read_pid()) or not _pid_alive(pid): raise RuntimeError("iron-proxy is not running — nothing to reload. Run `hermes egress start`.") - mgmt = _read_management_listen_from_config() - if mgmt is None: + if (mgmt := _read_management_listen_from_config()) is None: raise RuntimeError( - "The generated proxy.yaml has no management listener (written before reload support). " - "Re-run `hermes egress setup` and use `hermes egress restart` this one time." + "The generated proxy.yaml has no management listener (written before reload support). Re-run `hermes egress setup` and use `hermes egress restart` this one time." ) - token = _read_text_or_none(_proxy_state_dir_ro() / "management.token") - if not token: + if not (token := _read_text_or_none(_proxy_state_dir_ro() / "management.token")): raise RuntimeError("management.token is missing — re-run `hermes egress setup`, then `hermes egress restart`.") host, port = mgmt req = urllib.request.Request(f"http://{host}:{port}/v1/reload", method="POST", headers={"Authorization": f"Bearer {token}"}, data=b"") @@ -522,22 +462,19 @@ def reload_proxy() -> bool: except (urllib.error.URLError, OSError) as exc: # A daemon started from a pre-management config is alive but has no listener. raise RuntimeError( - f"could not reach the management API at {host}:{port} ({exc}). " - "If the daemon was started before reload support, run `hermes egress restart` once." + f"could not reach the management API at {host}:{port} ({exc}). If the daemon was started before reload support, run `hermes egress restart` once." ) from exc def _default_http_listen(tunnel_port: int) -> List[str]: - """Single bind (v0.39 allows one): docker bridge on Linux (what ``host.docker.internal`` - resolves to; loopback is unreachable from containers), loopback on Docker Desktop (VPNkit). - NEVER 0.0.0.0: a LAN peer with a leaked sandbox token could spend the operator's API quota.""" + """Single bind (v0.39 allows one): docker bridge on Linux (what ``host.docker.internal`` resolves to; loopback is + unreachable from containers), loopback on Docker Desktop (VPNkit). NEVER 0.0.0.0: a LAN peer with a leaked + sandbox token could spend the operator's API quota.""" if platform.system() == "Linux": - bridge_ip = _detect_docker_bridge_ip() - if bridge_ip and bridge_ip != "127.0.0.1": + if (bridge_ip := _detect_docker_bridge_ip()) and bridge_ip != "127.0.0.1": return [f"{bridge_ip}:{tunnel_port}"] logger.warning( - "No docker bridge (docker0) detected — binding iron-proxy to loopback only. " - "Docker sandboxes will NOT be able to reach the proxy until it is restarted with docker running." + "No docker bridge (docker0) detected — binding iron-proxy to loopback only. Docker sandboxes will NOT be able to reach the proxy until it is restarted with docker running." ) return [f"127.0.0.1:{tunnel_port}"] @@ -549,18 +486,11 @@ def _detect_docker_bridge_ip() -> Optional[str]: res = _run(["ip", "-4", "-o", "addr", "show", "docker0"], timeout=2, text=True) except (OSError, subprocess.TimeoutExpired): return None - if res.returncode != 0: - return None # Expected: ": docker0 inet 172.17.0.1/16 ..." — first inet token (per line) wins. - candidate = None - for parts in (line.split() for line in res.stdout.splitlines()): - if "inet" in parts[:-1]: - candidate = parts[parts.index("inet") + 1].split("/")[0] - break - if not candidate: - return None + lines = (line.split() for line in res.stdout.splitlines()) if res.returncode == 0 else () + candidate = next((parts[parts.index("inet") + 1].split("/")[0] for parts in lines if "inet" in parts[:-1]), None) try: - addr = ipaddress.IPv4Address(candidate) + addr = ipaddress.IPv4Address(candidate) # None/"" raise too except (ipaddress.AddressValueError, ValueError): return None if addr.is_unspecified or addr.is_loopback or addr.is_multicast or addr.is_reserved or addr.is_link_local or addr.is_global: @@ -570,38 +500,32 @@ def _detect_docker_bridge_ip() -> Optional[str]: def build_proxy_config( - *, mappings: List[TokenMapping], ca_cert: Path, ca_key: Path, tunnel_port: int = _DEFAULT_TUNNEL_PORT, - audit_log: Optional[Path] = None, allowed_hosts: Optional[List[str]] = None, - upstream_deny_cidrs: Optional[List[str]] = None, http_listen: Optional[List[str]] = None, + *, mappings: List[TokenMapping], ca_cert: Path, ca_key: Path, tunnel_port: int = _DEFAULT_TUNNEL_PORT, audit_log: Optional[Path] = None, + allowed_hosts: Optional[List[str]] = None, upstream_deny_cidrs: Optional[List[str]] = None, http_listen: Optional[List[str]] = None, ) -> Dict: - """iron-proxy YAML config dict (v0.39.0 schema). Real secrets come from iron-proxy's OWN env - (``source: {type: env}``); the sandbox never sees them. ``upstream_deny_cidrs=None`` = default - SSRF deny list, ``[]`` opts out. ``audit_log`` is forward-compat only (v0.39 rejects ``audit_path``).""" + """iron-proxy YAML config dict (v0.39.0 schema). Real secrets come from iron-proxy's OWN env (``source: {type: env}``); + the sandbox never sees them. ``upstream_deny_cidrs=None`` = default SSRF deny list, ``[]`` opts out. + ``audit_log`` is forward-compat only (v0.39 rejects ``audit_path``).""" hosts: List[str] = list(allowed_hosts or _DEFAULT_ALLOWED_HOSTS) - for m in mappings: - for h in m.upstream_hosts: - if h not in hosts: - hosts.append(h) + for h in (h for m in mappings for h in m.upstream_hosts): + if h not in hosts: + hosts.append(h) deny_cidrs = list(_DEFAULT_UPSTREAM_DENY_CIDRS if upstream_deny_cidrs is None else upstream_deny_cidrs) # Query scan covers ``?key=`` SDKs; body inspection deliberately off. ``require`` fails # closed: an allowlisted-host request WITHOUT the proxy token is rejected, so a real key sent # directly can't cross the boundary. - secrets_rules = [ - { - "source": {"type": "env", "var": m.real_env_name}, - "replace": { - "proxy_value": m.proxy_token, "match_headers": list(m.match_headers or ("Authorization",)), - "match_query": True, "match_body": False, "require": True, - }, - "rules": [{"host": h} for h in m.upstream_hosts], - } - for m in mappings - ] + secrets_rules = [{ + "source": {"type": "env", "var": m.real_env_name}, + "replace": { + "proxy_value": m.proxy_token, "match_headers": list(m.match_headers or ("Authorization",)), + "match_query": True, "match_body": False, "require": True, + }, + "rules": [{"host": h} for h in m.upstream_hosts], + } for m in mappings] # ONE string per listener field. tunnel_listen is the CONNECT+MITM listener sandboxes reach via # HTTPS_PROXY (a CONNECT to http_listen is forwarded upstream and 400s); http_listen is plain-HTTP # forward on tunnel_port+1. - listens = list(http_listen) if http_listen else _default_http_listen(tunnel_port) - primary_listen = listens[0] if listens else f"127.0.0.1:{tunnel_port}" + primary_listen = (list(http_listen) if http_listen else _default_http_listen(tunnel_port) or [f"127.0.0.1:{tunnel_port}"])[0] bind_host = primary_listen.rsplit(":", 1)[0] or "127.0.0.1" return { # Required by the parser; tunnel-only mode never binds an exposed DNS port. @@ -618,17 +542,14 @@ def build_proxy_config( # Loopback only: sandboxes must never reach the management surface. "management": {"listen": f"127.0.0.1:{tunnel_port + _MGMT_PORT_OFFSET}", "api_key_env": _MGMT_API_KEY_ENV}, "tls": {"ca_cert": str(ca_cert), "ca_key": str(ca_key), "cert_cache_size": 1000, "leaf_cert_expiry_hours": 168}, - "transforms": [ - {"name": "allowlist", "config": {"domains": hosts}}, - {"name": "secrets", "config": {"secrets": secrets_rules}}, - ], + "transforms": [{"name": "allowlist", "config": {"domains": hosts}}, {"name": "secrets", "config": {"secrets": secrets_rules}}], "log": {"level": "info"}, } def _open_private_append(path: Path, *, strict_chmod: bool) -> int: - """Open ``path`` O_APPEND|O_CREAT 0o600 with O_NOFOLLOW (planted symlinks refused); fchmod - tightens a pre-existing file (failure fatal iff ``strict_chmod``). Raises OSError; caller owns the fd.""" + """O_APPEND|O_CREAT 0o600 + O_NOFOLLOW (planted symlinks refused); fchmod tightens a pre-existing + file (failure fatal iff ``strict_chmod``). Raises OSError; caller owns the fd.""" fd = os.open(str(path), os.O_WRONLY | os.O_CREAT | os.O_APPEND | _O_NOFOLLOW, 0o600) try: os.fchmod(fd, 0o600) @@ -640,20 +561,17 @@ def _open_private_append(path: Path, *, strict_chmod: bool) -> int: def ensure_audit_log(audit_path: Path) -> None: - """Pre-create the audit log 0o600 (forward-compat: v0.39 never writes it). RuntimeError on any - OSError (planted symlink, immutable dir, full disk).""" + """Pre-create the audit log 0o600 (forward-compat: v0.39 never writes it); RuntimeError on any OSError.""" try: os.close(_open_private_append(audit_path, strict_chmod=True)) except OSError as exc: raise RuntimeError( - f"Refusing to start: could not pre-create audit log {audit_path} with restrictive permissions ({exc}). " - f"Move or chmod any existing file at that path and retry." + f"Refusing to start: could not pre-create audit log {audit_path} with restrictive permissions ({exc}). Move or chmod any existing file at that path and retry." ) from exc def _write_state_file_atomic(state: Path, name: str, dump) -> Path: - """0600 temp file + atomic replace: the file holds proxy tokens, and chmod-after-replace would - leave a world-readable TOCTOU window.""" + """0600 temp file + atomic replace: the file holds proxy tokens; chmod-after-replace would be a world-readable TOCTOU window.""" tmp_path = state / f".{name}.tmp" with open(tmp_path, "w", encoding="utf-8") as f: dump(f) @@ -664,33 +582,23 @@ def _write_state_file_atomic(state: Path, name: str, dump) -> Path: def write_proxy_config(config: Dict) -> Path: """Serialize the config dict to ``/proxy/proxy.yaml`` (safe_dump, no Python tags).""" - yaml = _yaml() - if yaml is None: + if (yaml := _yaml()) is None: raise RuntimeError("PyYAML is required to write the iron-proxy config but is not installed.") - return _write_state_file_atomic( - _proxy_state_dir(), "proxy.yaml", lambda f: yaml.safe_dump(config, f, default_flow_style=False, sort_keys=False), - ) + return _write_state_file_atomic(_proxy_state_dir(), "proxy.yaml", lambda f: yaml.safe_dump(config, f, default_flow_style=False, sort_keys=False)) def write_mappings(mappings: List[TokenMapping]) -> Path: """Persist sandbox-visible tokens to ``mappings.json`` (read by the Docker backend, not iron-proxy).""" - payload = { - "version": 1, - "tokens": [ - { - "proxy_token": m.proxy_token, "env_name": m.real_env_name, "upstream_hosts": list(m.upstream_hosts), - "match_headers": list(m.match_headers), "alias_env_names": list(m.alias_env_names), - } - for m in mappings - ], - } + payload = {"version": 1, "tokens": [{ + "proxy_token": m.proxy_token, "env_name": m.real_env_name, "upstream_hosts": list(m.upstream_hosts), + "match_headers": list(m.match_headers), "alias_env_names": list(m.alias_env_names), + } for m in mappings]} return _write_state_file_atomic(_proxy_state_dir(), "mappings.json", lambda f: json.dump(payload, f, indent=2)) def load_mappings() -> List[TokenMapping]: """Read mappings.json, if it exists. Empty list on any error.""" - f = _proxy_state_dir() / "mappings.json" - if not f.exists(): + if not (f := _proxy_state_dir() / "mappings.json").exists(): return [] try: payload = json.loads(f.read_text(encoding="utf-8")) @@ -699,28 +607,19 @@ def load_mappings() -> List[TokenMapping]: return [] out: List[TokenMapping] = [] for item in payload.get("tokens", []): - try: # pre-header-auth files load with the bearer defaults they were written under - out.append(TokenMapping( - item["proxy_token"], item["env_name"], tuple(item.get("upstream_hosts") or ()), - tuple(item.get("match_headers") or ("Authorization",)), tuple(item.get("alias_env_names") or ()), - )) - except (KeyError, TypeError): - continue + with suppress(KeyError, TypeError): # pre-header-auth files load with the bearer defaults they were written under + out.append(TokenMapping(item["proxy_token"], item["env_name"], tuple(item.get("upstream_hosts") or ()), + tuple(item.get("match_headers") or ("Authorization",)), tuple(item.get("alias_env_names") or ()))) return out -def _env_names(available_env_names: Optional[List[str]]) -> set: - """Explicit override (Bitwarden adapter) or the non-empty names in the host env.""" - return set(available_env_names) if available_env_names is not None else {k for k, v in os.environ.items() if v} - - def discover_provider_mappings(*, available_env_names: Optional[List[str]] = None) -> List[TokenMapping]: - """One TokenMapping per known provider whose env var is set (bearer providers first). Canonical - OR any alias present -> ONE mapping on the canonical name (the subprocess-env builder mirrors aliases).""" - names = _env_names(available_env_names) + """One TokenMapping per known provider whose env var is set (bearer providers first). Canonical OR any alias + present -> ONE mapping on the canonical name (the subprocess-env builder mirrors aliases). + ``available_env_names`` (Bitwarden adapter) overrides the non-empty names in the host env.""" + names = set(available_env_names) if available_env_names is not None else {k for k, v in os.environ.items() if v} specs = [(n, h, ("Authorization",), ()) for n, h in _BEARER_PROVIDERS.items()] + [ - (n, tuple(s["hosts"]), tuple(s["match_headers"]), tuple(s.get("aliases") or ())) - for n, s in _HEADER_AUTH_PROVIDERS.items() + (n, tuple(s["hosts"]), tuple(s["match_headers"]), tuple(s.get("aliases") or ())) for n, s in _HEADER_AUTH_PROVIDERS.items() ] return [ TokenMapping(mint_proxy_token(prefix=env_name.lower().replace("_api_key", "")), env_name, hosts, headers, aliases) @@ -731,18 +630,15 @@ def discover_provider_mappings(*, available_env_names: Optional[List[str]] = Non def discover_uncovered_providers(*, available_env_names: Optional[List[str]] = None) -> List[str]: """Env names of recognized providers the proxy can't swap (SigV4 / SDK-minted OAuth).""" - names = _env_names(available_env_names) + names = set(available_env_names) if available_env_names is not None else {k for k, v in os.environ.items() if v} return [n for n in _NON_BEARER_PROVIDERS if n in names] def merge_mappings(*, existing: List[TokenMapping], discovered: List[TokenMapping], rotate: bool = False) -> List[TokenMapping]: - """Tokens already in ``existing`` are preserved (containers baked with them keep working) while - hosts/headers/aliases refresh from ``discovered``; ``rotate=True`` re-mints; undiscovered providers drop.""" + """Existing tokens are preserved (containers baked with them keep working), hosts/headers/aliases refresh + from ``discovered``; ``rotate=True`` re-mints; undiscovered providers drop.""" by_name = {} if rotate else {m.real_env_name: m for m in existing} - return [ - replace(d, proxy_token=by_name[d.real_env_name].proxy_token) if d.real_env_name in by_name else d - for d in discovered - ] + return [replace(d, proxy_token=by_name[d.real_env_name].proxy_token) if d.real_env_name in by_name else d for d in discovered] def _pidfile() -> Path: @@ -751,8 +647,8 @@ def _pidfile() -> Path: def _read_pid() -> Optional[int]: try: - pid = int((_proxy_state_dir_ro() / "iron-proxy.pid").read_text(encoding="utf-8").strip()) - except (OSError, ValueError): + pid = int(_read_text_or_none(_proxy_state_dir_ro() / "iron-proxy.pid") or "") + except ValueError: return None return pid if pid > 0 else None @@ -764,20 +660,18 @@ def _pid_proc_starttime(pid: int) -> Optional[str]: except OSError: return None # comm may contain spaces/parens, so split after the LAST ")"; field 22 -> tail index 19. - rparen = text.rfind(")") - fields = text[rparen + 1:].split() if rparen >= 0 else [] + fields = text[rparen + 1:].split() if (rparen := text.rfind(")")) >= 0 else [] return fields[19] if len(fields) > 19 else None def _persisted_nonce_path() -> Path: - """On-disk nonce sibling of the pidfile, so stop/status in a later CLI process can still - defeat PID recycling.""" + """On-disk nonce sibling of the pidfile, so stop/status in a later CLI process can still defeat PID recycling.""" return _proxy_state_dir_ro() / "iron-proxy.nonce" def _read_persisted_nonce() -> Optional[str]: - """Nonce from disk, or None if missing/unreadable/empty/not owned by us (callers fall back to - argv0 matching). O_NOFOLLOW: this read decides whether stop_proxy SIGKILLs a PID.""" + """Nonce from disk, or None if missing/unreadable/empty/not owned by us (callers fall back to argv0 + matching). O_NOFOLLOW: this read decides whether stop_proxy SIGKILLs a PID.""" try: fd = os.open(str(_persisted_nonce_path()), os.O_RDONLY | _O_NOFOLLOW) except OSError: @@ -791,9 +685,9 @@ def _read_persisted_nonce() -> Optional[str]: def _pid_alive(pid: int) -> bool: - """True iff ``pid`` is alive AND an iron-proxy process. PID-reuse defense, in priority order: - nonce in /proc//environ, argv[0] basename in /proc//cmdline, ``ps -o comm=`` basename - (a loose ``"iron-proxy" in cmdline`` would hit ``tail iron-proxy.log``).""" + """True iff ``pid`` is alive AND an iron-proxy process. PID-reuse defense, in priority order: nonce in + /proc//environ, argv[0] basename in /proc//cmdline, ``ps -o comm=`` basename (a loose + ``"iron-proxy" in cmdline`` would hit ``tail iron-proxy.log``).""" if pid <= 0: return False try: @@ -815,10 +709,8 @@ def _pid_alive(pid: int) -> bool: if any(f"{_HERMES_IRON_PROXY_NONCE_ENV}={n}".encode() in env_bytes for n in nonce_candidates): return True with suppress(OSError): - cmdline_path = Path(f"/proc/{pid}/cmdline") - if cmdline_path.exists(): - argv0 = cmdline_path.read_bytes().split(b"\x00")[0].decode("utf-8", errors="ignore") - return os.path.basename(argv0).startswith("iron-proxy") + if (cmdline_path := Path(f"/proc/{pid}/cmdline")).exists(): + return os.path.basename(cmdline_path.read_bytes().split(b"\x00")[0].decode("utf-8", errors="ignore")).startswith("iron-proxy") with suppress(OSError, subprocess.TimeoutExpired): # macOS / non-Linux fallback res = _run(["ps", "-p", str(pid), "-o", "comm="], timeout=2, text=True) if res.returncode == 0: @@ -831,22 +723,17 @@ def start_proxy( *, binary: Optional[Path] = None, config_path: Optional[Path] = None, extra_env: Optional[Dict[str, str]] = None, install_if_missing: bool = True, refresh_secrets_from_bitwarden: bool = False, bitwarden_config: Optional[Dict] = None, ) -> ProxyStatus: - """Spawn iron-proxy as a managed background subprocess (idempotent if already running). - ``refresh_secrets_from_bitwarden`` re-fetches secrets from BWS — the ``credential_source: bitwarden`` rotation promise.""" + """Spawn iron-proxy as a managed background subprocess (idempotent if already running). ``refresh_secrets_from_bitwarden`` + re-fetches secrets from BWS — the ``credential_source: bitwarden`` rotation promise.""" global _proxy_nonce - existing = _read_pid() - if existing and _pid_alive(existing): + if (existing := _read_pid()) and _pid_alive(existing): return get_status() - bin_path = binary or find_iron_proxy(install_if_missing=install_if_missing) - if bin_path is None: + if (bin_path := binary or find_iron_proxy(install_if_missing=install_if_missing)) is None: raise RuntimeError("iron-proxy binary not available — run `hermes egress install`.") - cfg = config_path or (_proxy_state_dir() / "proxy.yaml") - if not cfg.exists(): + if not (cfg := config_path or (_proxy_state_dir() / "proxy.yaml")).exists(): raise RuntimeError(f"iron-proxy config not found at {cfg}. Run `hermes egress setup` first.") # Minimal env: os.environ.copy() would expose every operator secret via /proc//environ. - env = _build_proxy_subprocess_env( - extra_env=extra_env, refresh_from_bitwarden=refresh_secrets_from_bitwarden, bitwarden_config=bitwarden_config, - ) + env = _build_proxy_subprocess_env(extra_env=extra_env, refresh_from_bitwarden=refresh_secrets_from_bitwarden, bitwarden_config=bitwarden_config) # v0.39 validates api_key_env is non-empty when management.listen is set. if _read_management_listen_from_config(cfg) is not None: env[_MGMT_API_KEY_ENV] = ensure_management_token() @@ -870,14 +757,12 @@ def start_proxy( pidfile.unlink(missing_ok=True) return RuntimeError(f"{msg}Last log lines:\n{tail}") - def _exited_error() -> RuntimeError: - return _abort(f"iron-proxy exited immediately (code {proc.returncode}). ", kill=False) - def _interrupt_handler(_signum, _frame): # pragma: no cover - signal path _kill_and_wait(proc, grace_seconds=2) # Ctrl-C while waiting must not leak an orphan holding the port pidfile.unlink(missing_ok=True) raise KeyboardInterrupt() + _exited_error = lambda: _abort(f"iron-proxy exited immediately (code {proc.returncode}). ", kill=False) # noqa: E731 # Probe the CONFIGURED bind host (on Linux the docker bridge, where loopback never connects). probe_host, tunnel_port = _probe_target() with _interrupt_guard(_interrupt_handler): @@ -894,7 +779,7 @@ def start_proxy( @contextmanager def _interrupt_guard(handler): - """Route SIGINT/SIGTERM to ``handler`` for the block (POSIX main thread only), restoring the previous handlers after.""" + """Route SIGINT/SIGTERM to ``handler`` for the block (POSIX main thread only); previous handlers restored after.""" if platform.system() == "Windows" or threading.current_thread() is not threading.main_thread(): yield return @@ -907,22 +792,21 @@ def _interrupt_guard(handler): def _spawn_daemon(bin_path: Path, cfg: Path, env: Dict[str, str], log_path: Path) -> "subprocess.Popen": - """Popen with stdout/stderr appended to ``log_path`` (0o600 from the first byte, O_NOFOLLOW so a - planted symlink e.g. to authorized_keys can't receive output, owner-checked). Our log fd is - closed right after Popen — the child has its dup.""" + """Popen with stdout/stderr appended to ``log_path`` (0o600 from the first byte, O_NOFOLLOW so a planted + symlink e.g. to authorized_keys can't receive output, owner-checked). Our log fd closes after Popen — the child has its dup.""" try: log_fd = _open_private_append(log_path, strict_chmod=False) except OSError as exc: raise RuntimeError(f"Refusing to write iron-proxy log {log_path}: {exc}. Remove that path manually and retry.") from exc if not _fd_owned_by_us(log_fd): - st = os.fstat(log_fd) + uid = os.fstat(log_fd).st_uid os.close(log_fd) - raise RuntimeError(f"iron-proxy log {log_path} has unexpected owner uid={st.st_uid}; refusing to write.") + raise RuntimeError(f"iron-proxy log {log_path} has unexpected owner uid={uid}; refusing to write.") try: # start_new_session is POSIX-only (Windows isn't supported anyway — no upstream binary). - session_kw = {} if platform.system() == "Windows" else {"start_new_session": True} return subprocess.Popen( # noqa: S603 - [str(bin_path), "-config", str(cfg)], env=env, stdin=subprocess.DEVNULL, stdout=log_fd, stderr=subprocess.STDOUT, **session_kw, + [str(bin_path), "-config", str(cfg)], env=env, stdin=subprocess.DEVNULL, stdout=log_fd, stderr=subprocess.STDOUT, + **({} if platform.system() == "Windows" else {"start_new_session": True}), ) except OSError as exc: raise RuntimeError(f"failed to spawn iron-proxy: {exc}") from exc @@ -932,8 +816,7 @@ def _spawn_daemon(bin_path: Path, cfg: Path, env: Dict[str, str], log_path: Path def _await_listening(proc: "subprocess.Popen", host: str, port: int, *, on_exit) -> bool: - """Poll until ``host:port`` accepts or the grace window lapses (do-while: at least one check - even with a 0s window). Raises ``on_exit()`` if the child dies meanwhile.""" + """Poll until ``host:port`` accepts or the grace window lapses (do-while: >=1 check even at 0s); raises ``on_exit()`` if the child dies.""" deadline = time.time() + _STARTUP_GRACE_SECONDS while True: if proc.poll() is not None: @@ -946,17 +829,15 @@ def _await_listening(proc: "subprocess.Popen", host: str, port: int, *, on_exit) def _write_pidfile_safely(pidfile: Path, pid: int) -> None: - """O_EXCL + O_NOFOLLOW + ownership check, then persist the nonce. An existing pidfile is either - a concurrent start (fail cleanly) or a stale crash leftover (unlink and retry once).""" + """O_EXCL + O_NOFOLLOW + ownership check, then persist the nonce. An existing pidfile is either a concurrent + start (fail cleanly) or a stale crash leftover (unlink and retry once).""" open_flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | _O_NOFOLLOW try: fd = os.open(str(pidfile), open_flags, 0o600) except FileExistsError: - existing_pid = _read_pid() - if existing_pid and _pid_alive(existing_pid): + if (existing_pid := _read_pid()) and _pid_alive(existing_pid): raise RuntimeError( - f"Another iron-proxy start appears to be in progress (pidfile {pidfile} -> pid {existing_pid}). " - f"Run `hermes egress stop` if that proxy is stuck." + f"Another iron-proxy start appears to be in progress (pidfile {pidfile} -> pid {existing_pid}). Run `hermes egress stop` if that proxy is stuck." ) pidfile.unlink(missing_ok=True) fd = os.open(str(pidfile), open_flags, 0o600) @@ -976,7 +857,6 @@ def _write_pidfile_safely(pidfile: Path, pid: int) -> None: def _kill_and_wait(proc: "subprocess.Popen", *, grace_seconds: int = 2) -> None: - """Best-effort SIGTERM → wait → SIGKILL for a child we own.""" try: proc.terminate() except OSError: @@ -993,18 +873,16 @@ def _kill_and_wait(proc: "subprocess.Popen", *, grace_seconds: int = 2) -> None: def _build_proxy_subprocess_env( *, extra_env: Optional[Dict[str, str]] = None, refresh_from_bitwarden: bool = False, bitwarden_config: Optional[Dict] = None, ) -> Dict[str, str]: - """Allowlisted infra vars + the secrets named in mappings. With ``refresh_from_bitwarden`` and a - populated ``bitwarden_config`` secrets come from BWS (the rotation guarantee); without - ``allow_env_fallback`` any BWS shortfall fails closed instead of keeping stale host-env values.""" - env = _allowlisted_env() - parent = os.environ + """Allowlisted infra vars + the secrets named in mappings. With ``refresh_from_bitwarden`` and a populated + ``bitwarden_config`` secrets come from BWS (the rotation guarantee); without ``allow_env_fallback`` any BWS + shortfall fails closed instead of keeping stale host-env values.""" + env, parent = _allowlisted_env(), os.environ # Forward ONLY mapped secrets; the rule is keyed on the canonical name, so mirror an alias value into it. mappings = load_mappings() needed = {m.real_env_name for m in mappings} alias_sources = {m.real_env_name: m.alias_env_names for m in mappings if m.alias_env_names} for name in needed: - source = name if name in parent else next((a for a in alias_sources.get(name, ()) if parent.get(a)), None) - if source is not None: + if (source := name if name in parent else next((a for a in alias_sources.get(name, ()) if parent.get(a)), None)) is not None: env[name] = parent[source] if refresh_from_bitwarden and bitwarden_config: _refresh_secrets_from_bitwarden(env, needed, bitwarden_config, bool(bitwarden_config.get("allow_env_fallback"))) @@ -1017,16 +895,17 @@ def _build_proxy_subprocess_env( return env -def _bitwarden_shortfall(allow_env_fallback: bool, error: str, warning: str, *args) -> None: - """Raise ``error`` unless the operator opted into the legacy host-env fallback (then log ``warning``).""" +def _bitwarden_shortfall(allow_env_fallback: bool, error: str, warning: str, *args, cause: Optional[BaseException] = None) -> None: + """Raise ``error`` (chained to ``cause`` when given) unless the operator opted into the legacy host-env fallback (then log ``warning``).""" if not allow_env_fallback: + if cause is not None: + raise RuntimeError(error) from cause raise RuntimeError(error) logger.warning(warning, *args) def _refresh_secrets_from_bitwarden(env: Dict[str, str], needed: set, bitwarden_config: Dict, allow_env_fallback: bool) -> None: - """Overwrite ``env[needed]`` with fresh (uncached) BWS values; only mapped names are injected so - unrelated BWS secrets never leak into the daemon env.""" + """Overwrite ``env[needed]`` with fresh (uncached) BWS values; only mapped names are injected so unrelated BWS secrets never leak.""" try: # Lazy: the bitwarden module isn't importable in every install. from agent.secret_sources import bitwarden as bw @@ -1038,40 +917,35 @@ def _refresh_secrets_from_bitwarden(env: Dict[str, str], needed: set, bitwarden_ allow_env_fallback, "credential_source=bitwarden but the access-token env or project_id is empty. Either set both, switch to " "credential_source: env, or set `proxy.allow_env_fallback: true` to opt into the legacy fallback behaviour.", - "credential_source=bitwarden but access-token env or project_id is empty — " - "proxy will fall back to parent env (allow_env_fallback=true).", + "credential_source=bitwarden but access-token env or project_id is empty — proxy will fall back to parent env (allow_env_fallback=true).", ) return secrets, warnings = bw.fetch_bitwarden_secrets(access_token=access_token, project_id=project_id, cache_ttl_seconds=0, use_cache=False) except ImportError as exc: # A dependency vanishing between setup and restart must not silently degrade. - if not allow_env_fallback: - raise RuntimeError( - "Bitwarden refresh module unavailable at proxy start (credential_source=bitwarden with " - "proxy.allow_env_fallback: false). Either fix the import, switch to credential_source: env, or set " - "`proxy.allow_env_fallback: true` to opt into the legacy fallback behaviour." - ) from exc - logger.warning("Bitwarden refresh module unavailable at proxy start, falling back to parent env (allow_env_fallback=true): %s", exc) + _bitwarden_shortfall( + allow_env_fallback, + "Bitwarden refresh module unavailable at proxy start (credential_source=bitwarden with " + "proxy.allow_env_fallback: false). Either fix the import, switch to credential_source: env, or set " + "`proxy.allow_env_fallback: true` to opt into the legacy fallback behaviour.", + "Bitwarden refresh module unavailable at proxy start, falling back to parent env (allow_env_fallback=true): %s", exc, cause=exc, + ) return missing = sorted(needed - set(secrets)) - for n in needed: - if n in secrets: - env[n] = secrets[n] + env.update((n, secrets[n]) for n in needed if n in secrets) if missing: _bitwarden_shortfall( allow_env_fallback, f"Bitwarden refresh did not return secrets for {missing}. Either add the secrets to your BWS project, switch to " f"credential_source: env via `hermes egress setup --no-bitwarden`, or set `proxy.allow_env_fallback: true` in " f"config.yaml to opt into the legacy host-env fallback.", - "Bitwarden refresh did not return secrets for %s — falling back to host env for those names (allow_env_fallback=true).", - missing, + "Bitwarden refresh did not return secrets for %s — falling back to host env for those names (allow_env_fallback=true).", missing, ) if warnings: # log only the count: the taint analyzer can't tell bws status text is non-secret logger.warning("Bitwarden refresh produced %d warning(s); run `hermes secrets bitwarden status` for detail.", len(warnings)) def _forget_daemon() -> None: - """Best-effort removal of pidfile + persisted nonce, and drop the in-process nonce.""" global _proxy_nonce _pidfile().unlink(missing_ok=True) with suppress(OSError): @@ -1080,7 +954,7 @@ def _forget_daemon() -> None: def stop_proxy() -> bool: - """Stop the managed iron-proxy. Returns True if it was running.""" + """Returns True if it was running.""" pid = _read_pid() if not pid or not _pid_alive(pid): _forget_daemon() @@ -1100,8 +974,7 @@ def stop_proxy() -> bool: time.sleep(0.1) else: starttime_after = _pid_proc_starttime(pid) - recycled = (starttime_before is not None and starttime_after is not None and starttime_before != starttime_after) or not _pid_alive(pid) - if recycled: + if (starttime_before is not None and starttime_after is not None and starttime_before != starttime_after) or not _pid_alive(pid): logger.warning("iron-proxy pid=%s appears recycled before SIGKILL; not killing.", pid) else: with suppress(ProcessLookupError): @@ -1115,22 +988,17 @@ def get_status() -> ProxyStatus: """Snapshot the proxy state without side effects (called per Docker container create).""" status = ProxyStatus() probe_host, status.tunnel_port = _probe_target() - binary = find_iron_proxy(install_if_missing=False) - if binary: - status.binary_path = binary - status.binary_version = iron_proxy_version(binary) + if binary := find_iron_proxy(install_if_missing=False): + status.binary_path, status.binary_version = binary, iron_proxy_version(binary) state = _proxy_state_dir_ro() cfg, ca = state / "proxy.yaml", state / "ca.crt" - status.config_path = cfg if cfg.exists() else None - status.ca_cert_path = ca if ca.exists() else None - pid = _read_pid() - if pid and _pid_alive(pid): + status.config_path, status.ca_cert_path = (cfg if cfg.exists() else None), (ca if ca.exists() else None) + if (pid := _read_pid()) and _pid_alive(pid): status.pid, status.listening = pid, _port_listening(probe_host, status.tunnel_port) return status def _port_listening(host: str, port: int) -> bool: - """Cheap TCP connect probe — True iff something accepts on host:port.""" import socket try: with socket.create_connection((host, port), timeout=0.5): @@ -1149,7 +1017,6 @@ def _tail_log(path: Path, *, lines: int = 20) -> str: def _reset_for_tests() -> None: - """Clear the module's mutable globals (``_VERSION_CACHE``, ``_proxy_nonce``).""" global _proxy_nonce _VERSION_CACHE.clear() _proxy_nonce = None