Support dynamic environment variable injection in mcpgateway.translate for STDIO MCP servers

[crivetimihai]

Header Passthrough for STDIO MCP Servers

Research Summary

Current State

mcpgateway.translate supports OAuth2 Bearer tokens for outbound connections (when acting as a client), but does not support dynamic environment variable injection from incoming request headers when exposing STDIO MCP servers.

Key Findings

1. Existing OAuth2 Support: mcpgateway.translate already accepts --oauth2Bearer for outbound client connections
2. STDIO Process Creation: All subprocess creation uses asyncio.create_subprocess_exec() without custom environment variables
3. Header Passthrough Infrastructure: Exists in main gateway but not integrated with translate module
4. No Dynamic Environment Injection: STDIO processes inherit the parent process environment only

Architecture Gap

┌─────────────────┐    ┌──────────────────┐    ┌─────────────────┐
│ Client Request  │───▶│ mcpgateway       │───▶│ STDIO MCP       │
│ Authorization:  │    │ .translate       │    │ Server          │
│ Bearer <token>  │    │                  │    │ (needs env var) │
└─────────────────┘    └──────────────────┘    └─────────────────┘
                              │
                              ▼
                    ❌ Headers NOT converted
                       to environment variables

Code Locations

• STDIO Process Creation: mcpgateway/translate.py:369-374, 1037-1042, 1245-1250, 1439-1444
• Header Handling: Only for outbound client connections (oauth2_bearer parameter)
• Environment Variables: Not supported - all use shlex.split(cmd) without env parameter

Proposed Solution

Feature Request: Dynamic Environment Variable Injection

Title: Support dynamic environment variable injection in mcpgateway.translate for STDIO MCP servers

Description:
Enable mcpgateway.translate to extract headers from incoming HTTP requests and inject them as environment variables when starting STDIO MCP processes. This would allow secure token passthrough for services like GitHub Enterprise MCP servers.

Use Case: GitHub Enterprise MCP

Scenario:

• Self-hosted GitHub Enterprise with MCP server requiring Personal Access Token (PAT)
• Multiple users with different PATs
• Need to pass user's PAT from request header to MCP server environment

Current Limitation:

# This works for static tokens only
export GITHUB_TOKEN=static-token
python3 -m mcpgateway.translate --stdio "uvx mcp-server-github" --port 9000

Desired Functionality:

# Dynamic token injection from request headers
python3 -m mcpgateway.translate \
  --stdio "uvx mcp-server-github" \
  --port 9000 \
  --header-to-env "Authorization=GITHUB_TOKEN" \
  --header-to-env "X-GitHub-Enterprise-Host=GITHUB_HOST"

Implementation Requirements

1. New CLI Arguments:

   • --header-to-env HEADER=ENV_VAR: Map request header to environment variable
   • --enable-dynamic-env: Safety flag to enable the feature

2. Modified STDIO Process Creation:

   • Extract headers from incoming requests
   • Build custom environment with mapped variables
   • Pass to asyncio.create_subprocess_exec(env=custom_env)

3. Security Considerations:

   • Whitelist allowed headers (no arbitrary header injection)
   • Sanitize header values (prevent injection attacks)
   • Log security events

4. Integration Points:

   • Modify StdIOEndpoint.start() to accept environment variables
   • Update FastAPI endpoints to extract and forward headers
   • Integrate with existing OAuth2 bearer token support

Code Changes Required

1. StdIOEndpoint Class (mcpgateway/translate.py:298-460):

   async def start(self, env_vars: Optional[Dict[str, str]] = None) -> None:
       env = os.environ.copy()
       if env_vars:
           env.update(env_vars)
       
       self._proc = await asyncio.create_subprocess_exec(
           *shlex.split(self._cmd),
           stdin=asyncio.subprocess.PIPE,
           stdout=asyncio.subprocess.PIPE, 
           stderr=sys.stderr,
           env=env  # 🔑 Add environment variable support
       )

2. FastAPI Request Handlers:

   • Extract configured headers from requests
   • Pass to STDIO endpoint as environment variables
   • Log header extraction for debugging

3. CLI Argument Parsing (mcpgateway/translate.py:740-906):

   • Add --header-to-env and --enable-dynamic-env options
   • Validate header-to-environment mappings

Benefits

1. Multi-tenant Support: Different users can use different tokens
2. Security: Tokens don't need to be stored in process environment
3. Flexibility: Works with any STDIO MCP server requiring environment variables
4. Integration: Seamless with existing gateway passthrough headers

Priority: Medium-High

This enables important enterprise use cases and improves the multi-tenant capabilities of the MCP Gateway ecosystem.

---

Labels: enhancement, translate, stdio, security, enterprise
Milestone: Next minor release

[manavgup]

Implementation Plan for Dynamic Environment Variable Injection

Phase 1: Core Infrastructure Changes

1.1 Modify StdIOEndpoint Class

File: mcpgateway/translate.py (lines 301-463)

class StdIOEndpoint:
    def __init__(self, cmd: str, pubsub: _PubSub, env_vars: Optional[Dict[str, str]] = None) -> None:
        self._cmd = cmd
        self._pubsub = pubsub
        self._env_vars = env_vars or {}
        # ... existing attributes

    async def start(self, additional_env_vars: Optional[Dict[str, str]] = None) -> None:
        """Start the stdio subprocess with custom environment variables."""
        LOGGER.info(f"Starting stdio subprocess: {self._cmd}")
        
        # Build environment from base + configured + additional
        env = os.environ.copy()
        env.update(self._env_vars)
        if additional_env_vars:
            env.update(additional_env_vars)
        
        self._proc = await asyncio.create_subprocess_exec(
            *shlex.split(self._cmd),
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=sys.stderr,
            env=env  # 🔑 Add environment variable support
        )
        # ... rest of existing implementation

1.2 Add CLI Arguments

File: mcpgateway/translate.py (lines 744-909)

def _parse_args(argv: Sequence[str]) -> argparse.Namespace:
    # ... existing arguments
    
    # New header-to-environment mapping arguments
    p.add_argument(
        "--header-to-env",
        action="append",
        help="Map request header to environment variable (format: HEADER=ENV_VAR)",
    )
    p.add_argument(
        "--enable-dynamic-env",
        action="store_true",
        help="Enable dynamic environment variable injection from request headers",
    )
    
    return p.parse_args(argv)

1.3 Add Header Processing Utilities

New File: mcpgateway/translate_header_utils.py

"""Header processing utilities for dynamic environment injection in translate module."""

import logging
import re
from typing import Dict, List, Optional

logger = logging.getLogger(__name__)

# Security constants
ALLOWED_HEADERS_REGEX = re.compile(r"^[A-Za-z0-9\-]+$")
MAX_HEADER_VALUE_LENGTH = 4096
MAX_ENV_VAR_NAME_LENGTH = 64

class HeaderMappingError(Exception):
    """Raised when header mapping configuration is invalid."""

def validate_header_mapping(header_name: str, env_var_name: str) -> None:
    """Validate header name and environment variable name."""
    if not ALLOWED_HEADERS_REGEX.match(header_name):
        raise HeaderMappingError(f"Invalid header name '{header_name}' - must contain only alphanumeric characters and hyphens")
    
    if not re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", env_var_name):
        raise HeaderMappingError(f"Invalid environment variable name '{env_var_name}' - must start with letter/underscore and contain only alphanumeric characters and underscores")

def sanitize_header_value(value: str) -> str:
    """Sanitize header value for environment variable injection."""
    if len(value) > MAX_HEADER_VALUE_LENGTH:
        logger.warning(f"Header value truncated from {len(value)} to {MAX_HEADER_VALUE_LENGTH} characters")
        value = value[:MAX_HEADER_VALUE_LENGTH]
    
    # Remove potentially dangerous characters
    value = re.sub(r'[^\x20-\x7E]', '', value)  # Only printable ASCII
    value = value.replace('\x00', '')  # Remove null bytes
    
    return value

def parse_header_mappings(header_mappings: List[str]) -> Dict[str, str]:
    """Parse header-to-environment mappings from CLI arguments."""
    mappings = {}
    
    for mapping in header_mappings:
        if "=" not in mapping:
            raise HeaderMappingError(f"Invalid mapping format '{mapping}' - expected HEADER=ENV_VAR")
        
        header_name, env_var_name = mapping.split("=", 1)
        header_name = header_name.strip()
        env_var_name = env_var_name.strip()
        
        if not header_name or not env_var_name:
            raise HeaderMappingError(f"Empty header name or environment variable name in '{mapping}'")
        
        validate_header_mapping(header_name, env_var_name)
        
        if header_name in mappings:
            raise HeaderMappingError(f"Duplicate header mapping for '{header_name}'")
        
        mappings[header_name] = env_var_name
    
    return mappings

def extract_env_vars_from_headers(
    request_headers: Dict[str, str], 
    header_mappings: Dict[str, str]
) -> Dict[str, str]:
    """Extract environment variables from request headers."""
    env_vars = {}
    
    for header_name, env_var_name in header_mappings.items():
        # Case-insensitive header matching
        header_value = None
        for req_header, value in request_headers.items():
            if req_header.lower() == header_name.lower():
                header_value = value
                break
        
        if header_value is not None:
            try:
                sanitized_value = sanitize_header_value(header_value)
                if sanitized_value:  # Only add non-empty values
                    env_vars[env_var_name] = sanitized_value
                    logger.debug(f"Mapped header {header_name} to {env_var_name}")
                else:
                    logger.warning(f"Header {header_name} value became empty after sanitization")
            except Exception as e:
                logger.warning(f"Failed to process header {header_name}: {e}")
    
    return env_vars

Phase 2: FastAPI Integration

2.1 Modify FastAPI Endpoints

File: mcpgateway/translate.py (lines 552-736)

def _build_fastapi(
    pubsub: _PubSub,
    stdio: StdIOEndpoint,
    keep_alive: int = KEEP_ALIVE_INTERVAL,
    sse_path: str = "/sse",
    message_path: str = "/message",
    cors_origins: Optional[List[str]] = None,
    header_mappings: Optional[Dict[str, str]] = None,  # 🔑 New parameter
) -> FastAPI:
    """Build FastAPI application with header-to-environment support."""
    
    # ... existing setup ...

    @app.get(sse_path)
    async def get_sse(request: Request) -> EventSourceResponse:
        """Stream subprocess stdout with dynamic environment injection."""
        # 🔑 Extract environment variables from headers
        env_vars = {}
        if header_mappings:
            from mcpgateway.translate_header_utils import extract_env_vars_from_headers
            env_vars = extract_env_vars_from_headers(dict(request.headers), header_mappings)
        
        # 🔑 Restart stdio with new environment if needed
        if env_vars and not stdio._proc:
            await stdio.start(additional_env_vars=env_vars)
        
        # ... rest of existing implementation

    @app.post(message_path, status_code=status.HTTP_202_ACCEPTED)
    async def post_message(raw: Request, session_id: str | None = None) -> Response:
        """Forward JSON-RPC request with dynamic environment injection."""
        # 🔑 Extract environment variables from headers
        env_vars = {}
        if header_mappings:
            from mcpgateway.translate_header_utils import extract_env_vars_from_headers
            env_vars = extract_env_vars_from_headers(dict(raw.headers), header_mappings)
        
        # 🔑 Restart stdio with new environment if needed
        if env_vars and not stdio._proc:
            await stdio.start(additional_env_vars=env_vars)
        
        # ... rest of existing implementation

Phase 3: CLI Integration

3.1 Update Main Function

File: mcpgateway/translate.py (lines 2165-2232)

def main(argv: Optional[Sequence[str]] | None = None) -> None:
    """Entry point with header-to-environment support."""
    args = _parse_args(argv or sys.argv[1:])
    
    # ... existing logging setup ...
    
    try:
        # 🔑 Parse header mappings
        header_mappings = None
        if getattr(args, 'enable_dynamic_env', False):
            if not getattr(args, 'header_to_env', None):
                print("Error: --enable-dynamic-env requires --header-to-env mappings", file=sys.stderr)
                sys.exit(1)
            
            from mcpgateway.translate_header_utils import parse_header_mappings
            try:
                header_mappings = parse_header_mappings(args.header_to_env)
                LOGGER.info(f"Header mappings configured: {header_mappings}")
            except Exception as e:
                print(f"Error parsing header mappings: {e}", file=sys.stderr)
                sys.exit(1)
        
        # Handle local stdio server exposure
        if args.stdio:
            # ... existing protocol logic ...
            
            # 🔑 Pass header mappings to multi-protocol server
            asyncio.run(
                _run_multi_protocol_server(
                    cmd=args.stdio,
                    port=args.port,
                    log_level=args.logLevel,
                    cors=args.cors,
                    host=args.host,
                    expose_sse=expose_sse,
                    expose_streamable_http=expose_streamable_http,
                    sse_path=getattr(args, "ssePath", "/sse"),
                    message_path=getattr(args, "messagePath", "/message"),
                    keep_alive=getattr(args, "keepAlive", KEEP_ALIVE_INTERVAL),
                    stateless=getattr(args, "stateless", False),
                    json_response=getattr(args, "jsonResponse", False),
                    header_mappings=header_mappings,  # 🔑 New parameter
                )
            )
        
        # ... rest of existing implementation

Testing Strategy

Unit Tests

New File: tests/unit/mcpgateway/test_translate_header_utils.py

• Test header mapping validation
• Test header value sanitization
• Test environment variable extraction
• Test error handling for invalid configurations

Integration Tests

New File: tests/integration/test_translate_dynamic_env.py

• Test StdIOEndpoint with environment variables
• Test header-to-environment mapping integration
• Test FastAPI endpoint integration

End-to-End Tests

New File: tests/e2e/test_translate_dynamic_env_e2e.py

• Test complete HTTP request flow with header injection
• Test SSE and streamable HTTP protocols
• Test multiple concurrent requests with different headers

Security Considerations

Security Validations

1. Header Name Validation: Only allow alphanumeric characters and hyphens
2. Environment Variable Name Validation: Must start with letter/underscore, contain only alphanumeric characters and underscores
3. Value Sanitization: Remove dangerous characters, limit length
4. Case-Insensitive Matching: Prevent header name confusion attacks
5. Logging: Log all header extractions for security auditing

Configuration Security

# Security configuration
SECURITY_CONFIG = {
    "max_header_mappings": 10,  # Limit number of mappings
    "max_header_value_length": 4096,
    "max_env_var_name_length": 64,
    "allowed_header_pattern": r"^[A-Za-z0-9\-]+$",
    "allowed_env_var_pattern": r"^[A-Za-z_][A-Za-z0-9_]*$",
    "enable_security_logging": True,
}

Usage Examples

# Basic usage with GitHub Enterprise MCP
python3 -m mcpgateway.translate \
  --stdio "uvx mcp-server-github" \
  --port 9000 \
  --enable-dynamic-env \
  --header-to-env "Authorization=GITHUB_TOKEN" \
  --header-to-env "X-GitHub-Enterprise-Host=GITHUB_HOST"

# Multiple header mappings
python3 -m mcpgateway.translate \
  --stdio "uvx mcp-server-custom" \
  --port 9000 \
  --enable-dynamic-env \
  --header-to-env "Authorization=API_TOKEN" \
  --header-to-env "X-Tenant-Id=TENANT_ID" \
  --header-to-env "X-Environment=ENV_NAME"

# Test with curl
curl -X POST http://localhost:9000/message \
  -H "Authorization: Bearer github-token-123" \
  -H "X-Tenant-Id: acme-corp" \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"test","version":"1.0.0"}}}'

Security Best Practices

1. Whitelist Headers: Only map specific headers you need
2. Validate Environment Variables: Ensure MCP servers validate environment variables
3. Monitor Logs: Watch for suspicious header injection attempts
4. Use HTTPS: Always use HTTPS in production to protect headers in transit
5. Regular Rotation: Regularly rotate tokens passed via headers

Risk Assessment

Low Risk:

• Backward compatibility maintained
• Feature is opt-in via --enable-dynamic-env
• Extensive security validation
• Comprehensive testing coverage

Mitigation Strategies:

• Extensive input validation and sanitization
• Security logging for audit trails
• Clear documentation of security considerations
• Gradual rollout with monitoring

This implementation plan provides a secure, well-tested solution for dynamic environment variable injection while maintaining backward compatibility and following security best practices.