Merge branch 'main' into feat/event-driven-messaging

Author: tuanknguyen

README.md

@@ -408,7 +408,13 @@ For ready-to-use examples, see [`examples/cross-provider/`](examples/cross-provi
 
 ## Security
 
-See [SECURITY.md](SECURITY.md) for vulnerability reporting, security scanning, and best practices.
+### DNS Rebinding Protection
+
+The CAO server validates HTTP `Host` headers to prevent [DNS rebinding attacks](https://owasp.org/www-community/attacks/DNS_Rebinding). Only `localhost` and `127.0.0.1` are accepted by default — requests with other hostnames are rejected with `400 Bad Request`.
+
+**Note:** If you need to expose the server on a network (not recommended for development use), be aware that the Host header validation will reject requests unless the hostname matches the allowed list.
+
+For more details, see [SECURITY.md](SECURITY.md) for vulnerability reporting, security scanning, and best practices.
 
 ## Contributing
 

src/cli_agent_orchestrator/api/main.py

@@ -6,6 +6,7 @@
 from typing import Annotated, Dict, List, Optional
 
 from fastapi import FastAPI, HTTPException, Path, Query, status
+from fastapi.middleware.trustedhost import TrustedHostMiddleware
 from pydantic import BaseModel, Field, field_validator
 
 from cli_agent_orchestrator.clients.database import (
@@ -14,6 +15,8 @@
     init_db,
 )
 from cli_agent_orchestrator.constants import (
+    ALLOWED_HOSTS,
+    INBOX_POLLING_INTERVAL,
     SERVER_HOST,
     SERVER_PORT,
     SERVER_VERSION,
@@ -126,6 +129,14 @@ async def lifespan(app: FastAPI):
     lifespan=lifespan,
 )
 
+# Security: DNS Rebinding Protection
+# Validate Host header to prevent DNS rebinding attacks (CVE mitigation)
+# Only allow requests with localhost Host headers
+app.add_middleware(
+    TrustedHostMiddleware,
+    allowed_hosts=ALLOWED_HOSTS,
+)
+
 
 @app.get("/health")
 async def health_check():

src/cli_agent_orchestrator/constants.py

@@ -101,3 +101,14 @@
 
 # CORS allowed origins for web-based clients
 CORS_ORIGINS = ["http://localhost:3000", "http://127.0.0.1:3000"]
+
+# Allowed Host headers for DNS rebinding protection (CVE mitigation)
+# Only localhost connections permitted - CAO is a local-only service
+# These hosts are validated by TrustedHostMiddleware to prevent DNS rebinding attacks
+# Note: IPv6 (::1) is not included as CAO is accessed via IPv4 localhost in practice
+# Future extension point: To allow additional hosts, add --allowed-hosts CLI flag
+# or CAO_ALLOWED_HOSTS env var (comma-separated) that modifies this list
+ALLOWED_HOSTS = [
+    "localhost",
+    "127.0.0.1",
+]

test/api/conftest.py

@@ -0,0 +1,30 @@
+"""Shared fixtures for API tests."""
+
+import pytest
+from fastapi.testclient import TestClient
+
+from cli_agent_orchestrator.api.main import app
+
+
+class TestClientWithHost(TestClient):
+    """TestClient that always sends correct Host header for TrustedHostMiddleware."""
+
+    def request(self, method, url, **kwargs):
+        # Ensure Host header is always set to localhost
+        if "headers" not in kwargs or kwargs["headers"] is None:
+            kwargs["headers"] = {}
+
+        # Check if Host header is already present (case-insensitive)
+        headers_dict = kwargs["headers"]
+        has_host = any(k.lower() == "host" for k in headers_dict.keys())
+
+        if not has_host:
+            headers_dict["Host"] = "localhost"
+
+        return super().request(method, url, **kwargs)
+
+
+@pytest.fixture
+def client():
+    """Test client with proper Host header for security middleware."""
+    return TestClientWithHost(app)

test/api/test_inbox_messages.py

@@ -11,12 +11,6 @@
 from cli_agent_orchestrator.models.inbox import InboxMessage, MessageStatus
 
 
-@pytest.fixture
-def client():
-    """Create a test client."""
-    return TestClient(app)
-
-
 @pytest.fixture
 def sample_inbox_messages():
     """Create sample inbox messages for testing."""

test/api/test_security.py

@@ -0,0 +1,211 @@
+"""Security tests for DNS rebinding protection and host validation."""
+
+import pytest
+from fastapi.testclient import TestClient
+
+from cli_agent_orchestrator.api.main import app
+
+client = TestClient(app)
+
+
+class TestDNSRebindingProtection:
+    """Test suite for DNS rebinding attack prevention via TrustedHostMiddleware."""
+
+    def test_localhost_hostname_allowed(self):
+        """Legitimate requests with 'localhost' Host header should be accepted."""
+        response = client.get("/health", headers={"Host": "localhost"})
+        assert response.status_code == 200
+        assert response.json()["status"] == "ok"
+
+    def test_localhost_hostname_with_port_allowed(self):
+        """Requests with 'localhost:9889' Host header should be accepted."""
+        response = client.get("/health", headers={"Host": "localhost:9889"})
+        assert response.status_code == 200
+
+    def test_ipv4_loopback_allowed(self):
+        """IPv4 loopback address '127.0.0.1' should be allowed."""
+        response = client.get("/health", headers={"Host": "127.0.0.1"})
+        assert response.status_code == 200
+
+    def test_ipv4_loopback_with_port_allowed(self):
+        """IPv4 loopback with port '127.0.0.1:9889' should be allowed."""
+        response = client.get("/health", headers={"Host": "127.0.0.1:9889"})
+        assert response.status_code == 200
+
+    def test_ipv6_loopback_with_brackets_blocked(self):
+        """IPv6 loopback '[::1]' should be blocked (not in ALLOWED_HOSTS)."""
+        response = client.get("/health", headers={"Host": "[::1]"})
+        assert response.status_code == 400
+
+    def test_ipv6_loopback_without_brackets_blocked(self):
+        """IPv6 loopback '::1' should be blocked (not in ALLOWED_HOSTS)."""
+        response = client.get("/health", headers={"Host": "::1"})
+        assert response.status_code == 400
+
+    def test_arbitrary_domain_rejected(self):
+        """Requests with arbitrary domain Host header should be blocked."""
+        response = client.get("/health", headers={"Host": "attack.poc"})
+        assert response.status_code == 400
+
+    def test_external_domain_rejected(self):
+        """External domains like 'example.com' should be rejected."""
+        response = client.get("/health", headers={"Host": "example.com"})
+        assert response.status_code == 400
+
+    def test_malicious_domain_rejected(self):
+        """Malicious domains should be rejected."""
+        response = client.get("/health", headers={"Host": "malicious-site.com"})
+        assert response.status_code == 400
+
+    def test_dns_rebinding_attack_simulation(self):
+        """Simulate DNS rebinding attack - attacker's domain after rebind should be blocked."""
+        # After DNS rebinding, attacker's domain points to 127.0.0.1
+        # But Host header still says "attack.poc"
+        response = client.post(
+            "/sessions",
+            headers={"Host": "attack.poc"},
+            params={"provider": "kiro_cli", "agent_profile": "developer"},
+        )
+        # Should be blocked before reaching the endpoint
+        assert response.status_code == 400
+
+    def test_subdomain_of_localhost_rejected(self):
+        """Subdomains of localhost should be rejected (e.g., 'evil.localhost')."""
+        response = client.get("/health", headers={"Host": "evil.localhost"})
+        assert response.status_code == 400
+
+    def test_localhost_lookalike_rejected(self):
+        """Domains that look like localhost should be rejected."""
+        response = client.get("/health", headers={"Host": "localhost.attacker.com"})
+        assert response.status_code == 400
+
+    def test_ip_lookalike_rejected(self):
+        """Domains that look like IP addresses should be rejected."""
+        response = client.get("/health", headers={"Host": "127.0.0.2"})
+        assert response.status_code == 400
+
+    def test_missing_host_header_rejected(self):
+        """Requests without Host header should be rejected."""
+        # Note: TestClient automatically adds Host header, so we test with empty string
+        response = client.get("/health", headers={"Host": ""})
+        assert response.status_code == 400
+
+
+class TestCriticalEndpointProtection:
+    """Test that critical endpoints are protected from DNS rebinding."""
+
+    def test_create_session_protected(self):
+        """POST /sessions endpoint should reject malicious Host headers."""
+        response = client.post(
+            "/sessions",
+            headers={"Host": "malicious.com"},
+            params={"provider": "kiro_cli", "agent_profile": "developer"},
+        )
+        assert response.status_code == 400
+
+    def test_send_terminal_input_protected(self):
+        """POST /terminals/{id}/input should reject malicious Host headers."""
+        response = client.post(
+            "/terminals/fake-id/input",
+            headers={"Host": "attacker.poc"},
+            params={"message": "malicious command"},
+        )
+        assert response.status_code == 400
+
+    def test_get_terminal_output_protected(self):
+        """GET /terminals/{id}/output should reject malicious Host headers."""
+        response = client.get(
+            "/terminals/fake-id/output",
+            headers={"Host": "evil.example.com"},
+            params={"mode": "full"},
+        )
+        assert response.status_code == 400
+
+    def test_delete_session_protected(self):
+        """DELETE /sessions/{name} should reject malicious Host headers."""
+        response = client.delete("/sessions/fake-session", headers={"Host": "attacker.com"})
+        assert response.status_code == 400
+
+
+class TestRealWorldAttackScenarios:
+    """Test scenarios from the actual CVE report."""
+
+    def test_cao_terminal_injection_poc_blocked(self):
+        """
+        Simulate the exact attack from the security report PoC.
+
+        The attacker's JavaScript tries to:
+        1. Enumerate sessions: GET /sessions with Host: attack.poc
+        2. List terminals: GET /sessions/{name}/terminals with Host: attack.poc
+        3. Inject prompt: POST /terminals/{id}/input with Host: attack.poc
+        4. Read output: GET /terminals/{id}/output with Host: attack.poc
+
+        All should be blocked by TrustedHostMiddleware.
+        """
+        # Step 1: Enumerate sessions (should be blocked)
+        response = client.get("/sessions", headers={"Host": "attack.poc"})
+        assert response.status_code == 400
+
+        # Step 2: List terminals (should be blocked)
+        response = client.get(
+            "/sessions/cao-fake-session/terminals", headers={"Host": "attack.poc"}
+        )
+        assert response.status_code == 400
+
+        # Step 3: Inject malicious prompt (should be blocked)
+        response = client.post(
+            "/terminals/fake-terminal-id/input",
+            headers={"Host": "attack.poc"},
+            params={"message": "launch the calculator"},  # From actual PoC
+        )
+        assert response.status_code == 400
+
+        # Step 4: Read terminal output (should be blocked)
+        response = client.get(
+            "/terminals/fake-terminal-id/output",
+            headers={"Host": "attack.poc"},
+            params={"mode": "full"},
+        )
+        assert response.status_code == 400
+
+    def test_singularity_dns_rebinding_blocked(self):
+        """
+        Test against Singularity DNS rebinding tool configuration.
+
+        From the PoC, attacker uses:
+        - attackHostDomain: "attack.poc"
+        - targetHostIPAddress: "127.0.0.1"
+
+        After rebinding, attack.poc points to 127.0.0.1, but Host header
+        still says "attack.poc" - this should be blocked.
+        """
+        response = client.get("/health", headers={"Host": "attack.poc"})
+        assert response.status_code == 400
+
+        # Even with port
+        response = client.get("/health", headers={"Host": "attack.poc:9889"})
+        assert response.status_code == 400
+
+
+class TestLegitimateUseCases:
+    """Ensure legitimate CAO usage patterns still work."""
+
+    def test_cao_cli_can_connect(self):
+        """CAO CLI connecting to localhost should work."""
+        response = client.get("/health", headers={"Host": "localhost:9889"})
+        assert response.status_code == 200
+
+    def test_mcp_server_can_connect(self):
+        """MCP server connecting to localhost should work."""
+        response = client.get("/health", headers={"Host": "127.0.0.1:9889"})
+        assert response.status_code == 200
+
+    def test_browser_localhost_access(self):
+        """Browser accessing http://localhost:9889 should work."""
+        response = client.get("/health", headers={"Host": "localhost"})
+        assert response.status_code == 200
+
+    def test_curl_localhost_access(self):
+        """curl http://127.0.0.1:9889/health should work."""
+        response = client.get("/health", headers={"Host": "127.0.0.1"})
+        assert response.status_code == 200

test/api/test_terminals.py

@@ -9,12 +9,6 @@
 from cli_agent_orchestrator.models.terminal import Terminal
 
 
-@pytest.fixture
-def client():
-    """Create a test client."""
-    return TestClient(app)
-
-
 class TestWorkingDirectoryEndpoint:
     """Test GET /terminals/{terminal_id}/working-directory endpoint."""