Security: harden public gateway boundaries

docs/PRIVACY.md

@@ -72,7 +72,7 @@ Use an HTTPS tunnel URL. CodePilot still requires the gateway bearer token when
 
 When the iPhone app opens an `http` or `https` link to `localhost`, `127.0.0.1`, or `::1`, the authenticated Mac gateway creates a short-lived proxy session and requests that page from the Mac. Page bodies, URL paths, and query strings then pass through the gateway and the user-owned Cloudflare Tunnel to the iPhone. Local dashboards can contain source code, logs, filenames, prompts, or development data even when the link itself looks generic.
 
-The local-web workflow accepts only loopback hosts; it is not a general website proxy. Session identifiers expire, response sizes are bounded, and the gateway bearer token is not placed in the proxied page URL. These limits reduce exposure but do not make page contents public-safe. Close sensitive local pages when finished and remove local-web URLs, contents, and screenshots from public reports.
+The local-web workflow accepts only loopback hosts; it is not a general website proxy. Session identifiers expire, response sizes are bounded, the gateway bearer token is not placed in the proxied page URL, and the iPhone browser uses a nonpersistent WebKit data store. These limits reduce exposure but do not make page contents public-safe. Close sensitive local pages when finished and remove local-web URLs, contents, and screenshots from public reports.
 
 ## External Services
 

docs/SECURITY.md

@@ -50,7 +50,7 @@ CodePilot swaps local provider auth files. It waits for active turns to finish b
 
 ## Loopback Web Links
 
-CodePilot can proxy `http` and `https` links for `localhost`, `127.0.0.1`, and `::1` through an authenticated, short-lived gateway session. Non-loopback targets are rejected. Treat local page contents, URL paths, and query strings as private because they pass through the gateway and Cloudflare Tunnel to the iPhone.
+CodePilot can proxy `http` and `https` links for `localhost`, `127.0.0.1`, and `::1` through an authenticated, short-lived gateway session. Non-loopback targets are rejected, and the iPhone browser uses a nonpersistent WebKit data store. Treat local page contents, URL paths, and query strings as private because they pass through the gateway and Cloudflare Tunnel to the iPhone.
 
 Do not open local admin panels or dashboards containing credentials through this workflow. Do not share local-web session URLs, page contents, or screenshots in public issues.
 

gateway/codex_phone_gateway.py

@@ -68,6 +68,27 @@
 MAX_TOTAL_ATTACHMENT_BYTES = 50 * 1024 * 1024
 MAX_JSON_BODY_BYTES = 1 * 1024 * 1024
 MAX_ATTACHMENT_REQUEST_BYTES = 72 * 1024 * 1024
+MAX_AUTHORIZATION_HEADER_BYTES = 512
+MIN_APNS_TOKEN_HEX_CHARS = 32
+MAX_APNS_TOKEN_HEX_CHARS = 200
+MAX_APNS_TOPIC_CHARS = 255
+MAX_LIVE_ACTIVITY_ID_CHARS = 128
+LOCAL_WEB_CONTENT_TYPES = {
+    "application/javascript": "application/javascript; charset=utf-8",
+    "application/json": "application/json; charset=utf-8",
+    "application/pdf": "application/pdf",
+    "application/wasm": "application/wasm",
+    "image/avif": "image/avif",
+    "image/gif": "image/gif",
+    "image/jpeg": "image/jpeg",
+    "image/png": "image/png",
+    "image/svg+xml": "image/svg+xml",
+    "image/webp": "image/webp",
+    "text/css": "text/css; charset=utf-8",
+    "text/html": "text/html; charset=utf-8",
+    "text/javascript": "text/javascript; charset=utf-8",
+    "text/plain": "text/plain; charset=utf-8",
+}
 UPLOAD_RETENTION_SECONDS = 7 * 24 * 60 * 60
 VALID_REASONING_EFFORTS = {"none", "minimal", "low", "medium", "high", "xhigh"}
 PUBLIC_JOB_TEXT_LIMIT = 4_000
@@ -1008,16 +1029,20 @@ def file_metadata(path: Path) -> dict:
     }
 
 
+def canonical_local_web_content_type(value: object) -> str:
+    media_type = str(value or "").partition(";")[0].strip().lower()
+    return LOCAL_WEB_CONTENT_TYPES.get(media_type, "application/octet-stream")
+
+
 def file_response(handler: BaseHTTPRequestHandler, path: Path):
     metadata = file_metadata(path)
-    download_name = path.name.replace('"', "_")
     handler.send_response(200)
-    handler.send_header("Content-Type", metadata["mimeType"])
+    handler.send_header("Content-Type", "application/octet-stream")
     handler.send_header("Cache-Control", "no-store")
     handler.send_header("Referrer-Policy", "no-referrer")
     handler.send_header("X-Content-Type-Options", "nosniff")
     handler.send_header("Content-Length", str(metadata["size"]))
-    handler.send_header("Content-Disposition", f'attachment; filename="{download_name}"')
+    handler.send_header("Content-Disposition", "attachment")
     handler.end_headers()
     with path.open("rb") as handle:
         shutil.copyfileobj(handle, handler.wfile)
@@ -1028,8 +1053,9 @@ def local_web_response(handler: BaseHTTPRequestHandler, payload: dict):
     if isinstance(body, str):
         body = body.encode("utf-8")
     status = int(payload.get("status") or 502)
+    content_type = canonical_local_web_content_type(payload.get("contentType"))
     handler.send_response(status)
-    handler.send_header("Content-Type", str(payload.get("contentType") or "application/octet-stream"))
+    handler.send_header("Content-Type", content_type)
     handler.send_header("Cache-Control", "no-store")
     handler.send_header("Referrer-Policy", "no-referrer")
     handler.send_header("X-Content-Type-Options", "nosniff")
@@ -1231,11 +1257,49 @@ def is_connector_auth_warning(text: str) -> bool:
 
 def connector_name_from_text(text: str, fallback: str = "Connector") -> str:
     lower = str(text or "").lower()
-    if "mcp.cloudflare.com" in lower or "cloudflare" in lower:
+    if "cloudflare" in lower:
         return "cloudflare-api"
     return str(fallback or "Connector").strip() or "Connector"
 
 
+def normalized_apns_token(raw_token: object) -> str:
+    token = str(raw_token or "").strip().lower()
+    if (
+        len(token) < MIN_APNS_TOKEN_HEX_CHARS
+        or len(token) > MAX_APNS_TOKEN_HEX_CHARS
+        or len(token) % 2 != 0
+        or re.fullmatch(r"[0-9a-f]+", token) is None
+    ):
+        raise ValueError("APNs token must be an even-length hexadecimal value")
+    return token
+
+
+def validated_apns_topic(raw_topic: object) -> str:
+    topic = str(raw_topic or DEFAULT_APNS_TOPIC).strip() or DEFAULT_APNS_TOPIC
+    segments = topic.split(".")
+    if len(topic) > MAX_APNS_TOPIC_CHARS or len(segments) < 2 or any(
+        not segment
+        or any(
+            not (character.isascii() and (character.isalnum() or character == "-"))
+            for character in segment
+        )
+        for segment in segments
+    ):
+        raise ValueError("APNs bundle identifier is invalid")
+    return topic
+
+
+def validated_live_activity_id(raw_activity_id: object) -> str:
+    activity_id = str(raw_activity_id or "").strip()
+    if (
+        not activity_id
+        or len(activity_id) > MAX_LIVE_ACTIVITY_ID_CHARS
+        or re.fullmatch(r"[A-Za-z0-9._-]+", activity_id) is None
+    ):
+        raise ValueError("Live Activity identifier is invalid")
+    return activity_id
+
+
 def connector_display_name(name: str) -> str:
     raw = str(name or "").strip()
     if raw == "cloudflare-api":
@@ -3277,19 +3341,16 @@ def account_profile_names(self) -> list[str]:
         return sorted(names, key=str.casefold)
 
     def register_notification_device(self, payload: dict) -> dict:
-        raw_token = str(payload.get("token") or "")
-        token = re.sub(r"[^0-9a-fA-F]", "", raw_token).lower()
-        if not token:
-            raise ValueError("Device token is required")
+        token = normalized_apns_token(payload.get("token"))
         environment = str(payload.get("environment") or "production").strip().lower()
         if environment not in {"development", "production"}:
-            environment = "production"
-        bundle_id = str(payload.get("bundleId") or DEFAULT_APNS_TOPIC).strip() or DEFAULT_APNS_TOPIC
+            raise ValueError("Notification environment must be development or production")
+        bundle_id = validated_apns_topic(payload.get("bundleId"))
         device = {
             "token": token,
             "environment": environment,
             "bundleId": bundle_id,
-            "platform": str(payload.get("platform") or "ios"),
+            "platform": "ios",
             "updatedAt": int(time.time()),
         }
         devices = [
@@ -3305,16 +3366,12 @@ def register_notification_device(self, payload: dict) -> dict:
         return {"ok": True, "deviceCount": len(devices[:20])}
 
     def register_live_activity(self, payload: dict) -> dict:
-        activity_id = str(payload.get("activityId") or "").strip()
-        if not activity_id:
-            raise ValueError("Live Activity identifier is required")
-        token = re.sub(r"[^0-9a-fA-F]", "", str(payload.get("pushToken") or "")).lower()
-        if not token:
-            raise ValueError("Live Activity push token is required")
+        activity_id = validated_live_activity_id(payload.get("activityId"))
+        token = normalized_apns_token(payload.get("pushToken"))
         environment = str(payload.get("environment") or "production").strip().lower()
         if environment not in {"development", "production"}:
             raise ValueError("Live Activity environment must be development or production")
-        bundle_id = str(payload.get("bundleId") or DEFAULT_APNS_TOPIC).strip() or DEFAULT_APNS_TOPIC
+        bundle_id = validated_apns_topic(payload.get("bundleId"))
         existing = next(
             (item for item in self.read_live_activities() if item.get("activityId") == activity_id),
             None,
@@ -4912,7 +4969,14 @@ def authenticate(self) -> bool:
     def is_authenticated(self) -> bool:
         expected = self.state().token
         header = self.headers.get("authorization", "")
-        return secrets.compare_digest(header, f"Bearer {expected}")
+        try:
+            header_bytes = header.encode("ascii")
+            expected_bytes = f"Bearer {expected}".encode("ascii")
+        except (AttributeError, UnicodeEncodeError):
+            return False
+        if len(header_bytes) > MAX_AUTHORIZATION_HEADER_BYTES:
+            return False
+        return secrets.compare_digest(header_bytes, expected_bytes)
 
     def do_GET(self):
         parsed = urllib.parse.urlparse(self.path)

gateway/remote_desktop_gateway.py

@@ -200,6 +200,13 @@ def __init__(
         self._signals: dict[str, list[dict[str, Any]]] = {}
         self._last_signal_sequence: dict[str, int] = {}
 
+    def public_status(self) -> dict[str, Any]:
+        """Return policy state without probing the privileged native host."""
+        return {
+            "available": self.remote_control_enabled,
+            "remoteControlAvailable": self.remote_control_enabled,
+        }
+
     def handle(
         self,
         http_method: str,

gateway/test_codex_phone_gateway.py

@@ -1295,6 +1295,10 @@ def test_diagnostic_health_retains_authenticated_setup_status(self):
             self.assertIn("auth", health["accounts"])
             self.assertIn("notifications", health)
             self.assertIn("remoteDesktop", health)
+            self.assertEqual(
+                health["remoteDesktop"],
+                {"available": False, "remoteControlAvailable": False},
+            )
             self.assertIn("localWeb", health)
 
     def test_health_endpoint_is_available_without_bearer_token(self):
@@ -1391,6 +1395,14 @@ def test_health_endpoint_returns_diagnostics_with_bearer_token(self):
             self.assertIn("accounts", payload)
             self.assertIn("notifications", payload)
 
+    def test_authentication_rejects_non_ascii_header_without_raising(self):
+        state = GatewayState(Path("/tmp/codex"), "secret-token", Path("/missing-codex"), False)
+        handler = object.__new__(gateway.Handler)
+        handler.server = type("Server", (), {"state": state})()
+        handler.headers = {"authorization": "Bearer secrét-token"}
+
+        self.assertFalse(handler.is_authenticated())
+
     def test_error_payload_has_stable_code_message_and_recovery(self):
         payload = gateway.error_payload(
             "local_web_invalid_target",
@@ -1461,6 +1473,48 @@ def test_file_metadata_reports_download_details(self):
             self.assertEqual(metadata["mimeType"], "text/plain")
             self.assertEqual(metadata["size"], 5)
 
+    def test_file_response_does_not_include_filename_in_header(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            file_path = Path(tmp) / 'report\nX-Injected: yes.txt'
+            file_path.write_text("hello", encoding="utf-8")
+            handler = mock.Mock()
+            handler.wfile = io.BytesIO()
+
+            gateway.file_response(handler, file_path)
+
+            headers = dict(call.args for call in handler.send_header.call_args_list)
+            self.assertEqual(headers["Content-Type"], "application/octet-stream")
+            self.assertEqual(headers["Content-Disposition"], "attachment")
+            self.assertNotIn("\n", headers["Content-Disposition"])
+            self.assertNotIn("\r", headers["Content-Disposition"])
+
+    def test_local_web_response_rejects_control_characters_in_content_type(self):
+        handler = mock.Mock()
+        handler.wfile = io.BytesIO()
+
+        gateway.local_web_response(handler, {
+            "status": 200,
+            "contentType": "text/plain\r\nX-Injected: yes",
+            "body": b"hello",
+        })
+
+        headers = dict(call.args for call in handler.send_header.call_args_list)
+        self.assertEqual(headers["Content-Type"], "application/octet-stream")
+        self.assertNotIn("X-Injected", "\n".join(headers.values()))
+
+    def test_local_web_response_canonicalizes_allowed_content_type(self):
+        handler = mock.Mock()
+        handler.wfile = io.BytesIO()
+
+        gateway.local_web_response(handler, {
+            "status": 200,
+            "contentType": "Text/HTML; charset=untrusted",
+            "body": b"<p>hello</p>",
+        })
+
+        headers = dict(call.args for call in handler.send_header.call_args_list)
+        self.assertEqual(headers["Content-Type"], "text/html; charset=utf-8")
+
     def test_local_web_session_rejects_non_loopback_url(self):
         state = GatewayState(Path("/tmp/codex"), "token", Path("/missing-codex"), False)
 
@@ -2047,19 +2101,50 @@ def test_notification_device_registration_persists_token(self):
                 state = GatewayState(tmp_path / "codex", "token", Path("/missing-codex"), False)
 
                 response = state.register_notification_device({
-                    "token": " ABC123 ",
+                    "token": " " + ("AB" * 32) + " ",
                     "environment": "production",
                     "bundleId": "io.codepilot.iOS",
                 })
 
                 self.assertTrue(response["ok"])
                 devices = json.loads((switcher_home / "phone-notification-devices.json").read_text(encoding="utf-8"))
-                self.assertEqual(devices[0]["token"], "abc123")
+                self.assertEqual(devices[0]["token"], "ab" * 32)
                 self.assertEqual(devices[0]["environment"], "production")
                 self.assertEqual(devices[0]["bundleId"], "io.codepilot.iOS")
             finally:
                 gateway.DEFAULT_SWITCHER_HOME = old_home
 
+    def test_notification_registration_rejects_malformed_or_unbounded_fields(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_path = Path(tmp)
+            old_home = gateway.DEFAULT_SWITCHER_HOME
+            gateway.DEFAULT_SWITCHER_HOME = tmp_path / "switcher"
+            try:
+                state = GatewayState(tmp_path / "codex", "token", Path("/missing-codex"), False)
+                valid_token = "ab" * 32
+
+                invalid_payloads = [
+                    {"token": "ab:not-a-token", "environment": "production"},
+                    {"token": valid_token, "environment": "staging"},
+                    {"token": valid_token, "environment": "production", "bundleId": "invalid\nheader"},
+                    {"token": valid_token, "environment": "production", "bundleId": "a" * 256},
+                ]
+                for payload in invalid_payloads:
+                    with self.subTest(payload=payload):
+                        with self.assertRaises(ValueError):
+                            state.register_notification_device(payload)
+
+                with self.assertRaisesRegex(ValueError, "identifier"):
+                    state.register_live_activity({
+                        "activityId": "../unsafe",
+                        "pushToken": valid_token,
+                        "environment": "production",
+                    })
+                self.assertFalse((gateway.DEFAULT_SWITCHER_HOME / "phone-notification-devices.json").exists())
+                self.assertFalse((gateway.DEFAULT_SWITCHER_HOME / "phone-live-activities.json").exists())
+            finally:
+                gateway.DEFAULT_SWITCHER_HOME = old_home
+
     def test_live_activity_registration_is_separate_and_idempotent(self):
         with tempfile.TemporaryDirectory() as tmp:
             tmp_path = Path(tmp)
@@ -2070,18 +2155,18 @@ def test_live_activity_registration_is_separate_and_idempotent(self):
                 state = GatewayState(tmp_path / "codex", "token", Path("/missing-codex"), False)
                 payload = {
                     "activityId": "activity-1",
-                    "pushToken": " ABC123 ",
+                    "pushToken": " " + ("AB" * 32) + " ",
                     "environment": "production",
                     "bundleId": "io.codepilot.iOS",
                 }
 
                 first = state.register_live_activity(payload)
-                second = state.register_live_activity({**payload, "pushToken": "def456"})
+                second = state.register_live_activity({**payload, "pushToken": "de" * 32})
 
                 self.assertEqual(first["activityCount"], 1)
                 self.assertEqual(second["activityCount"], 1)
                 registrations = state.read_live_activities()
-                self.assertEqual(registrations[0]["pushToken"], "def456")
+                self.assertEqual(registrations[0]["pushToken"], "de" * 32)
                 self.assertFalse((switcher_home / "phone-notification-devices.json").exists())
             finally:
                 gateway.DEFAULT_SWITCHER_HOME = old_home
@@ -2096,7 +2181,7 @@ def test_live_activity_registration_rejects_invalid_environment(self):
                 with self.assertRaisesRegex(ValueError, "environment"):
                     state.register_live_activity({
                         "activityId": "activity-1",
-                        "pushToken": "abc123",
+                        "pushToken": "ab" * 32,
                         "environment": "staging",
                     })
             finally:
@@ -2118,7 +2203,7 @@ def test_unchanged_live_activity_state_is_not_pushed_twice(self):
                 )
                 state.register_live_activity({
                     "activityId": "activity-1",
-                    "pushToken": "abc123",
+                    "pushToken": "ab" * 32,
                     "environment": "production",
                     "bundleId": "io.codepilot.iOS",
                 })

gateway/test_remote_desktop_gateway.py

@@ -213,6 +213,16 @@ def call(self, method, payload=None):
 
 
 class RemoteDesktopGatewayTests(unittest.TestCase):
+    def test_public_status_reports_fail_closed_policy_without_probing_host(self):
+        host = FakeNativeHostClient()
+        gateway = RemoteDesktopGateway(host_client=host)
+
+        self.assertEqual(
+            gateway.public_status(),
+            {"available": False, "remoteControlAvailable": False},
+        )
+        self.assertEqual(host.calls, [])
+
     def test_remote_control_routes_fail_closed_by_default(self):
         host = FakeNativeHostClient()
         gateway = RemoteDesktopGateway(host_client=host)