From 8aba2fe9f2d8f332d7b04dddda2bcc12de0fbe09 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 19:25:11 -0700 Subject: [PATCH] refactor(agent/secrets): collapse defensive branches (alias target, ordered sources, zip containment, find_op, query-string sub) --- agent/redact.py | 39 ++++++++++------------------- agent/secret_scope.py | 7 ++---- agent/secret_sources/bitwarden.py | 13 +++------- agent/secret_sources/onepassword.py | 9 +++---- agent/secret_sources/registry.py | 32 ++++++++--------------- 5 files changed, 34 insertions(+), 66 deletions(-) diff --git a/agent/redact.py b/agent/redact.py index ed0677cdff..d2a1d22ebe 100644 --- a/agent/redact.py +++ b/agent/redact.py @@ -373,11 +373,10 @@ def _redact_query_string(query: str) -> str: """Replace values of sensitive ``k=v&k=v`` params with ``***``; others pass through.""" if not query: return query - parts = [] - for pair in query.split("&"): - key, sep, _value = pair.partition("=") - parts.append(f"{key}=***" if sep and key.lower() in _SENSITIVE_QUERY_PARAMS else pair) - return "&".join(parts) + return "&".join( + f"{key}=***" if sep and key.lower() in _SENSITIVE_QUERY_PARAMS else pair + for pair in query.split("&") for key, sep, _ in (pair.partition("="),) + ) def _canonical_url_param_name(name: str) -> str: @@ -394,19 +393,12 @@ def _canonical_url_param_name(name: str) -> str: def _redact_strict_url_credentials(text: str) -> str: """Strict egress-boundary redaction of URL credentials (absolute, relative and network references); preserves keys, separators, public params, hosts, paths.""" - def _redact_param(match: re.Match) -> str: - if _canonical_url_param_name(match.group(2)) not in _SENSITIVE_QUERY_PARAMS: - return match.group(0) - return f"{match.group(1)}{match.group(2)}=***" - - def _redact_userinfo(match: re.Match) -> str: - userinfo = match.group(2) - if ":" in userinfo: - return f"{match.group(1)}{userinfo.partition(':')[0]}:***@" - return f"{match.group(1)}***@" - - text = _STRICT_URL_PARAM_RE.sub(_redact_param, text) - return _STRICT_URL_USERINFO_RE.sub(_redact_userinfo, text) + text = _STRICT_URL_PARAM_RE.sub( + lambda m: f"{m.group(1)}{m.group(2)}=***" + if _canonical_url_param_name(m.group(2)) in _SENSITIVE_QUERY_PARAMS else m.group(0), text) + return _STRICT_URL_USERINFO_RE.sub( + lambda m: f"{m.group(1)}{m.group(2).partition(':')[0]}:***@" if ":" in m.group(2) else f"{m.group(1)}***@", + text) def redact_cdp_url(value: object) -> str: @@ -518,8 +510,7 @@ def redact_sensitive_text(text: str, *, force: bool = False, code_file: bool = F """ if text is None: return None - if not isinstance(text, str): - text = str(text) + text = text if isinstance(text, str) else str(text) if not text or not (force or _REDACT_ENABLED): return text code_file = code_file or file_read @@ -637,13 +628,9 @@ def _extract_literal_prefix(pattern: str) -> str: def _skip_char_class(pattern: str, i: int) -> int: """Given ``pattern[i] == "["``, return the index just past the closing ``]``.""" - i += 1 - if i < len(pattern) and pattern[i] == "]": - i += 1 + i += 2 if pattern[i + 1:i + 2] == "]" else 1 # a leading "]" is literal while i < len(pattern) and pattern[i] != "]": - if pattern[i] == "\\": - i += 1 - i += 1 + i += 2 if pattern[i] == "\\" else 1 return i diff --git a/agent/secret_scope.py b/agent/secret_scope.py index 7cfa956a15..9831aadb2c 100644 --- a/agent/secret_scope.py +++ b/agent/secret_scope.py @@ -118,14 +118,12 @@ def get_secret(name: str, default: Optional[str] = None) -> Optional[str]: """ if _is_global_env(name): return _environ_or(name, default) - scope = _SECRET_SCOPE.get() if scope is not None: val = scope.get(name) if val is not None: return val return default if _MULTIPLEX_ACTIVE else _environ_or(name, default) - if _MULTIPLEX_ACTIVE: raise UnscopedSecretError( f"get_secret({name!r}) called with no profile secret scope active " @@ -190,11 +188,10 @@ def build_profile_secret_scope(hermes_home: Path) -> Dict[str, str]: """Build a profile's secret mapping from ``/.env`` plus its external secret sources. Global vars are NOT copied in — ``get_secret`` reads those from ``os.environ`` — so the scope holds only profile secrets.""" - home = Path(hermes_home) - secrets = load_env_file(home / ".env") + secrets = load_env_file(Path(hermes_home) / ".env") try: from hermes_cli.env_loader import get_secret_source_values - external_secrets = get_secret_source_values(home) + external_secrets = get_secret_source_values(Path(hermes_home)) except Exception: external_secrets = {} secrets.update((k, v) for k, v in external_secrets.items() if not _is_global_env(k)) diff --git a/agent/secret_sources/bitwarden.py b/agent/secret_sources/bitwarden.py index 7aae5323b8..287e6170e1 100644 --- a/agent/secret_sources/bitwarden.py +++ b/agent/secret_sources/bitwarden.py @@ -222,10 +222,10 @@ def _safe_extract_member(zf: zipfile.ZipFile, member: str, dest_dir: Path) -> Pa dest_root = os.path.realpath(dest_dir) target = os.path.realpath(os.path.join(dest_root, member)) try: # commonpath raises for e.g. different Windows drives — treat as escape - contained = os.path.commonpath([dest_root, target]) == dest_root + contained = os.path.commonpath([dest_root, target]) == dest_root and target != dest_root except ValueError: contained = False - if not contained or target == dest_root: + if not contained: raise RuntimeError(f"Refusing to extract unsafe archive member {member!r}: " f"it escapes the extraction directory") zf.extract(member, dest_root) @@ -385,18 +385,13 @@ def _summarize_bws_stderr(raw: str) -> str: """Reduce a bws (color-eyre) error dump to its numbered cause lines joined with ``; `` (dropping ``Location:``/``Backtrace`` on); raw text if unrecognized.""" text = raw.replace("\x1b", "").strip() - if not text: - return text causes: List[str] = [] for line in text.splitlines(): stripped = line.strip() if stripped.startswith(("Location:", "Backtrace omitted", "Run with ")): break - if stripped in ("", "Error:"): - continue - stripped = re.sub(r"^\d+:\s*", "", stripped) - if stripped: - causes.append(stripped) + if stripped not in ("", "Error:") and (cause := re.sub(r"^\d+:\s*", "", stripped)): + causes.append(cause) return "; ".join(causes) if causes else text diff --git a/agent/secret_sources/onepassword.py b/agent/secret_sources/onepassword.py index 5bce7358fb..ed39bb63e9 100644 --- a/agent/secret_sources/onepassword.py +++ b/agent/secret_sources/onepassword.py @@ -114,11 +114,10 @@ def _refs_fingerprint(references: Dict[str, str]) -> str: def find_op(binary_path: str = "") -> Optional[Path]: """Resolve a usable ``op`` binary, or None. A pinned ``binary_path`` is used verbatim — pinned-but-missing returns None rather than falling back to PATH.""" - if binary_path: - pinned = Path(binary_path) - return pinned if pinned.exists() and os.access(pinned, os.X_OK) else None - found = shutil.which("op") - return Path(found) if found else None + found = binary_path or shutil.which("op") + if not found or (binary_path and not os.access(binary_path, os.X_OK)): + return None + return Path(found) def _scrub(text: str) -> str: diff --git a/agent/secret_sources/registry.py b/agent/secret_sources/registry.py index da72daad2a..936a278c99 100644 --- a/agent/secret_sources/registry.py +++ b/agent/secret_sources/registry.py @@ -146,8 +146,7 @@ def snapshot_registration(name: str, *, scope: Optional[str] = None) -> Optional """Return the registration owned by exactly one registry layer.""" _ensure_builtin_sources() with _REGISTRY_LOCK: - target = _SOURCES if scope is None else _SCOPED_SOURCES.get(scope, {}) - return target.get(name) + return (_SOURCES if scope is None else _SCOPED_SOURCES.get(scope, {})).get(name) def restore_registration(name: str, current: SecretSource, previous: Optional[SecretSource], *, @@ -260,22 +259,18 @@ def _ordered_enabled_sources(secrets_cfg: dict, *, scope: Optional[str] = None) sources = {source.name: source for source in list_sources(scope=scope)} explicit = secrets_cfg.get("sources") - order: Dict[str, None] = {} # insertion-ordered set - if isinstance(explicit, list): - names = [e for e in explicit if isinstance(e, str)] - order.update((n, None) for n in names if n in sources) - unknown = [n for n in names if n not in sources] - if unknown: - logger.warning("secrets.sources names unknown source(s): %s (known: %s)", - ", ".join(unknown), ", ".join(sources) or "none") - order.update((n, None) for n in sources) + names = [e for e in explicit if isinstance(e, str)] if isinstance(explicit, list) else [] + unknown = [n for n in names if n not in sources] + if unknown: + logger.warning("secrets.sources names unknown source(s): %s (known: %s)", + ", ".join(unknown), ", ".join(sources) or "none") + order = dict.fromkeys([n for n in names if n in sources] + list(sources)) # insertion-ordered set enabled: List[SecretSource] = [] for name in order: - source = sources[name] try: - if source.is_enabled(_section(secrets_cfg, name)): - enabled.append(source) + if sources[name].is_enabled(_section(secrets_cfg, name)): + enabled.append(sources[name]) except Exception: # noqa: BLE001 logger.warning("Secret source '%s' is_enabled() raised; skipping", name, exc_info=True) return enabled @@ -301,15 +296,11 @@ _ALIAS_SUFFIXES = ("_API_KEY", "_TOKEN", "_SECRET", "_KEY", "_PASSWORD") def _profile_alias_target(var: str, profile: str) -> Optional[str]: """Map ``FOO_`` to ``FOO`` for the active profile when safe.""" - if not profile: - return None suffix = "_" + profile.replace("-", "_").upper() - if not var.endswith(suffix): + if not profile or not var.endswith(suffix): return None alias = var[: -len(suffix)] - if not alias or not is_valid_env_name(alias) or not alias.endswith(_ALIAS_SUFFIXES): - return None - return alias + return alias if alias and is_valid_env_name(alias) and alias.endswith(_ALIAS_SUFFIXES) else None class _Applier: @@ -386,7 +377,6 @@ def apply_all(secrets_cfg: dict, home_path: Path, """ env = environ if environ is not None else os.environ report = ApplyReport() - secrets_cfg = secrets_cfg if isinstance(secrets_cfg, dict) else {} enabled = _ordered_enabled_sources(secrets_cfg, scope=hermes_home_key(home_path)) if not enabled: