refactor(agent/secrets): collapse defensive branches (alias target, ordered sources, zip containment, find_op, query-string sub)

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
 
 

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 ``<home>/.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))

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
 
 

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:

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_<PROFILE>`` 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: