feat: TASKFLOW_ENV_DENYLIST to filter env vars from MCP subprocesses

Author: anticomputer

README.md

@@ -115,6 +115,19 @@ Error: [BadRequestError] model 'foo' not found
 python -m seclab_taskflow_agent --debug -t examples.taskflows.echo
 ```
 
+### MCP Environment Denylist
+
+By default, MCP server subprocesses inherit the parent environment. To prevent
+specific variables from leaking to MCP servers, set `TASKFLOW_ENV_DENYLIST` to
+a comma-separated list of variable names:
+
+```bash
+export TASKFLOW_ENV_DENYLIST="MY_SECRET_TOKEN,PRIVATE_KEY,OTHER_CREDENTIAL"
+```
+
+Toolbox-level `env:` declarations in YAML still inject exactly what each server
+needs, so explicitly configured variables are unaffected.
+
 ## Use Cases and Examples
 
 The Seclab Taskflow Agent framework was primarily designed to fit the iterative feedback loop driven work involved in Agentic security research workflows and vulnerability triage tasks.

src/seclab_taskflow_agent/mcp_transport.py

@@ -18,6 +18,7 @@
     "AsyncDebugMCPServerStdio",
     "ReconnectingMCPServerStdio",
     "StreamableMCPThread",
+    "_filtered_env",
 ]
 
 import asyncio
@@ -38,6 +39,21 @@
 _EXPECTED_EXIT_CODES: frozenset[int] = frozenset({0, -signal.SIGTERM})
 
 
+def _filtered_env() -> dict[str, str]:
+    """Return a copy of ``os.environ`` with denied variables removed.
+
+    Set ``TASKFLOW_ENV_DENYLIST`` to a comma-separated list of variable
+    names that should not be forwarded to MCP server subprocesses.
+    Toolbox-level ``env:`` declarations in YAML still inject what each
+    server explicitly needs.
+    """
+    denylist_raw = os.environ.get("TASKFLOW_ENV_DENYLIST", "")
+    if not denylist_raw:
+        return os.environ.copy()
+    denied = {k.strip() for k in denylist_raw.split(",") if k.strip()}
+    return {k: v for k, v in os.environ.items() if k not in denied}
+
+
 class StreamableMCPThread(Thread):
     """Thread that manages a local streamable MCP server subprocess.
 
@@ -68,7 +84,7 @@ def __init__(
         self.on_output: Callable[[str], None] | None = on_output
         self.on_error: Callable[[str], None] | None = on_error
         self.poll_interval: float = poll_interval
-        self.env: dict[str, str] = os.environ.copy()  # XXX: potential for environment leak to MCP
+        self.env: dict[str, str] = _filtered_env()
         if env:
             self.env.update(env)
         self._stop_event: Event = Event()

tests/test_mcp_transport.py

@@ -0,0 +1,45 @@
+# SPDX-FileCopyrightText: GitHub, Inc.
+# SPDX-License-Identifier: MIT
+
+"""Tests for MCP transport utilities."""
+
+import os
+
+from seclab_taskflow_agent.mcp_transport import _filtered_env
+
+
+class TestFilteredEnv:
+    """Tests for _filtered_env environment denylist."""
+
+    def test_no_denylist_copies_env(self, monkeypatch):
+        """Without TASKFLOW_ENV_DENYLIST, returns full env copy."""
+        monkeypatch.delenv("TASKFLOW_ENV_DENYLIST", raising=False)
+        monkeypatch.setenv("TEST_VAR_A", "hello")
+        result = _filtered_env()
+        assert result["TEST_VAR_A"] == "hello"
+        assert result is not os.environ
+
+    def test_denylist_strips_variables(self, monkeypatch):
+        """Comma-separated denylist removes matching variables."""
+        monkeypatch.setenv("SECRET_TOKEN", "s3cret")
+        monkeypatch.setenv("MY_API_KEY", "key123")
+        monkeypatch.setenv("SAFE_VAR", "keep")
+        monkeypatch.setenv("TASKFLOW_ENV_DENYLIST", "SECRET_TOKEN,MY_API_KEY")
+        result = _filtered_env()
+        assert "SECRET_TOKEN" not in result
+        assert "MY_API_KEY" not in result
+        assert result["SAFE_VAR"] == "keep"
+
+    def test_denylist_handles_whitespace(self, monkeypatch):
+        """Whitespace around denylist entries is trimmed."""
+        monkeypatch.setenv("FOO", "bar")
+        monkeypatch.setenv("TASKFLOW_ENV_DENYLIST", " FOO , ")
+        result = _filtered_env()
+        assert "FOO" not in result
+
+    def test_empty_denylist_copies_env(self, monkeypatch):
+        """Empty TASKFLOW_ENV_DENYLIST behaves like unset."""
+        monkeypatch.setenv("TASKFLOW_ENV_DENYLIST", "")
+        monkeypatch.setenv("KEEP_ME", "yes")
+        result = _filtered_env()
+        assert result["KEEP_ME"] == "yes"