Request-Interception

Example Script

import asyncio
import json
from selenium_driverless import webdriver
from selenium_driverless.scripts.network_interceptor import NetworkInterceptor, InterceptedRequest, InterceptedAuth, \
    RequestPattern, RequestStages


async def on_request(data: InterceptedRequest):
    if data.request.url == "https://httpbin.org/post":
        await data.continue_request(url="https://httpbin.org/basic-auth/user/pass", intercept_response=True)


async def main():
    async with webdriver.Chrome(max_ws_size=2 ** 30) as driver:

        async with NetworkInterceptor(driver, on_request=on_request, patterns=[RequestPattern.AnyRequest],
                                      intercept_auth=True) as interceptor:

            asyncio.ensure_future(driver.get("https://httpbin.org/post", wait_load=False))
            async for data in interceptor:
                if data.request.url == "https://httpbin.org/basic-auth/user/pass":
                    if isinstance(data, InterceptedAuth):
                        # iteration should take virtually zero time, as that would block other requests
                        asyncio.ensure_future(data.continue_auth(username="user", password="pass"))
                    elif data.stage == RequestStages.Response:
                        print(json.loads(await data.body))
                        break


asyncio.run(main())

API

class selenium_driverless.scripts.network_interceptor.NetworkInterceptor(target, on_request=None, on_response=None, on_auth=None, patterns=None, intercept_auth=False, bypass_service_workers=False)[source]

__init__(target, on_request=None, on_response=None, on_auth=None, patterns=None, intercept_auth=False, bypass_service_workers=False)[source]

Parameters:

target (Union[Chrome, Target]) – the Target or Driver, on which requests get intercepted
on_request (Optional[Callable[[InterceptedRequest], Awaitable[None]]]) – onRequest callback
on_response (Optional[Callable[[InterceptedRequest], Awaitable[None]]]) – onResponse callback
on_auth (Optional[Callable[[InterceptedRequest], Awaitable[None]]]) – onAuth callback
patterns (Union[List[Dict[str, str]], List[RequestPattern], None]) – the request patterns to intercept
intercept_auth (bool) – whether to intercept authentification
bypass_service_workers (bool) – whether to bypass service workers for a single Target

__aiter__()[source]

iterate using async for over requests

Example

from selenium_driverless.scripts.network_interceptor import InterceptedRequest, RequestStages

# ... some code
async for data in interceptor:
    url = data.request.url
    if isinstance(data, InterceptedRequest):
        data.stage == RequestStages.Request:
            print("Request:"+url)
        else:
            print("Response:"+url)
    else:
        print("Authentification:"+url)

Return type:: AsyncIterator[Union[InterceptedRequest, InterceptedAuth]]

Warning

iterations should virtually take zero time, you might use asyncio.ensure_future where possible

property patterns: List[Dict[str, str]]: patters to intercept

property target: Target | BaseTarget: the Target, on which requests get intercepted

class selenium_driverless.scripts.network_interceptor.InterceptedRequest(params, target)[source]

property target: Target | BaseTarget: the target

property stage: Literal[0, 1]: one of RequestStages.Request, RequestStages.Response 0 => Request 1 => Response

property is_redirect: if the response is a redirect (response_status_code in [301, 302, 303, 307, 308])

property params: dict: the parameters from Fetch.requestPaused as a dict

property frame_id: str: the Page.FrameId of the frame that initiated the request

property request: Request: the corresponding request

property id: str: Fetch.RequestId

property resource_type: str: Network.resourceType

property network_id: str | None: the Network.RequestId if a corresponding Network.requestWillBeSent event had been fired

property response_error_reason: str | None: response error-reason at response stage

property response_headers: List[Dict[str, str]] | None: the headers at response stage

property response_status_code: int | None: the status code at response stage

property response_status_text: str | None: the status text at response stage

property redirected_id: str | None: Fetch.requestId reference id to the request that caused the redirect

property body: bytes | None: async get the response body for a request

Note

this uses network and might take some time

async bypass_browser(auth=None, allow_redirects=True, compress=None, proxy=None, proxy_auth=None, timeout=None)[source]: bypass browser by making the request externally

Warning

this method does not change the TLS fingerprint accordingly and is technically detectable

async continue_request(headers=None, method=None, post_data=None, url=None, intercept_response=None)[source]

continue the request. You might use InterceptedRequest.fulfill to provide a response body

Parameters:

headers (Optional[List[Dict[str, str]]]) – array of {“name”:name, “value”:value}, mind header order
method (Optional[str]) – the method of the request
post_data (Union[str, bytes, None]) – the binary post data to provide
url (Optional[str]) – the url to continue the request with, change not observable by the page
intercept_response (Optional[bool]) – overrides response interception behavior for this request

async continue_response(response_headers=None, binary_response_headers=None, response_code=None, response_phrase=None)[source]

continue the response. You might use By to provide a response body

Parameters:

response_code (Optional[int]) – response code
binary_response_headers (Optional[bytes]) – headers as a -separated series of name: value pairs, treated as base64 encode if a string is passed,
response_headers (Optional[List[Dict[str, str]]]) – array of {“name”:name, “value”:value}, mind header order
response_phrase (Optional[str]) – response phrase ("OK" for response_code=200)

async resume()[source]: continue the request or response doesn’t raise id the request is canceled or already done

async fail_request(error_reason)[source]

fail the request or response

Parameters:: error_reason (Literal['Failed', 'Aborted', 'TimedOut', 'AccessDenied', 'ConnectionClosed', 'ConnectionReset', 'ConnectionRefused', 'ConnectionAborted', 'ConnectionFailed', 'NameNotResolved', 'InternetDisconnected', 'AddressUnreachable', 'BlockedByClient', 'BlockedByResponse']) – the specified reason for the request failing

async fulfill(response_code, binary_response_headers=None, body=None, response_headers=None, response_phrase=None)[source]

fulfill the request or response

Parameters:

response_code (int) – response code
body (Union[str, bytes, None]) – the response body
binary_response_headers (Optional[str]) – headers as a -separated series of name: value pairs, treated as base64 encode if a string is passed,
response_headers (Optional[List[Dict[str, str]]]) – array of {“name”:name, “value”:value}, mind header order
response_phrase (Optional[str]) – response phrase ("OK" for response_code=200)

class selenium_driverless.scripts.network_interceptor.InterceptedAuth(params, target)[source]

property request: Request: the Request, for which auth is required

property id: str: the Fetch.RequestId

property frame_id: str: the Page.FrameId from which the request has been initiated

property params: dict: the parameters from Fetch.authRequired as a dict

property target: Target | BaseTarget: the target

property resource_type: str: the Network.ResourceType

property auth_challenge: AuthChallenge: the AuthChallenge

async continue_auth(response='Default', username=None, password=None)[source]

continue the auth

Parameters:

response (Literal['Default', 'CancelAuth', 'ProvideCredentials']) – “Default” by default, automatically set to “ProvideCredentials”, if username or password specified
username (Optional[str]) – username for the auth
password (Optional[str]) – password for the auth

async resume()[source]: continue the auth doesn’t raise id the request is canceled or already done

async cancel()[source]: cancel the auth

class selenium_driverless.scripts.network_interceptor.AuthChallenge(params, target)[source]

property target: Target | BaseTarget: the target

property params: dict: the parameters from Fetch.AuthChallenge as a dict

property source: Literal['Server', 'Proxy']: Source of the authentication challenge

property origin: str: Origin of the challenger.

property scheme: str: The authentication scheme used, such as basic or digest

property realm: str: corresponding value for Please enter your username and password for <realm name>:

class selenium_driverless.scripts.network_interceptor.Request(params, target)[source]

property target: Target | BaseTarget: the target

property params: dict: the parameters from Network.Request as a dict

property url: str: Request URL (without fragment).

property url_fragment: str | None: Fragment of the requested URL starting with hash, if present.

property method: str | None: HTTP request method.

property headers: Dict[str, str]: Request / response headers as keys / values of JSON object.

property post_data: str | None: HTTP POST request data

property has_post_data: bool | None: True when the request has POST data. Note that postData might still be omitted when this flag is true when the data is too long.

property post_data_entries: List[str] | None: an array of Network.PostDataEntry

property mixed_content_type: Literal['blockable', 'optionally-blockable', 'none', None]: The mixed content type of the request.

property initial_priority: Literal['VeryLow', 'Low', 'Medium', 'High', 'VeryHigh', None]: Priority of the resource request at the time request is sent.

property referrer_policy: Literal['unsafe-url', 'no-referrer-when-downgrade', 'no-referrer', 'origin', 'origin-when-cross-origin', 'same-origin', 'strict-origin', 'strict-origin-when-cross-origin']: The referrer policy of the request, as defined in w3.org/TR/referrer-policy

property is_link_preload: bool | None: Whether is loaded via link preload.

property trust_token_params: dict | None: see Network.TrustTokenParams for reference

property is_same_site: bool | None: True if this resource request is considered to be the ‘same site’ as the request corresponding to the main frame.

class selenium_driverless.scripts.network_interceptor.RequestStages[source]

Request = 0: stage at request

Response = 1: stage at response

class selenium_driverless.scripts.network_interceptor.RequestPattern(value)[source]

An enumeration.

AnyRequest = {'requestStage': 'Request', 'urlPattern': '*'}: pattern for any request

AnyResponse = {'requestStage': 'Response', 'urlPattern': '*'}: pattern for any response

static new(url_pattern=None, resource_type=None, request_stage=None)[source]: create a new request pattern

class selenium_driverless.scripts.network_interceptor.AuthAlreadyHandledException(data)[source]

raised when an auth challenge has already been handled by an external application (for example chrome-extension)

property request: InterceptedAuth: the corresponding auth challenge

class selenium_driverless.scripts.network_interceptor.RequestDoneException(data)[source]

raised when the request or auth challenge has already been resumed

property request: InterceptedRequest | InterceptedAuth: the corresponding request or auth challenge