fix(connectors): harden media response headers

Route signed connector media through a shared MIME-aware response helper so dangerous active content downloads with octet-stream and nosniff headers while safe images remain inline-compatible.

Preserve media token, expiry, and path-boundary checks for LINE and WhatsApp media routes.

Validation: targeted connector media tests passed; full Windows test gate passed.

connector/media_response.py

@@ -0,0 +1,66 @@
+"""Safe response helpers for connector-served local media."""
+
+from __future__ import annotations
+
+import mimetypes
+from pathlib import Path
+from typing import Any
+
+DANGEROUS_CONTENT_TYPES = {
+    "text/html",
+    "text/html-sandboxed",
+    "application/xhtml+xml",
+    "text/javascript",
+    "application/javascript",
+    "application/x-javascript",
+    "application/ecmascript",
+    "text/css",
+    "image/svg+xml",
+    "application/xml",
+    "text/xml",
+    "message/rfc822",
+}
+
+
+def is_dangerous_content_type(content_type: str | None) -> bool:
+    """Return True for browser-renderable active content types."""
+    if not content_type:
+        return False
+    normalized = content_type.split(";", 1)[0].strip().lower()
+    if normalized in DANGEROUS_CONTENT_TYPES:
+        return True
+    return normalized.endswith("+xml") or normalized.endswith("/xml")
+
+
+def _content_disposition_filename(name: str) -> str:
+    safe_name = name.replace("\r", "").replace("\n", "")
+    safe_name = safe_name.replace("\\", "\\\\").replace('"', '\\"')
+    return f'filename="{safe_name}"'
+
+
+def build_connector_media_response(web: Any, path: Path):
+    """Build a hardened FileResponse for signed connector media files."""
+    content_type = mimetypes.guess_type(str(path))[0] or "application/octet-stream"
+    disposition = _content_disposition_filename(path.name)
+
+    # IMPORTANT: connector media is user-controlled. Dangerous active content
+    # must download instead of rendering inline in the local media origin.
+    if is_dangerous_content_type(content_type):
+        content_type = "application/octet-stream"
+        disposition = f"attachment; {_content_disposition_filename(path.name)}"
+
+    return web.FileResponse(
+        path,
+        headers={
+            "Content-Disposition": disposition,
+            "Content-Type": content_type,
+            "X-Content-Type-Options": "nosniff",
+        },
+    )
+
+
+__all__ = [
+    "DANGEROUS_CONTENT_TYPES",
+    "build_connector_media_response",
+    "is_dangerous_content_type",
+]

connector/platforms/line_webhook.py

@@ -13,6 +13,7 @@ from typing import Optional
 
 from ..config import ConnectorConfig
 from ..contract import CommandRequest, CommandResponse
+from ..media_response import build_connector_media_response
 from ..router import CommandRouter
 from ..security_profile import AllowlistPolicy, ReplayGuard, verify_hmac_signature
 from ..transport_contract import RelayResponseClassifier
@@ -175,7 +176,7 @@ class LINEWebhookServer:
         if not path:
             return web.Response(status=404, text="Media Not Found or Expired")
 
-        return web.FileResponse(path)
+        return build_connector_media_response(web, path)
 
     async def _process_event(self, event: dict):
         """Convert LINE event to CommandRequest and route."""

connector/platforms/whatsapp_webhook.py

@@ -18,6 +18,7 @@ from typing import Optional
 
 from ..config import ConnectorConfig
 from ..contract import CommandRequest, CommandResponse
+from ..media_response import build_connector_media_response
 from ..router import CommandRouter
 from ..security_profile import AllowlistPolicy, ReplayGuard, verify_hmac_signature
 from ..transport_contract import RelayResponseClassifier
@@ -219,7 +220,7 @@ class WhatsAppWebhookServer:
         if not path:
             return web.Response(status=404, text="Media Not Found or Expired")
 
-        return web.FileResponse(path)
+        return build_connector_media_response(web, path)
 
     # ------------------------------------------------------------------
     # Message Processing

tests/connector/test_media_delivery.py

@@ -12,11 +12,35 @@ from pathlib import Path
 from unittest.mock import AsyncMock, MagicMock, patch
 
 from connector.config import ConnectorConfig
+from connector.media_response import (
+    build_connector_media_response,
+    is_dangerous_content_type,
+)
 from connector.media_store import MediaStore
 from connector.platforms.line_webhook import LINEWebhookServer
+from connector.platforms.whatsapp_webhook import WhatsAppWebhookServer
 from connector.router import CommandRouter
 
 
+class _FakeRequest:
+    def __init__(self, token: str):
+        self.match_info = {"token": token}
+
+
+class _FakeWeb:
+    class Response:
+        def __init__(self, status=200, text=""):
+            self.status = status
+            self.text = text
+            self.headers = {}
+
+    class FileResponse:
+        def __init__(self, path, headers=None):
+            self.status = 200
+            self.path = Path(path)
+            self.headers = headers or {}
+
+
 class TestMediaStore(unittest.TestCase):
     def setUp(self):
         self.tmp_dir = tempfile.mkdtemp()
@@ -81,6 +105,82 @@ class TestMediaStore(unittest.TestCase):
             self.store.store_image(b"123", ".png", "ch1")
 
 
+class TestConnectorMediaResponse(unittest.TestCase):
+    def setUp(self):
+        self.tmp_dir = tempfile.mkdtemp()
+
+    def tearDown(self):
+        shutil.rmtree(self.tmp_dir)
+
+    def _write_file(self, name: str, data: bytes = b"payload") -> Path:
+        path = Path(self.tmp_dir) / name
+        path.write_bytes(data)
+        return path
+
+    def test_dangerous_content_type_normalization(self):
+        for content_type in (
+            "text/html; charset=utf-8",
+            "TEXT/HTML",
+            "image/svg+xml; charset=utf-8",
+            "application/rss+xml",
+            "application/xml",
+            "message/rfc822",
+        ):
+            self.assertTrue(is_dangerous_content_type(content_type), content_type)
+
+        for content_type in ("image/png", "image/jpeg", "image/webp", "text/plain"):
+            self.assertFalse(is_dangerous_content_type(content_type), content_type)
+
+    def test_dangerous_media_forces_attachment_octet_stream_and_nosniff(self):
+        for filename in (
+            "evil.html",
+            "evil.svg",
+            "evil.js",
+            "evil.css",
+            "evil.xml",
+        ):
+            with self.subTest(filename=filename):
+                response = build_connector_media_response(
+                    _FakeWeb, self._write_file(filename)
+                )
+
+                self.assertEqual(
+                    response.headers["Content-Type"], "application/octet-stream"
+                )
+                self.assertEqual(response.headers["X-Content-Type-Options"], "nosniff")
+                self.assertIn(
+                    "attachment", response.headers["Content-Disposition"].lower()
+                )
+
+    def test_safe_images_remain_inline_compatible_with_nosniff(self):
+        for filename, expected_type in (
+            ("safe.png", "image/png"),
+            ("safe.jpg", "image/jpeg"),
+            ("safe.webp", "image/webp"),
+            ("safe.gif", "image/gif"),
+        ):
+            with self.subTest(filename=filename):
+                response = build_connector_media_response(
+                    _FakeWeb, self._write_file(filename)
+                )
+
+                self.assertEqual(response.headers["Content-Type"], expected_type)
+                self.assertEqual(response.headers["X-Content-Type-Options"], "nosniff")
+                self.assertNotIn(
+                    "attachment", response.headers["Content-Disposition"].lower()
+                )
+                self.assertIn("filename=", response.headers["Content-Disposition"])
+
+    def test_content_disposition_filename_is_escaped(self):
+        response = build_connector_media_response(
+            _FakeWeb, Path(self.tmp_dir) / 'bad"name.svg'
+        )
+
+        self.assertIn(
+            r'filename="bad\"name.svg"', response.headers["Content-Disposition"]
+        )
+
+
 class TestLINESendImage(unittest.IsolatedAsyncioTestCase):
     def setUp(self):
         self.config = ConnectorConfig()
@@ -129,5 +229,83 @@ class TestLINESendImage(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(url, "https://example.com/media/mock_token.sig")
 
 
+class TestConnectorMediaRoutes(unittest.IsolatedAsyncioTestCase):
+    def setUp(self):
+        self.tmp_dir = tempfile.mkdtemp()
+        self.config = ConnectorConfig()
+        self.router = MagicMock(spec=CommandRouter)
+
+    def tearDown(self):
+        shutil.rmtree(self.tmp_dir)
+
+    def _write_file(self, name: str, data: bytes = b"payload") -> Path:
+        path = Path(self.tmp_dir) / name
+        path.write_bytes(data)
+        return path
+
+    async def test_line_media_route_hardens_dangerous_media_after_token_validation(
+        self,
+    ):
+        server = LINEWebhookServer(self.config, self.router)
+        server.media_store = MagicMock()
+        server.media_store.get_image_path.return_value = self._write_file("evil.svg")
+
+        with patch(
+            "connector.platforms.line_webhook._import_aiohttp_web",
+            return_value=(None, _FakeWeb),
+        ):
+            response = await server._handle_media_request(_FakeRequest("signed.token"))
+
+        server.media_store.get_image_path.assert_called_once_with("signed.token")
+        self.assertEqual(response.status, 200)
+        self.assertEqual(response.headers["Content-Type"], "application/octet-stream")
+        self.assertIn("attachment", response.headers["Content-Disposition"].lower())
+        self.assertEqual(response.headers["X-Content-Type-Options"], "nosniff")
+
+    async def test_whatsapp_media_route_preserves_safe_image_delivery(self):
+        server = WhatsAppWebhookServer(self.config, self.router)
+        server.media_store = MagicMock()
+        server.media_store.get_image_path.return_value = self._write_file("safe.png")
+
+        with patch(
+            "connector.platforms.whatsapp_webhook._import_aiohttp_web",
+            return_value=(None, _FakeWeb),
+        ):
+            response = await server._handle_media_request(_FakeRequest("signed.token"))
+
+        server.media_store.get_image_path.assert_called_once_with("signed.token")
+        self.assertEqual(response.status, 200)
+        self.assertEqual(response.headers["Content-Type"], "image/png")
+        self.assertNotIn("attachment", response.headers["Content-Disposition"].lower())
+        self.assertEqual(response.headers["X-Content-Type-Options"], "nosniff")
+
+    async def test_media_routes_keep_invalid_tokens_fail_closed(self):
+        for server_cls, patch_target in (
+            (
+                LINEWebhookServer,
+                "connector.platforms.line_webhook._import_aiohttp_web",
+            ),
+            (
+                WhatsAppWebhookServer,
+                "connector.platforms.whatsapp_webhook._import_aiohttp_web",
+            ),
+        ):
+            with self.subTest(server=server_cls.__name__):
+                server = server_cls(self.config, self.router)
+                server.media_store = MagicMock()
+                server.media_store.get_image_path.return_value = None
+
+                with patch(patch_target, return_value=(None, _FakeWeb)):
+                    response = await server._handle_media_request(
+                        _FakeRequest("expired.token")
+                    )
+
+                server.media_store.get_image_path.assert_called_once_with(
+                    "expired.token"
+                )
+                self.assertEqual(response.status, 404)
+                self.assertEqual(response.text, "Media Not Found or Expired")
+
+
 if __name__ == "__main__":
     unittest.main()