feat(shell): scrub ambient credentials from child process environment (#618)

coworker/agent.py

@@ -264,7 +264,11 @@ def build_engine(
 
     workspace_trusted = bool(ws and WorkspaceTrustStore().is_trusted(ws))
     config = load_config(ws, workspace_trusted=workspace_trusted)
-    executor = LocalExecutor(cwd=ws) if ws is not None else None
+    executor = (
+        LocalExecutor(cwd=ws, allowed_env=config.shell_allowed_env)
+        if ws is not None
+        else None
+    )
     todo = TodoList()
     context = AgentContext(
         workspace=ws, executor=executor, todo=todo, roots=root_list or None

coworker/config.py

@@ -73,6 +73,10 @@ class Config:
     cloud_relay_ws_url: str = (
         "wss://l4z1paxb83.execute-api.us-east-1.amazonaws.com/ocw-connect"
     )
+    # Environment variable names explicitly allowed through to run_shell child processes
+    # (exempt from ambient credential scrubbing). Configured via [shell] allowed_env
+    # or shell_allowed_env in config.toml.
+    shell_allowed_env: list[str] = field(default_factory=list)
 
 
 _FIELDS = {
@@ -84,6 +88,7 @@ class Config:
     "allowed_domains",
     "auto_approve",
     "auto_approve_shadow",
+    "shell_allowed_env",
     "host",
     "port",
     "web_search_provider",
@@ -95,15 +100,16 @@ class Config:
 }
 
 # These fields change what consequential actions can run without a prompt, so the normal
-# workspace override pass never applies them. `allowed_commands` is added separately only
-# for a canonically trusted workspace; `auto_allow` and `allowed_domains` remain user-global
-# only (a repo must not be able to widen the agent's command or network reach).
+# workspace override pass never applies them. `allowed_commands` and `shell_allowed_env`
+# are added separately only for a canonically trusted workspace; `auto_allow` and
+# `allowed_domains` remain user-global only (a repo must not be able to widen reach).
 _GLOBAL_ONLY_FIELDS = {
     "allowed_commands",
     "auto_allow",
     "allowed_domains",
     "auto_approve",
     "auto_approve_shadow",
+    "shell_allowed_env",
 }
 _WORKSPACE_FIELDS = _FIELDS - _GLOBAL_ONLY_FIELDS
 
@@ -120,6 +126,18 @@ def _read(path: Path) -> dict[str, Any]:
         return {}
 
 
+def _extract_shell_allowed_env(data: dict[str, Any]) -> list[str]:
+    shell_sec = data.get("shell")
+    if isinstance(shell_sec, dict):
+        val = shell_sec.get("allowed_env")
+        if isinstance(val, list):
+            return list(dict.fromkeys(str(v).strip() for v in val if isinstance(v, str) and str(v).strip()))
+    val = data.get("shell_allowed_env")
+    if isinstance(val, list):
+        return list(dict.fromkeys(str(v).strip() for v in val if isinstance(v, str) and str(v).strip()))
+    return []
+
+
 def workspace_allowed_commands(workspace: str | Path) -> list[str]:
     """Command prefixes requested by repository config; advisory until workspace trust."""
     path = Path(workspace).expanduser() / ".coworker" / "config.toml"
@@ -139,13 +157,17 @@ def load_config(
 
     g = Path(global_path) if global_path is not None else global_config_path()
     if g.is_file():
-        for key, value in _read(g).items():
+        g_data = _read(g)
+        for key, value in g_data.items():
             if key in _FIELDS:
                 setattr(cfg, key, value)
+        if shell_env := _extract_shell_allowed_env(g_data):
+            cfg.shell_allowed_env = shell_env
     if workspace:
         w = Path(workspace).expanduser() / ".coworker" / "config.toml"
         if w.is_file():
-            for key, value in _read(w).items():
+            w_data = _read(w)
+            for key, value in w_data.items():
                 if key in _WORKSPACE_FIELDS:
                     setattr(cfg, key, value)
             if workspace_trusted:
@@ -154,4 +176,8 @@ def load_config(
                         [*cfg.allowed_commands, *workspace_allowed_commands(workspace)]
                     )
                 )
+                if w_shell_env := _extract_shell_allowed_env(w_data):
+                    cfg.shell_allowed_env = list(
+                        dict.fromkeys([*cfg.shell_allowed_env, *w_shell_env])
+                    )
     return cfg

coworker/tools/shell.py

@@ -51,6 +51,48 @@
     "PIP_NO_INPUT": "1",
 }
 
+# Ambient credential patterns scrubbed from inherited os.environ unless explicitly allowlisted
+_SENSITIVE_PREFIXES = ("AWS_", "AZURE_", "GITHUB_TOKEN", "GH_TOKEN")
+_SENSITIVE_SUBSTRINGS = (
+    "_API_KEY",
+    "_SECRET",
+    "_TOKEN",
+    "_PASSWORD",
+    "_PASSWD",
+    "_CREDENTIAL",
+)
+_SENSITIVE_SUFFIXES = ("_KEY", "_AUTH")
+
+
+def is_sensitive_env(name: str) -> bool:
+    """Return True if an environment variable name matches sensitive credential patterns."""
+    upper = name.upper()
+    if upper.startswith(_SENSITIVE_PREFIXES):
+        return True
+    if any(sub in upper for sub in _SENSITIVE_SUBSTRINGS):
+        return True
+    if upper.endswith(_SENSITIVE_SUFFIXES):
+        return True
+    return False
+
+
+def filter_ambient_env(
+    raw_env: dict[str, str] | os._Environ[str],
+    *,
+    allowed_env: Optional[set[str] | list[str]] = None,
+) -> dict[str, str]:
+    """Scrub ambient credentials from an inherited environment mapping.
+
+    Variables matching known sensitive patterns (AWS_*, *_API_KEY, *_SECRET_*, *_TOKEN,
+    etc.) are dropped unless explicitly listed in allowed_env.
+    """
+    allowed = set(allowed_env or [])
+    return {
+        k: v
+        for k, v in raw_env.items()
+        if k in allowed or not is_sensitive_env(k)
+    }
+
 
 class Executor(ABC):
     @abstractmethod
@@ -140,13 +182,15 @@ def __init__(
         *,
         cwd: str | Path,
         env: Optional[dict[str, str]] = None,
+        allowed_env: Optional[list[str] | set[str]] = None,
         shell_path: Optional[str] = None,
         default_timeout: float = _DEFAULT_TIMEOUT,
         max_output_chars: int = 20_000,
     ) -> None:
         self.cwd = str(Path(cwd).expanduser().resolve())
         self.default_timeout = default_timeout
         self.max_output_chars = max_output_chars
+        self.allowed_env = set(allowed_env or [])
         self._marker = f"__COWORKER_DONE_{uuid.uuid4().hex}__"
         self._is_windows = _IS_WINDOWS
         self._bg_tasks: dict[str, _BackgroundTask] = {}
@@ -161,7 +205,8 @@ def __init__(
         if shell_path is None:
             shell_path = "powershell.exe" if self._is_windows else "/bin/bash"
         self._shell_path = shell_path
-        self._env = {**os.environ, **_NONINTERACTIVE_ENV, **(env or {})}
+        ambient = filter_ambient_env(os.environ, allowed_env=self.allowed_env)
+        self._env = {**ambient, **_NONINTERACTIVE_ENV, **(env or {})}
         # Managed pinned tools (toolchain.install) land under one stable bin dir; putting
         # it on PATH up front — even before anything is installed there — means a tool the
         # user approves mid-session works in THIS shell immediately, by name, no respawn.

tests/test_config.py

@@ -142,3 +142,29 @@ def test_cloud_endpoints_default_to_production():
     cfg = Config()
     assert cfg.cloud_base_url == "https://api.openworker.com"
     assert cfg.cloud_relay_ws_url.startswith("wss://")
+
+
+def test_shell_allowed_env_from_global_config(tmp_path):
+    g = tmp_path / "global.toml"
+    g.write_text('[shell]\nallowed_env = ["AWS_PROFILE", "CUSTOM_VAR"]\n')
+    cfg = load_config(global_path=g)
+    assert cfg.shell_allowed_env == ["AWS_PROFILE", "CUSTOM_VAR"]
+
+
+def test_shell_allowed_env_workspace_trusted_only(tmp_path):
+    g = tmp_path / "global.toml"
+    g.write_text('[shell]\nallowed_env = ["AWS_PROFILE"]\n')
+    ws = tmp_path / "ws"
+    (ws / ".coworker").mkdir(parents=True)
+    (ws / ".coworker" / "config.toml").write_text(
+        '[shell]\nallowed_env = ["EXFIL_TOKEN"]\n'
+    )
+
+    # Untrusted: workspace shell_allowed_env ignored
+    cfg = load_config(ws, global_path=g, workspace_trusted=False)
+    assert cfg.shell_allowed_env == ["AWS_PROFILE"]
+
+    # Trusted: workspace shell_allowed_env merged
+    cfg_trusted = load_config(ws, global_path=g, workspace_trusted=True)
+    assert cfg_trusted.shell_allowed_env == ["AWS_PROFILE", "EXFIL_TOKEN"]
+

tests/test_shell.py

@@ -218,3 +218,82 @@ def test_background_unknown_task_errors(executor):
     assert (
         "unknown task" in reg.execute("shell_task_kill", {"task_id": "bg-99"})["error"]
     )
+
+
+def test_is_sensitive_env():
+    from coworker.tools.shell import is_sensitive_env
+
+    assert is_sensitive_env("AWS_SECRET_ACCESS_KEY") is True
+    assert is_sensitive_env("AWS_ACCESS_KEY_ID") is True
+    assert is_sensitive_env("AWS_PROFILE") is True
+    assert is_sensitive_env("AZURE_CLIENT_SECRET") is True
+    assert is_sensitive_env("OPENAI_API_KEY") is True
+    assert is_sensitive_env("GITHUB_TOKEN") is True
+    assert is_sensitive_env("MY_SECRET_TOKEN") is True
+    assert is_sensitive_env("DATABASE_PASSWORD") is True
+    assert is_sensitive_env("ADMIN_PASSWD") is True
+    assert is_sensitive_env("GCP_CREDENTIALS") is True
+    assert is_sensitive_env("SERVICE_KEY") is True
+    assert is_sensitive_env("PROXY_AUTH") is True
+
+    assert is_sensitive_env("PATH") is False
+    assert is_sensitive_env("HOME") is False
+    assert is_sensitive_env("USER") is False
+    assert is_sensitive_env("TERM") is False
+    assert is_sensitive_env("SHELL") is False
+    assert is_sensitive_env("LANG") is False
+    assert is_sensitive_env("TMPDIR") is False
+    assert is_sensitive_env("VIRTUAL_ENV") is False
+    assert is_sensitive_env("PYTHONPATH") is False
+
+
+def test_ambient_sensitive_env_scrubbed(tmp_path, monkeypatch):
+    monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "super-secret")
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-live-12345")
+    monkeypatch.setenv("GITHUB_TOKEN", "ghp_token999")
+    monkeypatch.setenv("DATABASE_PASSWORD", "dbpass")
+    monkeypatch.setenv("SAFE_APP_VAR", "visible")
+
+    ex = LocalExecutor(cwd=tmp_path, default_timeout=5)
+    try:
+        assert "AWS_SECRET_ACCESS_KEY" not in ex._env
+        assert "OPENAI_API_KEY" not in ex._env
+        assert "GITHUB_TOKEN" not in ex._env
+        assert "DATABASE_PASSWORD" not in ex._env
+        assert ex._env.get("SAFE_APP_VAR") == "visible"
+
+        # Verify child shell process cannot see the scrubbed variable
+        echo_cmd = "echo $OPENAI_API_KEY" if not _WIN else "echo $env:OPENAI_API_KEY"
+        res = ex.run(echo_cmd)
+        assert "sk-live-12345" not in res["output"]
+    finally:
+        ex.close()
+
+
+def test_allowed_env_exempts_sensitive_variables(tmp_path, monkeypatch):
+    monkeypatch.setenv("AWS_PROFILE", "staging")
+    monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "top-secret")
+
+    ex = LocalExecutor(
+        cwd=tmp_path, allowed_env=["AWS_PROFILE"], default_timeout=5
+    )
+    try:
+        assert ex._env.get("AWS_PROFILE") == "staging"
+        assert "AWS_SECRET_ACCESS_KEY" not in ex._env
+    finally:
+        ex.close()
+
+
+def test_explicit_env_argument_takes_precedence(tmp_path, monkeypatch):
+    monkeypatch.setenv("OPENAI_API_KEY", "ambient-secret")
+
+    ex = LocalExecutor(
+        cwd=tmp_path,
+        env={"OPENAI_API_KEY": "explicit-provided"},
+        default_timeout=5,
+    )
+    try:
+        assert ex._env.get("OPENAI_API_KEY") == "explicit-provided"
+    finally:
+        ex.close()
+