# Licensed to the Software Freedom Conservancy (SFC) under one # or more contributor license agreements. See the NOTICE file # distributed with this work for additional information # regarding copyright ownership. The SFC licenses this file # to you under the Apache License, Version 2.0 (the # "License"); you may not use this file except in compliance # with the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. """High-level request/response interception helpers for the WebDriver BiDi network module. This module is copied verbatim into the generated ``selenium.webdriver.common.bidi`` package by Bazel (see ``create-bidi-src`` in ``py/BUILD.bazel``). The generated ``network`` module re-exports :class:`Request` and :class:`Response` and instantiates the handler registries, which layer a user-friendly handler API on top of the CDDL-generated low-level commands (``network.addIntercept``, ``network.continueRequest``, ``network.continueResponse``, ``network.failRequest``, ``network.provideResponse``). Handlers registered through :meth:`RequestHandlerRegistry.add_handler` receive a :class:`Request` and may observe it, mutate it, fail it, or stub a response. After every matching handler has run, the registry reconciles the recorded outcome and issues exactly one BiDi command per request: 1. If any handler called :meth:`Request.fail`, the request is failed. 2. Else if any handler called :meth:`Request.provide_response`, the stubbed response is provided. 3. Else if any handler mutated the request, it is continued with the mutations. 4. Otherwise the request is continued unmodified. Handlers registered through :meth:`ResponseHandlerRegistry.add_handler` receive a :class:`Response` at the ``responseStarted`` phase and may observe or mutate it. Reconciliation works the same way: a mutated body requires ``network.provideResponse`` (the wire protocol cannot continue a response with a new body), other mutations are applied via ``network.continueResponse``, and untouched responses are continued unmodified. Handlers registered through :meth:`AuthHandlerRegistry.add_handler` receive an :class:`AuthenticationRequest` at the ``authRequired`` phase and may call :meth:`AuthenticationRequest.provide_credentials` or :meth:`AuthenticationRequest.cancel`. Reconciliation issues exactly one ``network.continueWithAuth`` command per challenge: ``cancel`` takes precedence over provided credentials, and if no handler responded the challenge is continued with action ``default`` so the browser's own behavior (usually the authentication prompt) applies. Extra headers registered through :meth:`RequestHandlerRegistry.set_extra_header` are merged into every subsequent request. BiDi has no dedicated command for this, so the registry pauses each request at ``beforeRequestSent`` with a match-everything intercept and merges the headers while reconciling — the same single continue cycle that applies user handler mutations. This mirrors the reconciliation rules in the cross-binding BiDi API design and means purely observational handlers never stall the page. """ from __future__ import annotations import logging import re from collections.abc import Callable from typing import Any from selenium.webdriver.common.bidi.common import command_builder logger = logging.getLogger(__name__) # Event names accepted by the legacy phase-based add_request_handler API. LEGACY_REQUEST_HANDLER_EVENTS = ("auth_required", "before_request", "before_request_sent") def looks_like_url_glob(value: Any) -> bool: """Heuristically distinguish a URL glob from a legacy event name. URL globs contain wildcard or URL punctuation (``* ? / : .``); bare word-like strings are assumed to be (possibly misspelled) event names so the legacy API can reject them with a helpful error. """ return isinstance(value, str) and any(char in value for char in "*?/:.") def _decode_bytes_value(value: Any) -> Any: """Decode a BiDi BytesValue dict to a plain string where possible.""" if isinstance(value, dict) and value.get("type") == "string": return value.get("value") return value def _encode_bytes_value(value: Any) -> Any: """Encode a plain string as a BiDi BytesValue dict; pass dicts through.""" if isinstance(value, str): return {"type": "string", "value": value} if hasattr(value, "to_bidi_dict"): return value.to_bidi_dict() return value def headers_to_dict(headers: list | None) -> dict[str, str]: """Convert a BiDi header list to a name → value mapping.""" result: dict[str, str] = {} for header in headers or []: if isinstance(header, dict): result[header.get("name")] = _decode_bytes_value(header.get("value")) return result def dict_to_headers(headers: dict[str, Any] | None) -> list[dict]: """Convert a name → value mapping to a BiDi header list.""" return [{"name": name, "value": _encode_bytes_value(value)} for name, value in (headers or {}).items()] def cookies_to_list(cookies: list | None) -> list[dict]: """Convert BiDi request cookies to plain dicts with decoded values.""" result = [] for cookie in cookies or []: if isinstance(cookie, dict): decoded = dict(cookie) decoded["value"] = _decode_bytes_value(cookie.get("value")) result.append(decoded) return result def list_to_cookie_headers(cookies: list | None) -> list[dict]: """Convert plain cookie dicts to BiDi CookieHeader entries.""" result = [] for cookie in cookies or []: if hasattr(cookie, "to_bidi_dict"): result.append(cookie.to_bidi_dict()) elif isinstance(cookie, dict): result.append({"name": cookie.get("name"), "value": _encode_bytes_value(cookie.get("value"))}) return result # Optional network.SetCookieHeader fields, accepting both snake_case (Python # style) and camelCase (wire style) keys from user-supplied cookie dicts. _SET_COOKIE_FIELD_ALIASES = { "domain": "domain", "expiry": "expiry", "http_only": "httpOnly", "httpOnly": "httpOnly", "max_age": "maxAge", "maxAge": "maxAge", "path": "path", "same_site": "sameSite", "sameSite": "sameSite", "secure": "secure", } def list_to_set_cookie_headers(cookies: list | None) -> list[dict]: """Convert plain cookie dicts to BiDi SetCookieHeader entries.""" result = [] for cookie in cookies or []: if hasattr(cookie, "to_bidi_dict"): result.append(cookie.to_bidi_dict()) elif isinstance(cookie, dict): entry = {"name": cookie.get("name"), "value": _encode_bytes_value(cookie.get("value"))} for key, wire_key in _SET_COOKIE_FIELD_ALIASES.items(): if cookie.get(key) is not None: entry[wire_key] = cookie[key] result.append(entry) return result def glob_to_regex(pattern: str) -> re.Pattern: """Compile a URL glob (``*``, ``**``, ``?``) into a regular expression. ``*`` matches within a path segment, ``**`` matches across segments, and ``?`` matches a single character. Matching is anchored at both ends. """ parts: list[str] = [] i = 0 while i < len(pattern): char = pattern[i] if char == "*": if pattern[i : i + 2] == "**": parts.append(".*") i += 2 else: parts.append("[^/]*") i += 1 elif char == "?": parts.append("[^/]") i += 1 else: parts.append(re.escape(char)) i += 1 return re.compile("".join(parts) + r"\Z") def _literal_component(component: str) -> str | None: """Return the component when it is literal, ``None`` when it has wildcards. ``UrlPatternPattern`` properties match literally and browsers reject wildcard characters in them ("Forbidden characters"), while omitted properties match anything — so wildcard-bearing components are omitted from the browser-side filter and Python-side glob matching narrows the results. """ if not component or "*" in component or "?" in component: return None return component def glob_to_url_pattern(pattern: str) -> dict | None: """Translate a URL glob into a BiDi ``network.UrlPatternPattern`` dict. Only the literal components of the glob are translated; components containing wildcards are omitted (omitted UrlPatternPattern properties match anything), so the browser-side filter may be broader than the glob and callers must still apply Python-side matching. Returns ``{}`` when no browser-side filter can be derived (match everything) and ``None`` when the glob is not a URL-shaped pattern. """ if pattern in ("*", "**"): return {} if "://" not in pattern: return None scheme, _, rest = pattern.partition("://") host, slash, path = rest.partition("/") port = None if ":" in host: host, _, port = host.partition(":") result: dict[str, Any] = {"type": "pattern"} if _literal_component(scheme): result["protocol"] = scheme if _literal_component(host): result["hostname"] = host if port and _literal_component(port): result["port"] = port if slash and _literal_component("/" + path): result["pathname"] = "/" + path if len(result) == 1: return {} return result def globs_to_url_patterns(patterns: list | None) -> list[dict] | None: """Translate URL globs into BiDi UrlPatterns for ``network.addIntercept``. Returns ``None`` when no browser-side filtering should be applied (match everything, or at least one glob is untranslatable). Raw dict patterns are passed through unchanged so callers can supply wire-level UrlPatterns. """ if not patterns: return None translated = [] for pattern in patterns: if isinstance(pattern, dict): translated.append(pattern) continue url_pattern = glob_to_url_pattern(pattern) if url_pattern is None or url_pattern == {}: return None translated.append(url_pattern) return translated or None class Request: """Wraps a BiDi network request event and provides request action methods. Attributes: url: The request URL. method: The HTTP method (e.g. ``"GET"``). headers: The request headers as a name → value dict. cookies: The request cookies as a list of dicts. body: The request body. BiDi does not expose the outgoing body at the ``beforeRequestSent`` phase, so this is ``None`` unless mutated. resource_type: The resource destination (e.g. ``"script"``, ``"image"``) when reported by the browser. """ def __init__(self, conn, params, deferred: bool = False): self._conn = conn self._params = params if isinstance(params, dict) else {} req = self._params.get("request", {}) or {} self.url = req.get("url", "") self._request_id = req.get("request") self.method = req.get("method") self.headers = headers_to_dict(req.get("headers")) self.cookies = cookies_to_list(req.get("cookies")) self.body = None self.resource_type = req.get("destination") or req.get("initiatorType") # Deferred requests record actions for later reconciliation by the # registry; non-deferred (legacy) requests execute actions immediately. self._deferred = deferred self._handled = False self._failed = False self._stub: dict | None = None self._mutations: dict[str, Any] = {} def set_url(self, url: str) -> None: """Change the request URL before it is continued.""" self.url = url self._mutations["url"] = url def set_method(self, method: str) -> None: """Change the HTTP method before the request is continued.""" self.method = method self._mutations["method"] = method def set_headers(self, headers: dict[str, Any]) -> None: """Replace the request headers before the request is continued.""" self.headers = dict(headers) self._mutations["headers"] = self.headers def set_cookies(self, cookies: list) -> None: """Replace the request cookies before the request is continued.""" self.cookies = list(cookies) self._mutations["cookies"] = self.cookies def set_body(self, body: str) -> None: """Set the request body before the request is continued.""" self.body = body self._mutations["body"] = body def fail(self) -> None: """Fail the request. Takes precedence over stubbed responses and mutations when multiple handlers act on the same request. """ if self._deferred: self._failed = True else: self._execute_fail() def provide_response(self, status=None, headers=None, body=None, reason_phrase=None) -> None: """Respond to the request with a stubbed response. Args: status: HTTP status code for the stubbed response. headers: Response headers as a name → value dict. body: Response body string. reason_phrase: Optional HTTP reason phrase. """ stub = { "status": status, "headers": headers, "body": body, "reason_phrase": reason_phrase, } if self._deferred: if self._stub is None: self._stub = stub else: self._stub = stub self._execute_provide_response() def continue_request( self, *, url: str | None = None, method: str | None = None, headers: dict[str, Any] | None = None, cookies: list | None = None, body: str | None = None, ) -> None: """Continue the intercepted request, applying any recorded mutations. Each keyword argument overrides the corresponding mutation recorded via ``set_url``/``set_method``/``set_headers``/``set_cookies``/``set_body``. Arguments use the same Python types as those setters and are translated to the BiDi wire format automatically. Data URLs (``data:``) are skipped silently because browsers do not create an interceptable request entry for them, so calling ``network.continueRequest`` would raise "no such request". Args: url: Replacement request URL. method: Replacement HTTP method. headers: Replacement request headers as a name → value dict. cookies: Replacement request cookies as a list of dicts. body: Replacement request body string. """ self._handled = True if self.url.startswith("data:"): return overrides = {"url": url, "method": method, "headers": headers, "cookies": cookies, "body": body} params = self._continue_params({k: v for k, v in overrides.items() if v is not None}) self._conn.execute(command_builder("network.continueRequest", params)) def _continue_params(self, overrides: dict | None = None) -> dict: params: dict[str, Any] = {"request": self._request_id} mutations = {**self._mutations, **(overrides or {})} if "url" in mutations: params["url"] = mutations["url"] if "method" in mutations: params["method"] = mutations["method"] if "headers" in mutations: params["headers"] = dict_to_headers(mutations["headers"]) if "cookies" in mutations: params["cookies"] = list_to_cookie_headers(mutations["cookies"]) if "body" in mutations: params["body"] = _encode_bytes_value(mutations["body"]) return params def _execute_fail(self) -> None: self._handled = True if self.url.startswith("data:"): return self._conn.execute(command_builder("network.failRequest", {"request": self._request_id})) def _execute_provide_response(self) -> None: self._handled = True if self.url.startswith("data:"): return stub = self._stub or {} params: dict[str, Any] = {"request": self._request_id} if stub.get("status") is not None: params["statusCode"] = stub["status"] if stub.get("reason_phrase") is not None: params["reasonPhrase"] = stub["reason_phrase"] if stub.get("headers") is not None: params["headers"] = dict_to_headers(stub["headers"]) if stub.get("body") is not None: params["body"] = _encode_bytes_value(stub["body"]) self._conn.execute(command_builder("network.provideResponse", params)) def _resolve(self) -> None: """Reconcile recorded handler actions into a single BiDi command.""" if self._handled: return if self._failed: self._execute_fail() elif self._stub is not None: self._execute_provide_response() else: self.continue_request() class Response: """Wraps a BiDi ``network.responseStarted`` event and provides response action methods. Attributes: url: The response URL. status: The HTTP status code. reason_phrase: The HTTP status text reported by the browser. headers: The response headers as a name → value dict. mime_type: The response MIME type when reported by the browser. cookies: Cookies to set on the response. BiDi does not expose parsed response cookies at the ``responseStarted`` phase, so this is empty unless mutated via :meth:`set_cookies`. body: The response body. BiDi does not expose the body at the ``responseStarted`` phase, so this is ``None`` unless mutated via :meth:`set_body`. """ def __init__(self, conn, params, deferred: bool = False): self._conn = conn self._params = params if isinstance(params, dict) else {} req = self._params.get("request", {}) or {} resp = self._params.get("response", {}) or {} self._request_id = req.get("request") self.url = resp.get("url") or req.get("url", "") self.status = resp.get("status") self.reason_phrase = resp.get("statusText") self.headers = headers_to_dict(resp.get("headers")) self.mime_type = resp.get("mimeType") self.cookies: list = [] self.body = None # Deferred responses record actions for later reconciliation by the # registry; non-deferred responses execute actions immediately. self._deferred = deferred self._handled = False self._mutations: dict[str, Any] = {} def set_status(self, status: int, reason_phrase: str | None = None) -> None: """Change the response status code (and optionally the reason phrase).""" self.status = status self._mutations["status"] = status if reason_phrase is not None: self.reason_phrase = reason_phrase self._mutations["reason_phrase"] = reason_phrase def set_headers(self, headers: dict[str, Any]) -> None: """Replace the response headers before the response is continued.""" self.headers = dict(headers) self._mutations["headers"] = self.headers def set_cookies(self, cookies: list) -> None: """Replace the cookies set by the response before it is continued.""" self.cookies = list(cookies) self._mutations["cookies"] = self.cookies def set_body(self, body: str) -> None: """Replace the response body. The wire protocol cannot continue a response with a new body, so a body mutation is reconciled via ``network.provideResponse``, carrying over the (possibly mutated) status and headers. """ self.body = body self._mutations["body"] = body def continue_response( self, *, status: int | None = None, reason_phrase: str | None = None, headers: dict[str, Any] | None = None, cookies: list | None = None, ) -> None: """Continue the intercepted response, applying any recorded mutations. Each keyword argument overrides the corresponding mutation recorded via ``set_status``/``set_headers``/``set_cookies``. Arguments use the same Python types as those setters and are translated to the BiDi wire format automatically. Data URLs (``data:``) are skipped silently because browsers do not create an interceptable entry for them. Args: status: Replacement HTTP status code. reason_phrase: Replacement HTTP reason phrase. headers: Replacement response headers as a name → value dict. cookies: Replacement set-cookie entries as a list of dicts. """ self._handled = True if self.url.startswith("data:"): return overrides = {"status": status, "reason_phrase": reason_phrase, "headers": headers, "cookies": cookies} params = self._continue_params({k: v for k, v in overrides.items() if v is not None}) self._conn.execute(command_builder("network.continueResponse", params)) def _continue_params(self, overrides: dict | None = None) -> dict: params: dict[str, Any] = {"request": self._request_id} mutations = {**self._mutations, **(overrides or {})} if "status" in mutations: params["statusCode"] = mutations["status"] if "reason_phrase" in mutations: params["reasonPhrase"] = mutations["reason_phrase"] if "headers" in mutations: params["headers"] = dict_to_headers(mutations["headers"]) if "cookies" in mutations: params["cookies"] = list_to_set_cookie_headers(mutations["cookies"]) return params def _execute_provide_response(self) -> None: self._handled = True if self.url.startswith("data:"): return # provideResponse replaces the whole response, so carry over the # current (possibly mutated) status and headers alongside the body. params: dict[str, Any] = {"request": self._request_id} if self.status is not None: params["statusCode"] = self.status if self.reason_phrase: params["reasonPhrase"] = self.reason_phrase if self.headers: params["headers"] = dict_to_headers(self.headers) if "cookies" in self._mutations: params["cookies"] = list_to_set_cookie_headers(self._mutations["cookies"]) if self.body is not None: params["body"] = _encode_bytes_value(self.body) self._conn.execute(command_builder("network.provideResponse", params)) def _resolve(self) -> None: """Reconcile recorded handler actions into a single BiDi command.""" if self._handled: return if "body" in self._mutations: try: self._execute_provide_response() except Exception: # Some browsers cannot replace a body at the responseStarted # phase; continue with the remaining mutations rather than # leaving the response blocked and stalling the page. logger.exception("provideResponse failed; continuing response without the body mutation") self._handled = False self.continue_response() else: self.continue_response() class AuthenticationRequest: """Wraps a BiDi ``network.authRequired`` event and provides auth action methods. Attributes: url: The URL of the request that triggered the challenge. realm: The authentication realm of the first challenge, when reported. scheme: The authentication scheme (e.g. ``"basic"``) of the first challenge, when reported. challenges: Every challenge as a list of ``{"scheme", "realm"}`` dicts. """ def __init__(self, conn, params, deferred: bool = False): self._conn = conn self._params = params if isinstance(params, dict) else {} req = self._params.get("request", {}) or {} resp = self._params.get("response", {}) or {} self._request_id = req.get("request") self.url = resp.get("url") or req.get("url", "") self.challenges = [challenge for challenge in resp.get("authChallenges") or [] if isinstance(challenge, dict)] first = self.challenges[0] if self.challenges else {} self.realm = first.get("realm") self.scheme = first.get("scheme") # Deferred challenges record actions for later reconciliation by the # registry; non-deferred challenges execute actions immediately. self._deferred = deferred self._handled = False self._cancelled = False self._credentials: dict | None = None def provide_credentials(self, username: str, password: str) -> None: """Respond to the challenge with the given credentials. When multiple handlers act on the same challenge the first provided credentials win, and a ``cancel()`` from any handler takes precedence. """ credentials = {"type": "password", "username": username, "password": password} if self._deferred: if self._credentials is None: self._credentials = credentials else: self._credentials = credentials self._execute_continue("provideCredentials") def cancel(self) -> None: """Cancel the challenge, failing the request with an auth error. Takes precedence over provided credentials when multiple handlers act on the same challenge. """ if self._deferred: self._cancelled = True else: self._execute_continue("cancel") def _execute_continue(self, action: str) -> None: self._handled = True params: dict[str, Any] = {"request": self._request_id, "action": action} if action == "provideCredentials": params["credentials"] = self._credentials self._conn.execute(command_builder("network.continueWithAuth", params)) def _resolve(self) -> None: """Reconcile recorded handler actions into a single BiDi command.""" if self._handled: return if self._cancelled: self._execute_continue("cancel") elif self._credentials is not None: self._execute_continue("provideCredentials") else: self._execute_continue("default") class _HandlerEntry: """A registered handler with its patterns and intercept.""" def __init__(self, handler_id: str, patterns: list | None, callback: Callable, intercept_id: str | None): self.handler_id = handler_id self.callback = callback self.intercept_id = intercept_id self._regexes = [glob_to_regex(p) for p in patterns or [] if isinstance(p, str)] def matches(self, url: str) -> bool: if not self._regexes: return True return any(regex.match(url) for regex in self._regexes) class _BaseHandlerRegistry: """Tracks high-level handlers for one intercept phase and reconciles outcomes. One event subscription dispatches each event to all matching handlers, then reconciles the request or response exactly once. Each handler gets its own browser-side intercept so removal restores prior behavior. """ # Subclasses configure the intercept phase, the subscription event key, # the handler-ID prefix and the wrapper class handed to callbacks. _phase: str _event_name: str _id_prefix: str _label: str def __init__(self, network): self._network = network self._handlers: dict[str, _HandlerEntry] = {} self._subscription_callback_id: int | None = None self._counter = 0 def _wrap(self, params): raise NotImplementedError def add_handler(self, url_patterns, callback: Callable) -> str: """Register a handler; returns a handler ID for later removal.""" if isinstance(url_patterns, str): url_patterns = [url_patterns] patterns = list(url_patterns) if url_patterns else None bidi_patterns = globs_to_url_patterns(patterns) intercept_result = self._network._add_intercept(phases=[self._phase], url_patterns=bidi_patterns) intercept_id = intercept_result.get("intercept") if intercept_result else None if self._subscription_callback_id is None: self._subscription_callback_id = self._network.add_event_handler(self._event_name, self._on_event) self._counter += 1 handler_id = f"{self._id_prefix}-{self._counter}" self._handlers[handler_id] = _HandlerEntry(handler_id, patterns, callback, intercept_id) logger.debug("Added %s %s (patterns=%s)", self._label, handler_id, patterns) return handler_id def remove_handler(self, handler_id: str) -> None: """Remove a handler and its intercept by handler ID.""" entry = self._handlers.pop(handler_id, None) if entry is None: raise ValueError(f"{self._label.capitalize()} '{handler_id}' not found") if entry.intercept_id: self._network._remove_intercept(entry.intercept_id) if not self._keep_subscription() and self._subscription_callback_id is not None: self._network.remove_event_handler(self._event_name, self._subscription_callback_id) self._subscription_callback_id = None logger.debug("Removed %s %s", self._label, handler_id) def clear(self) -> None: """Remove all registered handlers and their intercepts.""" for handler_id in list(self._handlers): self.remove_handler(handler_id) def intercept_ids(self) -> set: """Intercept IDs owned by this registry's handlers.""" return {entry.intercept_id for entry in self._handlers.values() if entry.intercept_id} def _keep_subscription(self) -> bool: """Whether the event subscription is still needed.""" return bool(self._handlers) def resubscribe(self) -> None: """Re-establish the event subscription after an external event-handler clear.""" if self._keep_subscription(): self._subscription_callback_id = self._network.add_event_handler(self._event_name, self._on_event) else: self._subscription_callback_id = None def _before_resolve(self, wrapped) -> None: """Hook run after the handlers and before reconciliation.""" def _on_event(self, params) -> None: if not isinstance(params, dict): return wrapped = self._wrap(params) for entry in list(self._handlers.values()): if not entry.matches(wrapped.url): continue try: entry.callback(wrapped) except Exception: logger.exception("%s %s raised; continuing processing", self._label.capitalize(), entry.handler_id) if not params.get("isBlocked"): return # Only reconcile requests paused by one of our intercepts; requests # blocked by other subsystems (e.g. legacy handlers) are theirs to # continue. blocking_intercepts = set(params.get("intercepts") or []) if self.intercept_ids() & blocking_intercepts: self._before_resolve(wrapped) wrapped._resolve() class RequestHandlerRegistry(_BaseHandlerRegistry): """Dispatches ``network.beforeRequestSent`` events to request handlers. Also owns the extra-headers store: BiDi has no dedicated set-extra-headers command, so while any extra header is set every request is paused by a dedicated match-everything intercept and continued with the merged headers during reconciliation. Sharing the registry's subscription and reconciliation means a request paused by both the extra-headers intercept and user handlers is still continued exactly once. """ _phase = "beforeRequestSent" _event_name = "before_request" _id_prefix = "request-handler" _label = "request handler" def __init__(self, network): super().__init__(network) # Header names are case-insensitive per HTTP, so keys are lowercased. self.extra_headers: dict[str, Any] = {} self._extra_headers_intercept: str | None = None def _wrap(self, params): return Request(self._network._conn, params, deferred=True) def set_extra_header(self, name: str, value: str) -> None: """Record a header to merge into every subsequent request.""" self.extra_headers[name.lower()] = value if self._extra_headers_intercept is None: result = self._network._add_intercept(phases=[self._phase]) self._extra_headers_intercept = result.get("intercept") if result else None if self._subscription_callback_id is None: self._subscription_callback_id = self._network.add_event_handler(self._event_name, self._on_event) logger.debug("Added extra header %s", name.lower()) def remove_extra_header(self, name: str) -> None: """Stop merging a header by (case-insensitive) name.""" if self.extra_headers.pop(name.lower(), None) is None: raise ValueError(f"Extra header '{name}' not found") if not self.extra_headers: self._drop_extra_headers_intercept() logger.debug("Removed extra header %s", name.lower()) def clear_extra_headers(self) -> None: """Stop merging all extra headers.""" self.extra_headers.clear() self._drop_extra_headers_intercept() def _drop_extra_headers_intercept(self) -> None: if self._extra_headers_intercept: self._network._remove_intercept(self._extra_headers_intercept) self._extra_headers_intercept = None if not self._keep_subscription() and self._subscription_callback_id is not None: self._network.remove_event_handler(self._event_name, self._subscription_callback_id) self._subscription_callback_id = None def intercept_ids(self) -> set: ids = super().intercept_ids() if self._extra_headers_intercept: ids.add(self._extra_headers_intercept) return ids def _keep_subscription(self) -> bool: return bool(self._handlers or self.extra_headers) def _before_resolve(self, request) -> None: """Merge extra headers into requests about to be continued. Failed and stubbed requests never reach the wire and manually continued requests have already been sent, so only the plain-continue path is merged. """ if not self.extra_headers: return if request._handled or request._failed or request._stub is not None: return merged = {name: value for name, value in request.headers.items() if name.lower() not in self.extra_headers} merged.update(self.extra_headers) request.set_headers(merged) class ResponseHandlerRegistry(_BaseHandlerRegistry): """Dispatches ``network.responseStarted`` events to response handlers.""" _phase = "responseStarted" _event_name = "response_started" _id_prefix = "response-handler" _label = "response handler" def _wrap(self, params): return Response(self._network._conn, params, deferred=True) class AuthHandlerRegistry(_BaseHandlerRegistry): """Dispatches ``network.authRequired`` events to authentication handlers.""" _phase = "authRequired" _event_name = "auth_required" _id_prefix = "auth-handler" _label = "authentication handler" def _wrap(self, params): return AuthenticationRequest(self._network._conn, params, deferred=True)