# V0.4.0 "ENTERPRISE" Release Plan

**Date:** 2026-02-01
**Codename:** Operation Enterprise Firewall
**Goal:** Transform from "Security Library" to "Default Agent Firewall" (Top 1% PyPI by Q3 2026)

---

## Executive Summary

V0.3.0 delivered the "Vaccine" features (network egress, honeypot, filesystem jail, framework injection). V0.4.0 addresses the **enterprise adoption blockers**:

1. **UnknownArgsMode** - Replace boolean `strict_mode` with explicit policy (block/strip_and_log/strip_silent)
2. **SandboxBackend Interface** - Make sandbox pluggable (E2B, Docker, Firejail, Local)
3. **Enhanced Audit Logging** - OTel export, forensics-grade events, args_hash, egress tracking
4. **SafePath/SafeURL Types** - Built-in safe types with automatic validation
5. **Capability Gating** - Per-tool capabilities (filesystem.read, network.http, process.exec)
6. **MCPProxyGuard** - Token passthrough prevention, session binding
7. **CLI Tools** - `airlock doctor` scan and verification

---

## Implementation Order (Dependency Graph)

```
[Phase 1 - Core Security Hardening]
   unknown_args.py  ─────┐
   safe_types.py    ─────┼──► [Phase 2 - Config Integration]
   capabilities.py  ─────┘          │
                                    ▼
                         config.py updates
                                    │
                                    ▼
                         [Phase 3 - Infrastructure]
                         sandbox_backend.py
                         audit_otel.py
                                    │
                                    ▼
                         [Phase 4 - MCP Security]
                         mcp_proxy_guard.py
                                    │
                                    ▼
                         [Phase 5 - CLI & Adoption]
                         cli/doctor.py
                         cli/verify.py
```

---

## Phase 1: Core Security Hardening

### 1.1 UnknownArgsMode (~100 lines)

**File:** `src/agent_airlock/unknown_args.py`

**Goal:** Replace boolean `strict_mode` with explicit policy enum that makes the security stance clear.

```python
from enum import Enum

class UnknownArgsMode(str, Enum):
    """How to handle unknown/hallucinated arguments."""

    BLOCK = "block"
    """Reject the call with an error (production default). RECOMMENDED."""

    STRIP_AND_LOG = "strip_and_log"
    """Strip unknown args and emit audit event. For staged rollouts."""

    STRIP_SILENT = "strip_silent"
    """Strip silently (DANGEROUS - hides attack attempts). Dev only."""

# Migration helper
def mode_from_strict_bool(strict_mode: bool) -> UnknownArgsMode:
    """Convert legacy strict_mode boolean to UnknownArgsMode."""
    return UnknownArgsMode.BLOCK if strict_mode else UnknownArgsMode.STRIP_SILENT
```

**Changes to `config.py`:**
```python
@dataclass
class AirlockConfig:
    # DEPRECATED: Use unknown_args instead
    strict_mode: bool = False

    # NEW: Explicit unknown args handling
    unknown_args: UnknownArgsMode = UnknownArgsMode.STRIP_AND_LOG  # Migration default
```

**Changes to `core.py`:**
- Replace `if self.config.strict_mode:` with `UnknownArgsMode` switch
- Emit audit event when `STRIP_AND_LOG` mode and args are stripped

**Migration Path:**
- Default `unknown_args=STRIP_AND_LOG` (safe migration from v0.3)
- `STRICT_POLICY` sets `unknown_args=BLOCK` (recommended for production)
- Deprecation warning if `strict_mode` is explicitly set

---

### 1.2 SafePath and SafeURL Types (~200 lines)

**File:** `src/agent_airlock/safe_types.py`

**Goal:** Built-in types that validate paths/URLs at the type level, preventing traversal and exfil.

```python
from pathlib import Path
from typing import Annotated
from pydantic import AfterValidator, HttpUrl

class SafePathValidator:
    """Validates paths are safe (no traversal, no secrets)."""

    def __init__(
        self,
        root_dir: Path | None = None,
        deny_patterns: list[str] | None = None,
        allow_absolute: bool = False,
    ):
        self.root_dir = root_dir
        self.deny_patterns = deny_patterns or [
            "~",           # Home directory
            "..",          # Parent traversal
            ".env",        # Environment files
            ".ssh",        # SSH keys
            ".git",        # Git config
            "*.pem",       # Certificates
            "*.key",       # Private keys
            "/etc/passwd", # System files
            "/etc/shadow",
        ]
        self.allow_absolute = allow_absolute

    def __call__(self, value: str | Path) -> Path:
        """Validate and return safe path."""
        path = Path(value)

        # Check against deny patterns
        path_str = str(path)
        for pattern in self.deny_patterns:
            if fnmatch.fnmatch(path_str, pattern) or pattern in path_str:
                raise ValueError(f"Path matches denied pattern: {pattern}")

        # Check absolute path if not allowed
        if not self.allow_absolute and path.is_absolute():
            raise ValueError("Absolute paths not allowed")

        # Check within root if specified
        if self.root_dir:
            resolved = (self.root_dir / path).resolve()
            if not str(resolved).startswith(str(self.root_dir.resolve())):
                raise ValueError(f"Path escapes root directory: {self.root_dir}")

        return path

# Type aliases for common use cases
SafePath = Annotated[Path, AfterValidator(SafePathValidator())]
SafePathInTmp = Annotated[Path, AfterValidator(SafePathValidator(root_dir=Path("/tmp/airlock")))]

class SafeURLValidator:
    """Validates URLs are safe (no file://, no metadata, no link-local)."""

    def __init__(
        self,
        allowed_schemes: list[str] | None = None,
        allowed_hosts: list[str] | None = None,
        block_private_ips: bool = True,
    ):
        self.allowed_schemes = allowed_schemes or ["https"]
        self.allowed_hosts = allowed_hosts  # None = allow all
        self.block_private_ips = block_private_ips

    def __call__(self, value: str) -> str:
        """Validate and return safe URL."""
        from urllib.parse import urlparse

        parsed = urlparse(value)

        # Check scheme
        if parsed.scheme not in self.allowed_schemes:
            raise ValueError(f"URL scheme '{parsed.scheme}' not allowed. Allowed: {self.allowed_schemes}")

        # Check host allowlist
        if self.allowed_hosts is not None and parsed.hostname not in self.allowed_hosts:
            raise ValueError(f"Host '{parsed.hostname}' not in allowed list")

        # Block metadata URLs (AWS/GCP)
        if parsed.hostname in ["169.254.169.254", "metadata.google.internal"]:
            raise ValueError("Cloud metadata URLs blocked")

        # Block file:// URLs
        if parsed.scheme == "file":
            raise ValueError("file:// URLs not allowed")

        return value

SafeURL = Annotated[str, AfterValidator(SafeURLValidator())]
SafeURLAllowHttp = Annotated[str, AfterValidator(SafeURLValidator(allowed_schemes=["http", "https"]))]
```

**Integration with Pydantic:**
- These types work automatically with `@Airlock()` validation
- Users can use them in tool function signatures

---

### 1.3 Capability Gating (~150 lines)

**File:** `src/agent_airlock/capabilities.py`

**Goal:** Per-tool capability requirements that can be checked against policy.

```python
from enum import Flag, auto
from dataclasses import dataclass

class Capability(Flag):
    """Capabilities that tools can require."""

    NONE = 0

    # Filesystem
    FILESYSTEM_READ = auto()
    FILESYSTEM_WRITE = auto()
    FILESYSTEM_DELETE = auto()

    # Network
    NETWORK_HTTP = auto()
    NETWORK_HTTPS = auto()
    NETWORK_ARBITRARY = auto()

    # Process
    PROCESS_EXEC = auto()
    PROCESS_SHELL = auto()

    # Sensitive Data
    DATA_PII = auto()
    DATA_SECRETS = auto()

    # Convenience combinations
    FILESYSTEM_ALL = FILESYSTEM_READ | FILESYSTEM_WRITE | FILESYSTEM_DELETE
    NETWORK_ALL = NETWORK_HTTP | NETWORK_HTTPS | NETWORK_ARBITRARY
    DANGEROUS = PROCESS_SHELL | FILESYSTEM_DELETE | NETWORK_ARBITRARY

@dataclass
class CapabilityPolicy:
    """Policy for capability gating."""

    granted: Capability = Capability.NONE
    denied: Capability = Capability.NONE
    require_sandbox_for: Capability = Capability.DANGEROUS

    def check(self, required: Capability, tool_name: str) -> None:
        """Check if required capabilities are allowed."""
        # Explicit deny takes precedence
        if required & self.denied:
            denied_caps = [c.name for c in Capability if c in (required & self.denied)]
            raise CapabilityDeniedError(
                f"Tool '{tool_name}' requires denied capabilities: {denied_caps}"
            )

        # Check if granted
        if required & ~self.granted:
            missing = [c.name for c in Capability if c in (required & ~self.granted)]
            raise CapabilityDeniedError(
                f"Tool '{tool_name}' requires capabilities not granted: {missing}"
            )

    def requires_sandbox(self, required: Capability) -> bool:
        """Check if capabilities require sandbox execution."""
        return bool(required & self.require_sandbox_for)

# Decorator to declare tool capabilities
def requires(*capabilities: Capability):
    """Decorator to declare capabilities required by a tool."""
    combined = Capability.NONE
    for cap in capabilities:
        combined |= cap

    def decorator(func):
        func.__airlock_capabilities__ = combined
        return func

    return decorator
```

**Usage:**
```python
from agent_airlock import Airlock, requires, Capability

@Airlock()
@requires(Capability.FILESYSTEM_READ)
def read_config(path: SafePath) -> str:
    return Path(path).read_text()

@Airlock(sandbox=True)
@requires(Capability.PROCESS_SHELL)
def run_command(cmd: str) -> str:
    return subprocess.check_output(cmd, shell=True)
```

---

## Phase 2: Config Integration

### 2.1 Config Updates (~50 lines)

**File:** `src/agent_airlock/config.py`

Add new fields:

```python
@dataclass
class AirlockConfig:
    # ... existing fields ...

    # V0.4.0 additions
    unknown_args: UnknownArgsMode = UnknownArgsMode.STRIP_AND_LOG
    capability_policy: CapabilityPolicy | None = None

    # Audit enhancements
    audit_otel_enabled: bool = False
    audit_otel_endpoint: str | None = None
    audit_include_args_hash: bool = True
```

### 2.2 Policy Updates (~30 lines)

**File:** `src/agent_airlock/policy.py`

Add capability policy to SecurityPolicy:

```python
@dataclass
class SecurityPolicy:
    # ... existing fields ...

    capability_policy: CapabilityPolicy | None = None

# Update predefined policies
STRICT_POLICY = SecurityPolicy(
    # ... existing ...
    capability_policy=CapabilityPolicy(
        granted=Capability.FILESYSTEM_READ | Capability.NETWORK_HTTPS,
        denied=Capability.PROCESS_SHELL | Capability.FILESYSTEM_DELETE,
        require_sandbox_for=Capability.DANGEROUS,
    ),
)
```

---

## Phase 3: Infrastructure

### 3.1 SandboxBackend Interface (~250 lines)

**File:** `src/agent_airlock/sandbox_backend.py`

**Goal:** Make sandbox execution pluggable for enterprise environments.

```python
from abc import ABC, abstractmethod
from typing import Protocol

class SandboxBackend(Protocol):
    """Protocol for sandbox backends."""

    @abstractmethod
    def execute(
        self,
        func: Callable[..., R],
        args: tuple[Any, ...],
        kwargs: dict[str, Any],
        timeout: int = 60,
    ) -> SandboxResult:
        """Execute a function in the sandbox."""
        ...

    @abstractmethod
    def is_available(self) -> bool:
        """Check if the backend is available."""
        ...

    @property
    @abstractmethod
    def name(self) -> str:
        """Backend name for logging."""
        ...

class E2BBackend(SandboxBackend):
    """E2B Firecracker MicroVM backend (default)."""

    def __init__(self, api_key: str | None = None, pool_size: int = 2):
        self.api_key = api_key
        self.pool_size = pool_size
        self._pool: SandboxPool | None = None

    @property
    def name(self) -> str:
        return "e2b"

    def is_available(self) -> bool:
        try:
            import e2b_code_interpreter
            return True
        except ImportError:
            return False

    def execute(self, func, args, kwargs, timeout=60) -> SandboxResult:
        # Existing execute_in_sandbox logic
        ...

class DockerBackend(SandboxBackend):
    """Docker container backend for enterprise/on-prem."""

    def __init__(self, image: str = "python:3.11-slim", network_mode: str = "none"):
        self.image = image
        self.network_mode = network_mode

    @property
    def name(self) -> str:
        return "docker"

    def is_available(self) -> bool:
        try:
            import docker
            client = docker.from_env()
            client.ping()
            return True
        except Exception:
            return False

    def execute(self, func, args, kwargs, timeout=60) -> SandboxResult:
        import docker
        import cloudpickle
        import base64

        # Serialize and execute in container
        ...

class LocalBackend(SandboxBackend):
    """Local execution (UNSAFE - for development only)."""

    def __init__(self, allow_unsafe: bool = False):
        if not allow_unsafe:
            raise ValueError(
                "LocalBackend is UNSAFE. Set allow_unsafe=True to acknowledge risk."
            )

    @property
    def name(self) -> str:
        return "local_unsafe"

    def is_available(self) -> bool:
        return True

    def execute(self, func, args, kwargs, timeout=60) -> SandboxResult:
        # Direct local execution (no isolation)
        start = time.time()
        try:
            result = func(*args, **kwargs)
            return SandboxResult(
                success=True,
                result=result,
                execution_time_ms=(time.time() - start) * 1000,
            )
        except Exception as e:
            return SandboxResult(
                success=False,
                error=str(e),
            )
```

### 3.2 Enhanced Audit with OTel (~150 lines)

**File:** `src/agent_airlock/audit_otel.py`

```python
from opentelemetry import trace
from opentelemetry.trace import StatusCode

class OTelAuditExporter:
    """Export audit events to OpenTelemetry."""

    def __init__(self, endpoint: str | None = None):
        self.tracer = trace.get_tracer("agent-airlock")
        self.endpoint = endpoint

    def export(self, record: AuditRecord) -> None:
        """Export audit record as OTel span."""
        with self.tracer.start_span(f"airlock.{record.tool_name}") as span:
            span.set_attribute("airlock.tool_name", record.tool_name)
            span.set_attribute("airlock.blocked", record.blocked)
            span.set_attribute("airlock.agent_id", record.agent_id or "unknown")
            span.set_attribute("airlock.session_id", record.session_id or "unknown")
            span.set_attribute("airlock.duration_ms", record.duration_ms or 0)
            span.set_attribute("airlock.sanitized_count", record.sanitized_count)
            span.set_attribute("airlock.truncated", record.truncated)

            if record.blocked:
                span.set_status(StatusCode.ERROR, record.block_reason or "blocked")
                span.set_attribute("airlock.block_reason", record.block_reason or "")
```

**Enhanced AuditRecord fields:**
```python
@dataclass
class AuditRecord:
    # ... existing fields ...

    # V0.4.0 additions
    args_hash: str | None = None  # SHA256 of args (for forensics)
    policy_id: str | None = None
    policy_hash: str | None = None  # Hash of policy config
    redaction_applied: bool = False
    sandbox_backend: str | None = None
    sandbox_id: str | None = None
    egress_domains: list[str] = field(default_factory=list)
    capabilities_required: list[str] = field(default_factory=list)
    capabilities_granted: list[str] = field(default_factory=list)
```

---

## Phase 4: MCP Security

### 4.1 MCPProxyGuard (~200 lines)

**File:** `src/agent_airlock/mcp_proxy_guard.py`

**Goal:** Prevent token passthrough and enforce session binding per MCP security guidance.

```python
from dataclasses import dataclass
from typing import Any

@dataclass
class MCPProxyConfig:
    """Configuration for MCP proxy security."""

    # Token handling
    block_token_passthrough: bool = True
    required_token_audience: str | None = None
    required_token_issuer: str | None = None

    # Session binding
    bind_to_session: bool = True
    session_id_header: str = "X-MCP-Session-ID"
    rotate_session_on_auth: bool = True

    # Consent
    require_consent_for_tools: list[str] | None = None

class MCPProxyGuard:
    """Security guard for MCP proxy servers.

    Implements MCP security best practices:
    - No token passthrough (tokens should be scoped, not forwarded)
    - Session binding (user_id, session_id tracking)
    - Consent hooks for sensitive tools
    """

    def __init__(self, config: MCPProxyConfig | None = None):
        self.config = config or MCPProxyConfig()
        self._sessions: dict[str, MCPSession] = {}

    def validate_request(self, request: dict[str, Any]) -> None:
        """Validate an incoming MCP request."""
        # Check for token passthrough
        if self.config.block_token_passthrough:
            headers = request.get("headers", {})
            for header in ["Authorization", "X-API-Key", "X-Token"]:
                if header in headers:
                    # Check if it's a passthrough (not scoped to this proxy)
                    token = headers[header]
                    if self._is_passthrough_token(token):
                        raise MCPSecurityError(
                            f"Token passthrough detected in {header}. "
                            "Tokens should be scoped to this proxy, not forwarded."
                        )

        # Validate session binding
        if self.config.bind_to_session:
            session_id = request.get("headers", {}).get(self.config.session_id_header)
            if not session_id:
                raise MCPSecurityError(
                    f"Missing session ID header: {self.config.session_id_header}"
                )

    def _is_passthrough_token(self, token: str) -> bool:
        """Check if token appears to be a passthrough."""
        # Check audience/issuer if configured
        if self.config.required_token_audience or self.config.required_token_issuer:
            try:
                import jwt
                decoded = jwt.decode(token, options={"verify_signature": False})

                if self.config.required_token_audience:
                    if decoded.get("aud") != self.config.required_token_audience:
                        return True

                if self.config.required_token_issuer:
                    if decoded.get("iss") != self.config.required_token_issuer:
                        return True
            except Exception:
                pass

        return False
```

---

## Phase 5: CLI & Adoption

### 5.1 airlock doctor CLI (~150 lines)

**File:** `src/agent_airlock/cli/doctor.py`

```python
import click
from pathlib import Path

@click.command()
@click.option("--path", default=".", help="Path to scan")
@click.option("--fix", is_flag=True, help="Apply suggested fixes")
def doctor(path: str, fix: bool):
    """Scan a codebase and suggest Airlock improvements."""

    issues = []
    suggestions = []

    # Scan for tool decorators
    for py_file in Path(path).rglob("*.py"):
        content = py_file.read_text()

        # Check for @tool without @Airlock
        if "@tool" in content and "@Airlock" not in content:
            issues.append(f"{py_file}: @tool decorator without @Airlock protection")
            suggestions.append(f"  Add: from agent_airlock import Airlock")
            suggestions.append(f"  Add: @Airlock() above @tool")

        # Check for **kwargs in tool functions
        if "def " in content and "**kwargs" in content:
            issues.append(f"{py_file}: Function accepts **kwargs (bypasses schema)")

        # Check for unsafe patterns
        if "subprocess.call" in content and "shell=True" in content:
            issues.append(f"{py_file}: subprocess with shell=True (use sandbox)")

    # Report
    click.echo(f"\n=== Airlock Doctor Report ===\n")

    if issues:
        click.echo(f"Found {len(issues)} issues:\n")
        for issue in issues:
            click.echo(f"  ❌ {issue}")

        click.echo(f"\nSuggestions:")
        for suggestion in suggestions:
            click.echo(f"  💡 {suggestion}")
    else:
        click.echo("✅ No issues found!")
```

### 5.2 GitHub Action (~50 lines)

**File:** `.github/actions/airlock-verify/action.yml`

```yaml
name: 'Airlock Verify'
description: 'Verify Airlock protection on tool decorators'

inputs:
  path:
    description: 'Path to scan'
    default: '.'
  fail-on-issues:
    description: 'Fail if issues found'
    default: 'true'

outputs:
  seal-status:
    description: 'Seal status (protected/lite/unsealed)'

runs:
  using: 'composite'
  steps:
    - run: pip install agent-airlock
      shell: bash
    - run: airlock doctor --path ${{ inputs.path }}
      shell: bash
```

---

## Exports

### `__init__.py` Updates

```python
__all__ = [
    # ... existing ...

    # V0.4.0 additions
    "UnknownArgsMode",
    "SafePath", "SafePathInTmp", "SafeURL", "SafeURLAllowHttp",
    "SafePathValidator", "SafeURLValidator",
    "Capability", "CapabilityPolicy", "requires",
    "SandboxBackend", "E2BBackend", "DockerBackend", "LocalBackend",
    "MCPProxyGuard", "MCPProxyConfig",
]
```

---

## Test Plan

| File | Tests | Coverage Target |
|------|-------|-----------------|
| `test_unknown_args.py` | ~40 | UnknownArgsMode, migration, audit events |
| `test_safe_types.py` | ~60 | SafePath, SafeURL, edge cases, attacks |
| `test_capabilities.py` | ~50 | Capability enum, policy, decorator |
| `test_sandbox_backend.py` | ~40 | Backend protocol, Docker, Local |
| `test_audit_otel.py` | ~30 | OTel export, spans, attributes |
| `test_mcp_proxy_guard.py` | ~50 | Token passthrough, session binding |
| `test_cli_doctor.py` | ~30 | Scan, report, suggestions |

**Total new tests:** ~300

---

## Timeline

| Week | Deliverable | Priority |
|------|-------------|----------|
| Week 1 | UnknownArgsMode + SafeTypes | P0 |
| Week 2 | Capabilities + Config integration | P0 |
| Week 3 | SandboxBackend interface | P1 |
| Week 4 | Enhanced Audit + OTel | P1 |
| Week 5 | MCPProxyGuard | P1 |
| Week 6 | CLI + GitHub Action + Tests | P1 |

---

## Version Roadmap

- **v0.4.0-alpha**: UnknownArgsMode, SafeTypes, Capabilities (Week 2)
- **v0.4.0-beta**: SandboxBackend, Audit OTel (Week 4)
- **v0.4.0**: MCPProxyGuard, CLI, full release (Week 6)
- **v0.5.0**: Signed policy bundles, SIEM integrations (Q2 2026)
- **v1.0.0**: Production certified, all integrations (Q3 2026)

---

## Success Criteria

- [ ] `unknown_args=BLOCK` is default in STRICT_POLICY
- [ ] SafePath/SafeURL prevent all traversal attacks
- [ ] Capability gating works with @requires decorator
- [ ] At least 2 sandbox backends (E2B + Docker)
- [ ] OTel audit export functional
- [ ] MCPProxyGuard blocks token passthrough
- [ ] `airlock doctor` CLI working
- [ ] 95%+ test coverage maintained
- [ ] No breaking changes to v0.3 API
