# DO NOT EDIT THIS FILE!
|
#
|
# This file is generated from the CDP specification. If you need to make
|
# changes, edit the generator and regenerate all of the modules.
|
#
|
# CDP domain: Network
|
from __future__ import annotations
|
from .util import event_class, T_JSON_DICT
|
from dataclasses import dataclass
|
import enum
|
import typing
|
from . import debugger
|
from . import emulation
|
from . import io
|
from . import page
|
from . import runtime
|
from . import security
|
|
|
class ResourceType(enum.Enum):
|
'''
|
Resource type as it was perceived by the rendering engine.
|
'''
|
DOCUMENT = "Document"
|
STYLESHEET = "Stylesheet"
|
IMAGE = "Image"
|
MEDIA = "Media"
|
FONT = "Font"
|
SCRIPT = "Script"
|
TEXT_TRACK = "TextTrack"
|
XHR = "XHR"
|
FETCH = "Fetch"
|
EVENT_SOURCE = "EventSource"
|
WEB_SOCKET = "WebSocket"
|
MANIFEST = "Manifest"
|
SIGNED_EXCHANGE = "SignedExchange"
|
PING = "Ping"
|
CSP_VIOLATION_REPORT = "CSPViolationReport"
|
OTHER = "Other"
|
|
def to_json(self):
|
return self.value
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(json)
|
|
|
class LoaderId(str):
|
'''
|
Unique loader identifier.
|
'''
|
def to_json(self) -> str:
|
return self
|
|
@classmethod
|
def from_json(cls, json: str) -> LoaderId:
|
return cls(json)
|
|
def __repr__(self):
|
return 'LoaderId({})'.format(super().__repr__())
|
|
|
class RequestId(str):
|
'''
|
Unique request identifier.
|
'''
|
def to_json(self) -> str:
|
return self
|
|
@classmethod
|
def from_json(cls, json: str) -> RequestId:
|
return cls(json)
|
|
def __repr__(self):
|
return 'RequestId({})'.format(super().__repr__())
|
|
|
class InterceptionId(str):
|
'''
|
Unique intercepted request identifier.
|
'''
|
def to_json(self) -> str:
|
return self
|
|
@classmethod
|
def from_json(cls, json: str) -> InterceptionId:
|
return cls(json)
|
|
def __repr__(self):
|
return 'InterceptionId({})'.format(super().__repr__())
|
|
|
class ErrorReason(enum.Enum):
|
'''
|
Network level fetch failure reason.
|
'''
|
FAILED = "Failed"
|
ABORTED = "Aborted"
|
TIMED_OUT = "TimedOut"
|
ACCESS_DENIED = "AccessDenied"
|
CONNECTION_CLOSED = "ConnectionClosed"
|
CONNECTION_RESET = "ConnectionReset"
|
CONNECTION_REFUSED = "ConnectionRefused"
|
CONNECTION_ABORTED = "ConnectionAborted"
|
CONNECTION_FAILED = "ConnectionFailed"
|
NAME_NOT_RESOLVED = "NameNotResolved"
|
INTERNET_DISCONNECTED = "InternetDisconnected"
|
ADDRESS_UNREACHABLE = "AddressUnreachable"
|
BLOCKED_BY_CLIENT = "BlockedByClient"
|
BLOCKED_BY_RESPONSE = "BlockedByResponse"
|
|
def to_json(self):
|
return self.value
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(json)
|
|
|
class TimeSinceEpoch(float):
|
'''
|
UTC time in seconds, counted from January 1, 1970.
|
'''
|
def to_json(self) -> float:
|
return self
|
|
@classmethod
|
def from_json(cls, json: float) -> TimeSinceEpoch:
|
return cls(json)
|
|
def __repr__(self):
|
return 'TimeSinceEpoch({})'.format(super().__repr__())
|
|
|
class MonotonicTime(float):
|
'''
|
Monotonically increasing time in seconds since an arbitrary point in the past.
|
'''
|
def to_json(self) -> float:
|
return self
|
|
@classmethod
|
def from_json(cls, json: float) -> MonotonicTime:
|
return cls(json)
|
|
def __repr__(self):
|
return 'MonotonicTime({})'.format(super().__repr__())
|
|
|
class Headers(dict):
|
'''
|
Request / response headers as keys / values of JSON object.
|
'''
|
def to_json(self) -> dict:
|
return self
|
|
@classmethod
|
def from_json(cls, json: dict) -> Headers:
|
return cls(json)
|
|
def __repr__(self):
|
return 'Headers({})'.format(super().__repr__())
|
|
|
class ConnectionType(enum.Enum):
|
'''
|
The underlying connection technology that the browser is supposedly using.
|
'''
|
NONE = "none"
|
CELLULAR2G = "cellular2g"
|
CELLULAR3G = "cellular3g"
|
CELLULAR4G = "cellular4g"
|
BLUETOOTH = "bluetooth"
|
ETHERNET = "ethernet"
|
WIFI = "wifi"
|
WIMAX = "wimax"
|
OTHER = "other"
|
|
def to_json(self):
|
return self.value
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(json)
|
|
|
class CookieSameSite(enum.Enum):
|
'''
|
Represents the cookie's 'SameSite' status:
|
https://tools.ietf.org/html/draft-west-first-party-cookies
|
'''
|
STRICT = "Strict"
|
LAX = "Lax"
|
NONE = "None"
|
|
def to_json(self):
|
return self.value
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(json)
|
|
|
class CookiePriority(enum.Enum):
|
'''
|
Represents the cookie's 'Priority' status:
|
https://tools.ietf.org/html/draft-west-cookie-priority-00
|
'''
|
LOW = "Low"
|
MEDIUM = "Medium"
|
HIGH = "High"
|
|
def to_json(self):
|
return self.value
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(json)
|
|
|
@dataclass
|
class ResourceTiming:
|
'''
|
Timing information for the request.
|
'''
|
#: Timing's requestTime is a baseline in seconds, while the other numbers are ticks in
|
#: milliseconds relatively to this requestTime.
|
request_time: float
|
|
#: Started resolving proxy.
|
proxy_start: float
|
|
#: Finished resolving proxy.
|
proxy_end: float
|
|
#: Started DNS address resolve.
|
dns_start: float
|
|
#: Finished DNS address resolve.
|
dns_end: float
|
|
#: Started connecting to the remote host.
|
connect_start: float
|
|
#: Connected to the remote host.
|
connect_end: float
|
|
#: Started SSL handshake.
|
ssl_start: float
|
|
#: Finished SSL handshake.
|
ssl_end: float
|
|
#: Started running ServiceWorker.
|
worker_start: float
|
|
#: Finished Starting ServiceWorker.
|
worker_ready: float
|
|
#: Started fetch event.
|
worker_fetch_start: float
|
|
#: Settled fetch event respondWith promise.
|
worker_respond_with_settled: float
|
|
#: Started sending request.
|
send_start: float
|
|
#: Finished sending request.
|
send_end: float
|
|
#: Time the server started pushing request.
|
push_start: float
|
|
#: Time the server finished pushing request.
|
push_end: float
|
|
#: Finished receiving response headers.
|
receive_headers_end: float
|
|
def to_json(self):
|
json = dict()
|
json['requestTime'] = self.request_time
|
json['proxyStart'] = self.proxy_start
|
json['proxyEnd'] = self.proxy_end
|
json['dnsStart'] = self.dns_start
|
json['dnsEnd'] = self.dns_end
|
json['connectStart'] = self.connect_start
|
json['connectEnd'] = self.connect_end
|
json['sslStart'] = self.ssl_start
|
json['sslEnd'] = self.ssl_end
|
json['workerStart'] = self.worker_start
|
json['workerReady'] = self.worker_ready
|
json['workerFetchStart'] = self.worker_fetch_start
|
json['workerRespondWithSettled'] = self.worker_respond_with_settled
|
json['sendStart'] = self.send_start
|
json['sendEnd'] = self.send_end
|
json['pushStart'] = self.push_start
|
json['pushEnd'] = self.push_end
|
json['receiveHeadersEnd'] = self.receive_headers_end
|
return json
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(
|
request_time=float(json['requestTime']),
|
proxy_start=float(json['proxyStart']),
|
proxy_end=float(json['proxyEnd']),
|
dns_start=float(json['dnsStart']),
|
dns_end=float(json['dnsEnd']),
|
connect_start=float(json['connectStart']),
|
connect_end=float(json['connectEnd']),
|
ssl_start=float(json['sslStart']),
|
ssl_end=float(json['sslEnd']),
|
worker_start=float(json['workerStart']),
|
worker_ready=float(json['workerReady']),
|
worker_fetch_start=float(json['workerFetchStart']),
|
worker_respond_with_settled=float(json['workerRespondWithSettled']),
|
send_start=float(json['sendStart']),
|
send_end=float(json['sendEnd']),
|
push_start=float(json['pushStart']),
|
push_end=float(json['pushEnd']),
|
receive_headers_end=float(json['receiveHeadersEnd']),
|
)
|
|
|
class ResourcePriority(enum.Enum):
|
'''
|
Loading priority of a resource request.
|
'''
|
VERY_LOW = "VeryLow"
|
LOW = "Low"
|
MEDIUM = "Medium"
|
HIGH = "High"
|
VERY_HIGH = "VeryHigh"
|
|
def to_json(self):
|
return self.value
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(json)
|
|
|
@dataclass
|
class Request:
|
'''
|
HTTP request data.
|
'''
|
#: Request URL (without fragment).
|
url: str
|
|
#: HTTP request method.
|
method: str
|
|
#: HTTP request headers.
|
headers: Headers
|
|
#: Priority of the resource request at the time request is sent.
|
initial_priority: ResourcePriority
|
|
#: The referrer policy of the request, as defined in https://www.w3.org/TR/referrer-policy/
|
referrer_policy: str
|
|
#: Fragment of the requested URL starting with hash, if present.
|
url_fragment: typing.Optional[str] = None
|
|
#: HTTP POST request data.
|
post_data: typing.Optional[str] = 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.
|
has_post_data: typing.Optional[bool] = None
|
|
#: The mixed content type of the request.
|
mixed_content_type: typing.Optional[security.MixedContentType] = None
|
|
#: Whether is loaded via link preload.
|
is_link_preload: typing.Optional[bool] = None
|
|
def to_json(self):
|
json = dict()
|
json['url'] = self.url
|
json['method'] = self.method
|
json['headers'] = self.headers.to_json()
|
json['initialPriority'] = self.initial_priority.to_json()
|
json['referrerPolicy'] = self.referrer_policy
|
if self.url_fragment is not None:
|
json['urlFragment'] = self.url_fragment
|
if self.post_data is not None:
|
json['postData'] = self.post_data
|
if self.has_post_data is not None:
|
json['hasPostData'] = self.has_post_data
|
if self.mixed_content_type is not None:
|
json['mixedContentType'] = self.mixed_content_type.to_json()
|
if self.is_link_preload is not None:
|
json['isLinkPreload'] = self.is_link_preload
|
return json
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(
|
url=str(json['url']),
|
method=str(json['method']),
|
headers=Headers.from_json(json['headers']),
|
initial_priority=ResourcePriority.from_json(json['initialPriority']),
|
referrer_policy=str(json['referrerPolicy']),
|
url_fragment=str(json['urlFragment']) if 'urlFragment' in json else None,
|
post_data=str(json['postData']) if 'postData' in json else None,
|
has_post_data=bool(json['hasPostData']) if 'hasPostData' in json else None,
|
mixed_content_type=security.MixedContentType.from_json(json['mixedContentType']) if 'mixedContentType' in json else None,
|
is_link_preload=bool(json['isLinkPreload']) if 'isLinkPreload' in json else None,
|
)
|
|
|
@dataclass
|
class SignedCertificateTimestamp:
|
'''
|
Details of a signed certificate timestamp (SCT).
|
'''
|
#: Validation status.
|
status: str
|
|
#: Origin.
|
origin: str
|
|
#: Log name / description.
|
log_description: str
|
|
#: Log ID.
|
log_id: str
|
|
#: Issuance date.
|
timestamp: TimeSinceEpoch
|
|
#: Hash algorithm.
|
hash_algorithm: str
|
|
#: Signature algorithm.
|
signature_algorithm: str
|
|
#: Signature data.
|
signature_data: str
|
|
def to_json(self):
|
json = dict()
|
json['status'] = self.status
|
json['origin'] = self.origin
|
json['logDescription'] = self.log_description
|
json['logId'] = self.log_id
|
json['timestamp'] = self.timestamp.to_json()
|
json['hashAlgorithm'] = self.hash_algorithm
|
json['signatureAlgorithm'] = self.signature_algorithm
|
json['signatureData'] = self.signature_data
|
return json
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(
|
status=str(json['status']),
|
origin=str(json['origin']),
|
log_description=str(json['logDescription']),
|
log_id=str(json['logId']),
|
timestamp=TimeSinceEpoch.from_json(json['timestamp']),
|
hash_algorithm=str(json['hashAlgorithm']),
|
signature_algorithm=str(json['signatureAlgorithm']),
|
signature_data=str(json['signatureData']),
|
)
|
|
|
@dataclass
|
class SecurityDetails:
|
'''
|
Security details about a request.
|
'''
|
#: Protocol name (e.g. "TLS 1.2" or "QUIC").
|
protocol: str
|
|
#: Key Exchange used by the connection, or the empty string if not applicable.
|
key_exchange: str
|
|
#: Cipher name.
|
cipher: str
|
|
#: Certificate ID value.
|
certificate_id: security.CertificateId
|
|
#: Certificate subject name.
|
subject_name: str
|
|
#: Subject Alternative Name (SAN) DNS names and IP addresses.
|
san_list: typing.List[str]
|
|
#: Name of the issuing CA.
|
issuer: str
|
|
#: Certificate valid from date.
|
valid_from: TimeSinceEpoch
|
|
#: Certificate valid to (expiration) date
|
valid_to: TimeSinceEpoch
|
|
#: List of signed certificate timestamps (SCTs).
|
signed_certificate_timestamp_list: typing.List[SignedCertificateTimestamp]
|
|
#: Whether the request complied with Certificate Transparency policy
|
certificate_transparency_compliance: CertificateTransparencyCompliance
|
|
#: (EC)DH group used by the connection, if applicable.
|
key_exchange_group: typing.Optional[str] = None
|
|
#: TLS MAC. Note that AEAD ciphers do not have separate MACs.
|
mac: typing.Optional[str] = None
|
|
def to_json(self):
|
json = dict()
|
json['protocol'] = self.protocol
|
json['keyExchange'] = self.key_exchange
|
json['cipher'] = self.cipher
|
json['certificateId'] = self.certificate_id.to_json()
|
json['subjectName'] = self.subject_name
|
json['sanList'] = [i for i in self.san_list]
|
json['issuer'] = self.issuer
|
json['validFrom'] = self.valid_from.to_json()
|
json['validTo'] = self.valid_to.to_json()
|
json['signedCertificateTimestampList'] = [i.to_json() for i in self.signed_certificate_timestamp_list]
|
json['certificateTransparencyCompliance'] = self.certificate_transparency_compliance.to_json()
|
if self.key_exchange_group is not None:
|
json['keyExchangeGroup'] = self.key_exchange_group
|
if self.mac is not None:
|
json['mac'] = self.mac
|
return json
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(
|
protocol=str(json['protocol']),
|
key_exchange=str(json['keyExchange']),
|
cipher=str(json['cipher']),
|
certificate_id=security.CertificateId.from_json(json['certificateId']),
|
subject_name=str(json['subjectName']),
|
san_list=[str(i) for i in json['sanList']],
|
issuer=str(json['issuer']),
|
valid_from=TimeSinceEpoch.from_json(json['validFrom']),
|
valid_to=TimeSinceEpoch.from_json(json['validTo']),
|
signed_certificate_timestamp_list=[SignedCertificateTimestamp.from_json(i) for i in json['signedCertificateTimestampList']],
|
certificate_transparency_compliance=CertificateTransparencyCompliance.from_json(json['certificateTransparencyCompliance']),
|
key_exchange_group=str(json['keyExchangeGroup']) if 'keyExchangeGroup' in json else None,
|
mac=str(json['mac']) if 'mac' in json else None,
|
)
|
|
|
class CertificateTransparencyCompliance(enum.Enum):
|
'''
|
Whether the request complied with Certificate Transparency policy.
|
'''
|
UNKNOWN = "unknown"
|
NOT_COMPLIANT = "not-compliant"
|
COMPLIANT = "compliant"
|
|
def to_json(self):
|
return self.value
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(json)
|
|
|
class BlockedReason(enum.Enum):
|
'''
|
The reason why request was blocked.
|
'''
|
OTHER = "other"
|
CSP = "csp"
|
MIXED_CONTENT = "mixed-content"
|
ORIGIN = "origin"
|
INSPECTOR = "inspector"
|
SUBRESOURCE_FILTER = "subresource-filter"
|
CONTENT_TYPE = "content-type"
|
COLLAPSED_BY_CLIENT = "collapsed-by-client"
|
COEP_FRAME_RESOURCE_NEEDS_COEP_HEADER = "coep-frame-resource-needs-coep-header"
|
COOP_SANDBOXED_IFRAME_CANNOT_NAVIGATE_TO_COOP_PAGE = "coop-sandboxed-iframe-cannot-navigate-to-coop-page"
|
CORP_NOT_SAME_ORIGIN = "corp-not-same-origin"
|
CORP_NOT_SAME_ORIGIN_AFTER_DEFAULTED_TO_SAME_ORIGIN_BY_COEP = "corp-not-same-origin-after-defaulted-to-same-origin-by-coep"
|
CORP_NOT_SAME_SITE = "corp-not-same-site"
|
|
def to_json(self):
|
return self.value
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(json)
|
|
|
class ServiceWorkerResponseSource(enum.Enum):
|
'''
|
Source of serviceworker response.
|
'''
|
CACHE_STORAGE = "cache-storage"
|
HTTP_CACHE = "http-cache"
|
FALLBACK_CODE = "fallback-code"
|
NETWORK = "network"
|
|
def to_json(self):
|
return self.value
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(json)
|
|
|
@dataclass
|
class Response:
|
'''
|
HTTP response data.
|
'''
|
#: Response URL. This URL can be different from CachedResource.url in case of redirect.
|
url: str
|
|
#: HTTP response status code.
|
status: int
|
|
#: HTTP response status text.
|
status_text: str
|
|
#: HTTP response headers.
|
headers: Headers
|
|
#: Resource mimeType as determined by the browser.
|
mime_type: str
|
|
#: Specifies whether physical connection was actually reused for this request.
|
connection_reused: bool
|
|
#: Physical connection id that was actually used for this request.
|
connection_id: float
|
|
#: Total number of bytes received for this request so far.
|
encoded_data_length: float
|
|
#: Security state of the request resource.
|
security_state: security.SecurityState
|
|
#: HTTP response headers text.
|
headers_text: typing.Optional[str] = None
|
|
#: Refined HTTP request headers that were actually transmitted over the network.
|
request_headers: typing.Optional[Headers] = None
|
|
#: HTTP request headers text.
|
request_headers_text: typing.Optional[str] = None
|
|
#: Remote IP address.
|
remote_ip_address: typing.Optional[str] = None
|
|
#: Remote port.
|
remote_port: typing.Optional[int] = None
|
|
#: Specifies that the request was served from the disk cache.
|
from_disk_cache: typing.Optional[bool] = None
|
|
#: Specifies that the request was served from the ServiceWorker.
|
from_service_worker: typing.Optional[bool] = None
|
|
#: Specifies that the request was served from the prefetch cache.
|
from_prefetch_cache: typing.Optional[bool] = None
|
|
#: Timing information for the given request.
|
timing: typing.Optional[ResourceTiming] = None
|
|
#: Response source of response from ServiceWorker.
|
service_worker_response_source: typing.Optional[ServiceWorkerResponseSource] = None
|
|
#: The time at which the returned response was generated.
|
response_time: typing.Optional[TimeSinceEpoch] = None
|
|
#: Cache Storage Cache Name.
|
cache_storage_cache_name: typing.Optional[str] = None
|
|
#: Protocol used to fetch this request.
|
protocol: typing.Optional[str] = None
|
|
#: Security details for the request.
|
security_details: typing.Optional[SecurityDetails] = None
|
|
def to_json(self):
|
json = dict()
|
json['url'] = self.url
|
json['status'] = self.status
|
json['statusText'] = self.status_text
|
json['headers'] = self.headers.to_json()
|
json['mimeType'] = self.mime_type
|
json['connectionReused'] = self.connection_reused
|
json['connectionId'] = self.connection_id
|
json['encodedDataLength'] = self.encoded_data_length
|
json['securityState'] = self.security_state.to_json()
|
if self.headers_text is not None:
|
json['headersText'] = self.headers_text
|
if self.request_headers is not None:
|
json['requestHeaders'] = self.request_headers.to_json()
|
if self.request_headers_text is not None:
|
json['requestHeadersText'] = self.request_headers_text
|
if self.remote_ip_address is not None:
|
json['remoteIPAddress'] = self.remote_ip_address
|
if self.remote_port is not None:
|
json['remotePort'] = self.remote_port
|
if self.from_disk_cache is not None:
|
json['fromDiskCache'] = self.from_disk_cache
|
if self.from_service_worker is not None:
|
json['fromServiceWorker'] = self.from_service_worker
|
if self.from_prefetch_cache is not None:
|
json['fromPrefetchCache'] = self.from_prefetch_cache
|
if self.timing is not None:
|
json['timing'] = self.timing.to_json()
|
if self.service_worker_response_source is not None:
|
json['serviceWorkerResponseSource'] = self.service_worker_response_source.to_json()
|
if self.response_time is not None:
|
json['responseTime'] = self.response_time.to_json()
|
if self.cache_storage_cache_name is not None:
|
json['cacheStorageCacheName'] = self.cache_storage_cache_name
|
if self.protocol is not None:
|
json['protocol'] = self.protocol
|
if self.security_details is not None:
|
json['securityDetails'] = self.security_details.to_json()
|
return json
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(
|
url=str(json['url']),
|
status=int(json['status']),
|
status_text=str(json['statusText']),
|
headers=Headers.from_json(json['headers']),
|
mime_type=str(json['mimeType']),
|
connection_reused=bool(json['connectionReused']),
|
connection_id=float(json['connectionId']),
|
encoded_data_length=float(json['encodedDataLength']),
|
security_state=security.SecurityState.from_json(json['securityState']),
|
headers_text=str(json['headersText']) if 'headersText' in json else None,
|
request_headers=Headers.from_json(json['requestHeaders']) if 'requestHeaders' in json else None,
|
request_headers_text=str(json['requestHeadersText']) if 'requestHeadersText' in json else None,
|
remote_ip_address=str(json['remoteIPAddress']) if 'remoteIPAddress' in json else None,
|
remote_port=int(json['remotePort']) if 'remotePort' in json else None,
|
from_disk_cache=bool(json['fromDiskCache']) if 'fromDiskCache' in json else None,
|
from_service_worker=bool(json['fromServiceWorker']) if 'fromServiceWorker' in json else None,
|
from_prefetch_cache=bool(json['fromPrefetchCache']) if 'fromPrefetchCache' in json else None,
|
timing=ResourceTiming.from_json(json['timing']) if 'timing' in json else None,
|
service_worker_response_source=ServiceWorkerResponseSource.from_json(json['serviceWorkerResponseSource']) if 'serviceWorkerResponseSource' in json else None,
|
response_time=TimeSinceEpoch.from_json(json['responseTime']) if 'responseTime' in json else None,
|
cache_storage_cache_name=str(json['cacheStorageCacheName']) if 'cacheStorageCacheName' in json else None,
|
protocol=str(json['protocol']) if 'protocol' in json else None,
|
security_details=SecurityDetails.from_json(json['securityDetails']) if 'securityDetails' in json else None,
|
)
|
|
|
@dataclass
|
class WebSocketRequest:
|
'''
|
WebSocket request data.
|
'''
|
#: HTTP request headers.
|
headers: Headers
|
|
def to_json(self):
|
json = dict()
|
json['headers'] = self.headers.to_json()
|
return json
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(
|
headers=Headers.from_json(json['headers']),
|
)
|
|
|
@dataclass
|
class WebSocketResponse:
|
'''
|
WebSocket response data.
|
'''
|
#: HTTP response status code.
|
status: int
|
|
#: HTTP response status text.
|
status_text: str
|
|
#: HTTP response headers.
|
headers: Headers
|
|
#: HTTP response headers text.
|
headers_text: typing.Optional[str] = None
|
|
#: HTTP request headers.
|
request_headers: typing.Optional[Headers] = None
|
|
#: HTTP request headers text.
|
request_headers_text: typing.Optional[str] = None
|
|
def to_json(self):
|
json = dict()
|
json['status'] = self.status
|
json['statusText'] = self.status_text
|
json['headers'] = self.headers.to_json()
|
if self.headers_text is not None:
|
json['headersText'] = self.headers_text
|
if self.request_headers is not None:
|
json['requestHeaders'] = self.request_headers.to_json()
|
if self.request_headers_text is not None:
|
json['requestHeadersText'] = self.request_headers_text
|
return json
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(
|
status=int(json['status']),
|
status_text=str(json['statusText']),
|
headers=Headers.from_json(json['headers']),
|
headers_text=str(json['headersText']) if 'headersText' in json else None,
|
request_headers=Headers.from_json(json['requestHeaders']) if 'requestHeaders' in json else None,
|
request_headers_text=str(json['requestHeadersText']) if 'requestHeadersText' in json else None,
|
)
|
|
|
@dataclass
|
class WebSocketFrame:
|
'''
|
WebSocket message data. This represents an entire WebSocket message, not just a fragmented frame as the name suggests.
|
'''
|
#: WebSocket message opcode.
|
opcode: float
|
|
#: WebSocket message mask.
|
mask: bool
|
|
#: WebSocket message payload data.
|
#: If the opcode is 1, this is a text message and payloadData is a UTF-8 string.
|
#: If the opcode isn't 1, then payloadData is a base64 encoded string representing binary data.
|
payload_data: str
|
|
def to_json(self):
|
json = dict()
|
json['opcode'] = self.opcode
|
json['mask'] = self.mask
|
json['payloadData'] = self.payload_data
|
return json
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(
|
opcode=float(json['opcode']),
|
mask=bool(json['mask']),
|
payload_data=str(json['payloadData']),
|
)
|
|
|
@dataclass
|
class CachedResource:
|
'''
|
Information about the cached resource.
|
'''
|
#: Resource URL. This is the url of the original network request.
|
url: str
|
|
#: Type of this resource.
|
type_: ResourceType
|
|
#: Cached response body size.
|
body_size: float
|
|
#: Cached response data.
|
response: typing.Optional[Response] = None
|
|
def to_json(self):
|
json = dict()
|
json['url'] = self.url
|
json['type'] = self.type_.to_json()
|
json['bodySize'] = self.body_size
|
if self.response is not None:
|
json['response'] = self.response.to_json()
|
return json
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(
|
url=str(json['url']),
|
type_=ResourceType.from_json(json['type']),
|
body_size=float(json['bodySize']),
|
response=Response.from_json(json['response']) if 'response' in json else None,
|
)
|
|
|
@dataclass
|
class Initiator:
|
'''
|
Information about the request initiator.
|
'''
|
#: Type of this initiator.
|
type_: str
|
|
#: Initiator JavaScript stack trace, set for Script only.
|
stack: typing.Optional[runtime.StackTrace] = None
|
|
#: Initiator URL, set for Parser type or for Script type (when script is importing module) or for SignedExchange type.
|
url: typing.Optional[str] = None
|
|
#: Initiator line number, set for Parser type or for Script type (when script is importing
|
#: module) (0-based).
|
line_number: typing.Optional[float] = None
|
|
def to_json(self):
|
json = dict()
|
json['type'] = self.type_
|
if self.stack is not None:
|
json['stack'] = self.stack.to_json()
|
if self.url is not None:
|
json['url'] = self.url
|
if self.line_number is not None:
|
json['lineNumber'] = self.line_number
|
return json
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(
|
type_=str(json['type']),
|
stack=runtime.StackTrace.from_json(json['stack']) if 'stack' in json else None,
|
url=str(json['url']) if 'url' in json else None,
|
line_number=float(json['lineNumber']) if 'lineNumber' in json else None,
|
)
|
|
|
@dataclass
|
class Cookie:
|
'''
|
Cookie object
|
'''
|
#: Cookie name.
|
name: str
|
|
#: Cookie value.
|
value: str
|
|
#: Cookie domain.
|
domain: str
|
|
#: Cookie path.
|
path: str
|
|
#: Cookie expiration date as the number of seconds since the UNIX epoch.
|
expires: float
|
|
#: Cookie size.
|
size: int
|
|
#: True if cookie is http-only.
|
http_only: bool
|
|
#: True if cookie is secure.
|
secure: bool
|
|
#: True in case of session cookie.
|
session: bool
|
|
#: Cookie Priority
|
priority: CookiePriority
|
|
#: Cookie SameSite type.
|
same_site: typing.Optional[CookieSameSite] = None
|
|
def to_json(self):
|
json = dict()
|
json['name'] = self.name
|
json['value'] = self.value
|
json['domain'] = self.domain
|
json['path'] = self.path
|
json['expires'] = self.expires
|
json['size'] = self.size
|
json['httpOnly'] = self.http_only
|
json['secure'] = self.secure
|
json['session'] = self.session
|
json['priority'] = self.priority.to_json()
|
if self.same_site is not None:
|
json['sameSite'] = self.same_site.to_json()
|
return json
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(
|
name=str(json['name']),
|
value=str(json['value']),
|
domain=str(json['domain']),
|
path=str(json['path']),
|
expires=float(json['expires']),
|
size=int(json['size']),
|
http_only=bool(json['httpOnly']),
|
secure=bool(json['secure']),
|
session=bool(json['session']),
|
priority=CookiePriority.from_json(json['priority']),
|
same_site=CookieSameSite.from_json(json['sameSite']) if 'sameSite' in json else None,
|
)
|
|
|
class SetCookieBlockedReason(enum.Enum):
|
'''
|
Types of reasons why a cookie may not be stored from a response.
|
'''
|
SECURE_ONLY = "SecureOnly"
|
SAME_SITE_STRICT = "SameSiteStrict"
|
SAME_SITE_LAX = "SameSiteLax"
|
SAME_SITE_UNSPECIFIED_TREATED_AS_LAX = "SameSiteUnspecifiedTreatedAsLax"
|
SAME_SITE_NONE_INSECURE = "SameSiteNoneInsecure"
|
USER_PREFERENCES = "UserPreferences"
|
SYNTAX_ERROR = "SyntaxError"
|
SCHEME_NOT_SUPPORTED = "SchemeNotSupported"
|
OVERWRITE_SECURE = "OverwriteSecure"
|
INVALID_DOMAIN = "InvalidDomain"
|
INVALID_PREFIX = "InvalidPrefix"
|
UNKNOWN_ERROR = "UnknownError"
|
|
def to_json(self):
|
return self.value
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(json)
|
|
|
class CookieBlockedReason(enum.Enum):
|
'''
|
Types of reasons why a cookie may not be sent with a request.
|
'''
|
SECURE_ONLY = "SecureOnly"
|
NOT_ON_PATH = "NotOnPath"
|
DOMAIN_MISMATCH = "DomainMismatch"
|
SAME_SITE_STRICT = "SameSiteStrict"
|
SAME_SITE_LAX = "SameSiteLax"
|
SAME_SITE_UNSPECIFIED_TREATED_AS_LAX = "SameSiteUnspecifiedTreatedAsLax"
|
SAME_SITE_NONE_INSECURE = "SameSiteNoneInsecure"
|
USER_PREFERENCES = "UserPreferences"
|
UNKNOWN_ERROR = "UnknownError"
|
|
def to_json(self):
|
return self.value
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(json)
|
|
|
@dataclass
|
class BlockedSetCookieWithReason:
|
'''
|
A cookie which was not stored from a response with the corresponding reason.
|
'''
|
#: The reason(s) this cookie was blocked.
|
blocked_reasons: typing.List[SetCookieBlockedReason]
|
|
#: The string representing this individual cookie as it would appear in the header.
|
#: This is not the entire "cookie" or "set-cookie" header which could have multiple cookies.
|
cookie_line: str
|
|
#: The cookie object which represents the cookie which was not stored. It is optional because
|
#: sometimes complete cookie information is not available, such as in the case of parsing
|
#: errors.
|
cookie: typing.Optional[Cookie] = None
|
|
def to_json(self):
|
json = dict()
|
json['blockedReasons'] = [i.to_json() for i in self.blocked_reasons]
|
json['cookieLine'] = self.cookie_line
|
if self.cookie is not None:
|
json['cookie'] = self.cookie.to_json()
|
return json
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(
|
blocked_reasons=[SetCookieBlockedReason.from_json(i) for i in json['blockedReasons']],
|
cookie_line=str(json['cookieLine']),
|
cookie=Cookie.from_json(json['cookie']) if 'cookie' in json else None,
|
)
|
|
|
@dataclass
|
class BlockedCookieWithReason:
|
'''
|
A cookie with was not sent with a request with the corresponding reason.
|
'''
|
#: The reason(s) the cookie was blocked.
|
blocked_reasons: typing.List[CookieBlockedReason]
|
|
#: The cookie object representing the cookie which was not sent.
|
cookie: Cookie
|
|
def to_json(self):
|
json = dict()
|
json['blockedReasons'] = [i.to_json() for i in self.blocked_reasons]
|
json['cookie'] = self.cookie.to_json()
|
return json
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(
|
blocked_reasons=[CookieBlockedReason.from_json(i) for i in json['blockedReasons']],
|
cookie=Cookie.from_json(json['cookie']),
|
)
|
|
|
@dataclass
|
class CookieParam:
|
'''
|
Cookie parameter object
|
'''
|
#: Cookie name.
|
name: str
|
|
#: Cookie value.
|
value: str
|
|
#: The request-URI to associate with the setting of the cookie. This value can affect the
|
#: default domain and path values of the created cookie.
|
url: typing.Optional[str] = None
|
|
#: Cookie domain.
|
domain: typing.Optional[str] = None
|
|
#: Cookie path.
|
path: typing.Optional[str] = None
|
|
#: True if cookie is secure.
|
secure: typing.Optional[bool] = None
|
|
#: True if cookie is http-only.
|
http_only: typing.Optional[bool] = None
|
|
#: Cookie SameSite type.
|
same_site: typing.Optional[CookieSameSite] = None
|
|
#: Cookie expiration date, session cookie if not set
|
expires: typing.Optional[TimeSinceEpoch] = None
|
|
#: Cookie Priority.
|
priority: typing.Optional[CookiePriority] = None
|
|
def to_json(self):
|
json = dict()
|
json['name'] = self.name
|
json['value'] = self.value
|
if self.url is not None:
|
json['url'] = self.url
|
if self.domain is not None:
|
json['domain'] = self.domain
|
if self.path is not None:
|
json['path'] = self.path
|
if self.secure is not None:
|
json['secure'] = self.secure
|
if self.http_only is not None:
|
json['httpOnly'] = self.http_only
|
if self.same_site is not None:
|
json['sameSite'] = self.same_site.to_json()
|
if self.expires is not None:
|
json['expires'] = self.expires.to_json()
|
if self.priority is not None:
|
json['priority'] = self.priority.to_json()
|
return json
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(
|
name=str(json['name']),
|
value=str(json['value']),
|
url=str(json['url']) if 'url' in json else None,
|
domain=str(json['domain']) if 'domain' in json else None,
|
path=str(json['path']) if 'path' in json else None,
|
secure=bool(json['secure']) if 'secure' in json else None,
|
http_only=bool(json['httpOnly']) if 'httpOnly' in json else None,
|
same_site=CookieSameSite.from_json(json['sameSite']) if 'sameSite' in json else None,
|
expires=TimeSinceEpoch.from_json(json['expires']) if 'expires' in json else None,
|
priority=CookiePriority.from_json(json['priority']) if 'priority' in json else None,
|
)
|
|
|
@dataclass
|
class AuthChallenge:
|
'''
|
Authorization challenge for HTTP status code 401 or 407.
|
'''
|
#: Origin of the challenger.
|
origin: str
|
|
#: The authentication scheme used, such as basic or digest
|
scheme: str
|
|
#: The realm of the challenge. May be empty.
|
realm: str
|
|
#: Source of the authentication challenge.
|
source: typing.Optional[str] = None
|
|
def to_json(self):
|
json = dict()
|
json['origin'] = self.origin
|
json['scheme'] = self.scheme
|
json['realm'] = self.realm
|
if self.source is not None:
|
json['source'] = self.source
|
return json
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(
|
origin=str(json['origin']),
|
scheme=str(json['scheme']),
|
realm=str(json['realm']),
|
source=str(json['source']) if 'source' in json else None,
|
)
|
|
|
@dataclass
|
class AuthChallengeResponse:
|
'''
|
Response to an AuthChallenge.
|
'''
|
#: The decision on what to do in response to the authorization challenge. Default means
|
#: deferring to the default behavior of the net stack, which will likely either the Cancel
|
#: authentication or display a popup dialog box.
|
response: str
|
|
#: The username to provide, possibly empty. Should only be set if response is
|
#: ProvideCredentials.
|
username: typing.Optional[str] = None
|
|
#: The password to provide, possibly empty. Should only be set if response is
|
#: ProvideCredentials.
|
password: typing.Optional[str] = None
|
|
def to_json(self):
|
json = dict()
|
json['response'] = self.response
|
if self.username is not None:
|
json['username'] = self.username
|
if self.password is not None:
|
json['password'] = self.password
|
return json
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(
|
response=str(json['response']),
|
username=str(json['username']) if 'username' in json else None,
|
password=str(json['password']) if 'password' in json else None,
|
)
|
|
|
class InterceptionStage(enum.Enum):
|
'''
|
Stages of the interception to begin intercepting. Request will intercept before the request is
|
sent. Response will intercept after the response is received.
|
'''
|
REQUEST = "Request"
|
HEADERS_RECEIVED = "HeadersReceived"
|
|
def to_json(self):
|
return self.value
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(json)
|
|
|
@dataclass
|
class RequestPattern:
|
'''
|
Request pattern for interception.
|
'''
|
#: Wildcards ('*' -> zero or more, '?' -> exactly one) are allowed. Escape character is
|
#: backslash. Omitting is equivalent to "*".
|
url_pattern: typing.Optional[str] = None
|
|
#: If set, only requests for matching resource types will be intercepted.
|
resource_type: typing.Optional[ResourceType] = None
|
|
#: Stage at wich to begin intercepting requests. Default is Request.
|
interception_stage: typing.Optional[InterceptionStage] = None
|
|
def to_json(self):
|
json = dict()
|
if self.url_pattern is not None:
|
json['urlPattern'] = self.url_pattern
|
if self.resource_type is not None:
|
json['resourceType'] = self.resource_type.to_json()
|
if self.interception_stage is not None:
|
json['interceptionStage'] = self.interception_stage.to_json()
|
return json
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(
|
url_pattern=str(json['urlPattern']) if 'urlPattern' in json else None,
|
resource_type=ResourceType.from_json(json['resourceType']) if 'resourceType' in json else None,
|
interception_stage=InterceptionStage.from_json(json['interceptionStage']) if 'interceptionStage' in json else None,
|
)
|
|
|
@dataclass
|
class SignedExchangeSignature:
|
'''
|
Information about a signed exchange signature.
|
https://wicg.github.io/webpackage/draft-yasskin-httpbis-origin-signed-exchanges-impl.html#rfc.section.3.1
|
'''
|
#: Signed exchange signature label.
|
label: str
|
|
#: The hex string of signed exchange signature.
|
signature: str
|
|
#: Signed exchange signature integrity.
|
integrity: str
|
|
#: Signed exchange signature validity Url.
|
validity_url: str
|
|
#: Signed exchange signature date.
|
date: int
|
|
#: Signed exchange signature expires.
|
expires: int
|
|
#: Signed exchange signature cert Url.
|
cert_url: typing.Optional[str] = None
|
|
#: The hex string of signed exchange signature cert sha256.
|
cert_sha256: typing.Optional[str] = None
|
|
#: The encoded certificates.
|
certificates: typing.Optional[typing.List[str]] = None
|
|
def to_json(self):
|
json = dict()
|
json['label'] = self.label
|
json['signature'] = self.signature
|
json['integrity'] = self.integrity
|
json['validityUrl'] = self.validity_url
|
json['date'] = self.date
|
json['expires'] = self.expires
|
if self.cert_url is not None:
|
json['certUrl'] = self.cert_url
|
if self.cert_sha256 is not None:
|
json['certSha256'] = self.cert_sha256
|
if self.certificates is not None:
|
json['certificates'] = [i for i in self.certificates]
|
return json
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(
|
label=str(json['label']),
|
signature=str(json['signature']),
|
integrity=str(json['integrity']),
|
validity_url=str(json['validityUrl']),
|
date=int(json['date']),
|
expires=int(json['expires']),
|
cert_url=str(json['certUrl']) if 'certUrl' in json else None,
|
cert_sha256=str(json['certSha256']) if 'certSha256' in json else None,
|
certificates=[str(i) for i in json['certificates']] if 'certificates' in json else None,
|
)
|
|
|
@dataclass
|
class SignedExchangeHeader:
|
'''
|
Information about a signed exchange header.
|
https://wicg.github.io/webpackage/draft-yasskin-httpbis-origin-signed-exchanges-impl.html#cbor-representation
|
'''
|
#: Signed exchange request URL.
|
request_url: str
|
|
#: Signed exchange response code.
|
response_code: int
|
|
#: Signed exchange response headers.
|
response_headers: Headers
|
|
#: Signed exchange response signature.
|
signatures: typing.List[SignedExchangeSignature]
|
|
#: Signed exchange header integrity hash in the form of "sha256-<base64-hash-value>".
|
header_integrity: str
|
|
def to_json(self):
|
json = dict()
|
json['requestUrl'] = self.request_url
|
json['responseCode'] = self.response_code
|
json['responseHeaders'] = self.response_headers.to_json()
|
json['signatures'] = [i.to_json() for i in self.signatures]
|
json['headerIntegrity'] = self.header_integrity
|
return json
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(
|
request_url=str(json['requestUrl']),
|
response_code=int(json['responseCode']),
|
response_headers=Headers.from_json(json['responseHeaders']),
|
signatures=[SignedExchangeSignature.from_json(i) for i in json['signatures']],
|
header_integrity=str(json['headerIntegrity']),
|
)
|
|
|
class SignedExchangeErrorField(enum.Enum):
|
'''
|
Field type for a signed exchange related error.
|
'''
|
SIGNATURE_SIG = "signatureSig"
|
SIGNATURE_INTEGRITY = "signatureIntegrity"
|
SIGNATURE_CERT_URL = "signatureCertUrl"
|
SIGNATURE_CERT_SHA256 = "signatureCertSha256"
|
SIGNATURE_VALIDITY_URL = "signatureValidityUrl"
|
SIGNATURE_TIMESTAMPS = "signatureTimestamps"
|
|
def to_json(self):
|
return self.value
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(json)
|
|
|
@dataclass
|
class SignedExchangeError:
|
'''
|
Information about a signed exchange response.
|
'''
|
#: Error message.
|
message: str
|
|
#: The index of the signature which caused the error.
|
signature_index: typing.Optional[int] = None
|
|
#: The field which caused the error.
|
error_field: typing.Optional[SignedExchangeErrorField] = None
|
|
def to_json(self):
|
json = dict()
|
json['message'] = self.message
|
if self.signature_index is not None:
|
json['signatureIndex'] = self.signature_index
|
if self.error_field is not None:
|
json['errorField'] = self.error_field.to_json()
|
return json
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(
|
message=str(json['message']),
|
signature_index=int(json['signatureIndex']) if 'signatureIndex' in json else None,
|
error_field=SignedExchangeErrorField.from_json(json['errorField']) if 'errorField' in json else None,
|
)
|
|
|
@dataclass
|
class SignedExchangeInfo:
|
'''
|
Information about a signed exchange response.
|
'''
|
#: The outer response of signed HTTP exchange which was received from network.
|
outer_response: Response
|
|
#: Information about the signed exchange header.
|
header: typing.Optional[SignedExchangeHeader] = None
|
|
#: Security details for the signed exchange header.
|
security_details: typing.Optional[SecurityDetails] = None
|
|
#: Errors occurred while handling the signed exchagne.
|
errors: typing.Optional[typing.List[SignedExchangeError]] = None
|
|
def to_json(self):
|
json = dict()
|
json['outerResponse'] = self.outer_response.to_json()
|
if self.header is not None:
|
json['header'] = self.header.to_json()
|
if self.security_details is not None:
|
json['securityDetails'] = self.security_details.to_json()
|
if self.errors is not None:
|
json['errors'] = [i.to_json() for i in self.errors]
|
return json
|
|
@classmethod
|
def from_json(cls, json):
|
return cls(
|
outer_response=Response.from_json(json['outerResponse']),
|
header=SignedExchangeHeader.from_json(json['header']) if 'header' in json else None,
|
security_details=SecurityDetails.from_json(json['securityDetails']) if 'securityDetails' in json else None,
|
errors=[SignedExchangeError.from_json(i) for i in json['errors']] if 'errors' in json else None,
|
)
|
|
|
def can_clear_browser_cache() -> typing.Generator[T_JSON_DICT,T_JSON_DICT,bool]:
|
'''
|
Tells whether clearing browser cache is supported.
|
|
:returns: True if browser cache can be cleared.
|
'''
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.canClearBrowserCache',
|
}
|
json = yield cmd_dict
|
return bool(json['result'])
|
|
|
def can_clear_browser_cookies() -> typing.Generator[T_JSON_DICT,T_JSON_DICT,bool]:
|
'''
|
Tells whether clearing browser cookies is supported.
|
|
:returns: True if browser cookies can be cleared.
|
'''
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.canClearBrowserCookies',
|
}
|
json = yield cmd_dict
|
return bool(json['result'])
|
|
|
def can_emulate_network_conditions() -> typing.Generator[T_JSON_DICT,T_JSON_DICT,bool]:
|
'''
|
Tells whether emulation of network conditions is supported.
|
|
:returns: True if emulation of network conditions is supported.
|
'''
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.canEmulateNetworkConditions',
|
}
|
json = yield cmd_dict
|
return bool(json['result'])
|
|
|
def clear_browser_cache() -> typing.Generator[T_JSON_DICT,T_JSON_DICT,None]:
|
'''
|
Clears browser cache.
|
'''
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.clearBrowserCache',
|
}
|
json = yield cmd_dict
|
|
|
def clear_browser_cookies() -> typing.Generator[T_JSON_DICT,T_JSON_DICT,None]:
|
'''
|
Clears browser cookies.
|
'''
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.clearBrowserCookies',
|
}
|
json = yield cmd_dict
|
|
|
def continue_intercepted_request(
|
interception_id: InterceptionId,
|
error_reason: typing.Optional[ErrorReason] = None,
|
raw_response: typing.Optional[str] = None,
|
url: typing.Optional[str] = None,
|
method: typing.Optional[str] = None,
|
post_data: typing.Optional[str] = None,
|
headers: typing.Optional[Headers] = None,
|
auth_challenge_response: typing.Optional[AuthChallengeResponse] = None
|
) -> typing.Generator[T_JSON_DICT,T_JSON_DICT,None]:
|
'''
|
Response to Network.requestIntercepted which either modifies the request to continue with any
|
modifications, or blocks it, or completes it with the provided response bytes. If a network
|
fetch occurs as a result which encounters a redirect an additional Network.requestIntercepted
|
event will be sent with the same InterceptionId.
|
Deprecated, use Fetch.continueRequest, Fetch.fulfillRequest and Fetch.failRequest instead.
|
|
**EXPERIMENTAL**
|
|
:param interception_id:
|
:param error_reason: *(Optional)* If set this causes the request to fail with the given reason. Passing ```Aborted```` for requests marked with ````isNavigationRequest``` also cancels the navigation. Must not be set in response to an authChallenge.
|
:param raw_response: *(Optional)* If set the requests completes using with the provided base64 encoded raw response, including HTTP status line and headers etc... Must not be set in response to an authChallenge.
|
:param url: *(Optional)* If set the request url will be modified in a way that's not observable by page. Must not be set in response to an authChallenge.
|
:param method: *(Optional)* If set this allows the request method to be overridden. Must not be set in response to an authChallenge.
|
:param post_data: *(Optional)* If set this allows postData to be set. Must not be set in response to an authChallenge.
|
:param headers: *(Optional)* If set this allows the request headers to be changed. Must not be set in response to an authChallenge.
|
:param auth_challenge_response: *(Optional)* Response to a requestIntercepted with an authChallenge. Must not be set otherwise.
|
'''
|
params: T_JSON_DICT = dict()
|
params['interceptionId'] = interception_id.to_json()
|
if error_reason is not None:
|
params['errorReason'] = error_reason.to_json()
|
if raw_response is not None:
|
params['rawResponse'] = raw_response
|
if url is not None:
|
params['url'] = url
|
if method is not None:
|
params['method'] = method
|
if post_data is not None:
|
params['postData'] = post_data
|
if headers is not None:
|
params['headers'] = headers.to_json()
|
if auth_challenge_response is not None:
|
params['authChallengeResponse'] = auth_challenge_response.to_json()
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.continueInterceptedRequest',
|
'params': params,
|
}
|
json = yield cmd_dict
|
|
|
def delete_cookies(
|
name: str,
|
url: typing.Optional[str] = None,
|
domain: typing.Optional[str] = None,
|
path: typing.Optional[str] = None
|
) -> typing.Generator[T_JSON_DICT,T_JSON_DICT,None]:
|
'''
|
Deletes browser cookies with matching name and url or domain/path pair.
|
|
:param name: Name of the cookies to remove.
|
:param url: *(Optional)* If specified, deletes all the cookies with the given name where domain and path match provided URL.
|
:param domain: *(Optional)* If specified, deletes only cookies with the exact domain.
|
:param path: *(Optional)* If specified, deletes only cookies with the exact path.
|
'''
|
params: T_JSON_DICT = dict()
|
params['name'] = name
|
if url is not None:
|
params['url'] = url
|
if domain is not None:
|
params['domain'] = domain
|
if path is not None:
|
params['path'] = path
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.deleteCookies',
|
'params': params,
|
}
|
json = yield cmd_dict
|
|
|
def disable() -> typing.Generator[T_JSON_DICT,T_JSON_DICT,None]:
|
'''
|
Disables network tracking, prevents network events from being sent to the client.
|
'''
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.disable',
|
}
|
json = yield cmd_dict
|
|
|
def emulate_network_conditions(
|
offline: bool,
|
latency: float,
|
download_throughput: float,
|
upload_throughput: float,
|
connection_type: typing.Optional[ConnectionType] = None
|
) -> typing.Generator[T_JSON_DICT,T_JSON_DICT,None]:
|
'''
|
Activates emulation of network conditions.
|
|
:param offline: True to emulate internet disconnection.
|
:param latency: Minimum latency from request sent to response headers received (ms).
|
:param download_throughput: Maximal aggregated download throughput (bytes/sec). -1 disables download throttling.
|
:param upload_throughput: Maximal aggregated upload throughput (bytes/sec). -1 disables upload throttling.
|
:param connection_type: *(Optional)* Connection type if known.
|
'''
|
params: T_JSON_DICT = dict()
|
params['offline'] = offline
|
params['latency'] = latency
|
params['downloadThroughput'] = download_throughput
|
params['uploadThroughput'] = upload_throughput
|
if connection_type is not None:
|
params['connectionType'] = connection_type.to_json()
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.emulateNetworkConditions',
|
'params': params,
|
}
|
json = yield cmd_dict
|
|
|
def enable(
|
max_total_buffer_size: typing.Optional[int] = None,
|
max_resource_buffer_size: typing.Optional[int] = None,
|
max_post_data_size: typing.Optional[int] = None
|
) -> typing.Generator[T_JSON_DICT,T_JSON_DICT,None]:
|
'''
|
Enables network tracking, network events will now be delivered to the client.
|
|
:param max_total_buffer_size: **(EXPERIMENTAL)** *(Optional)* Buffer size in bytes to use when preserving network payloads (XHRs, etc).
|
:param max_resource_buffer_size: **(EXPERIMENTAL)** *(Optional)* Per-resource buffer size in bytes to use when preserving network payloads (XHRs, etc).
|
:param max_post_data_size: *(Optional)* Longest post body size (in bytes) that would be included in requestWillBeSent notification
|
'''
|
params: T_JSON_DICT = dict()
|
if max_total_buffer_size is not None:
|
params['maxTotalBufferSize'] = max_total_buffer_size
|
if max_resource_buffer_size is not None:
|
params['maxResourceBufferSize'] = max_resource_buffer_size
|
if max_post_data_size is not None:
|
params['maxPostDataSize'] = max_post_data_size
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.enable',
|
'params': params,
|
}
|
json = yield cmd_dict
|
|
|
def get_all_cookies() -> typing.Generator[T_JSON_DICT,T_JSON_DICT,typing.List[Cookie]]:
|
'''
|
Returns all browser cookies. Depending on the backend support, will return detailed cookie
|
information in the ``cookies`` field.
|
|
:returns: Array of cookie objects.
|
'''
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.getAllCookies',
|
}
|
json = yield cmd_dict
|
return [Cookie.from_json(i) for i in json['cookies']]
|
|
|
def get_certificate(
|
origin: str
|
) -> typing.Generator[T_JSON_DICT,T_JSON_DICT,typing.List[str]]:
|
'''
|
Returns the DER-encoded certificate.
|
|
**EXPERIMENTAL**
|
|
:param origin: Origin to get certificate for.
|
:returns:
|
'''
|
params: T_JSON_DICT = dict()
|
params['origin'] = origin
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.getCertificate',
|
'params': params,
|
}
|
json = yield cmd_dict
|
return [str(i) for i in json['tableNames']]
|
|
|
def get_cookies(
|
urls: typing.Optional[typing.List[str]] = None
|
) -> typing.Generator[T_JSON_DICT,T_JSON_DICT,typing.List[Cookie]]:
|
'''
|
Returns all browser cookies for the current URL. Depending on the backend support, will return
|
detailed cookie information in the ``cookies`` field.
|
|
:param urls: *(Optional)* The list of URLs for which applicable cookies will be fetched
|
:returns: Array of cookie objects.
|
'''
|
params: T_JSON_DICT = dict()
|
if urls is not None:
|
params['urls'] = [i for i in urls]
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.getCookies',
|
'params': params,
|
}
|
json = yield cmd_dict
|
return [Cookie.from_json(i) for i in json['cookies']]
|
|
|
def get_response_body(
|
request_id: RequestId
|
) -> typing.Generator[T_JSON_DICT,T_JSON_DICT,typing.Tuple[str, bool]]:
|
'''
|
Returns content served for the given request.
|
|
:param request_id: Identifier of the network request to get content for.
|
:returns: A tuple with the following items:
|
|
0. **body** - Response body.
|
1. **base64Encoded** - True, if content was sent as base64.
|
'''
|
params: T_JSON_DICT = dict()
|
params['requestId'] = request_id.to_json()
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.getResponseBody',
|
'params': params,
|
}
|
json = yield cmd_dict
|
return (
|
str(json['body']),
|
bool(json['base64Encoded'])
|
)
|
|
|
def get_request_post_data(
|
request_id: RequestId
|
) -> typing.Generator[T_JSON_DICT,T_JSON_DICT,str]:
|
'''
|
Returns post data sent with the request. Returns an error when no data was sent with the request.
|
|
:param request_id: Identifier of the network request to get content for.
|
:returns: Request body string, omitting files from multipart requests
|
'''
|
params: T_JSON_DICT = dict()
|
params['requestId'] = request_id.to_json()
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.getRequestPostData',
|
'params': params,
|
}
|
json = yield cmd_dict
|
return str(json['postData'])
|
|
|
def get_response_body_for_interception(
|
interception_id: InterceptionId
|
) -> typing.Generator[T_JSON_DICT,T_JSON_DICT,typing.Tuple[str, bool]]:
|
'''
|
Returns content served for the given currently intercepted request.
|
|
**EXPERIMENTAL**
|
|
:param interception_id: Identifier for the intercepted request to get body for.
|
:returns: A tuple with the following items:
|
|
0. **body** - Response body.
|
1. **base64Encoded** - True, if content was sent as base64.
|
'''
|
params: T_JSON_DICT = dict()
|
params['interceptionId'] = interception_id.to_json()
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.getResponseBodyForInterception',
|
'params': params,
|
}
|
json = yield cmd_dict
|
return (
|
str(json['body']),
|
bool(json['base64Encoded'])
|
)
|
|
|
def take_response_body_for_interception_as_stream(
|
interception_id: InterceptionId
|
) -> typing.Generator[T_JSON_DICT,T_JSON_DICT,io.StreamHandle]:
|
'''
|
Returns a handle to the stream representing the response body. Note that after this command,
|
the intercepted request can't be continued as is -- you either need to cancel it or to provide
|
the response body. The stream only supports sequential read, IO.read will fail if the position
|
is specified.
|
|
**EXPERIMENTAL**
|
|
:param interception_id:
|
:returns:
|
'''
|
params: T_JSON_DICT = dict()
|
params['interceptionId'] = interception_id.to_json()
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.takeResponseBodyForInterceptionAsStream',
|
'params': params,
|
}
|
json = yield cmd_dict
|
return io.StreamHandle.from_json(json['stream'])
|
|
|
def replay_xhr(
|
request_id: RequestId
|
) -> typing.Generator[T_JSON_DICT,T_JSON_DICT,None]:
|
'''
|
This method sends a new XMLHttpRequest which is identical to the original one. The following
|
parameters should be identical: method, url, async, request body, extra headers, withCredentials
|
attribute, user, password.
|
|
**EXPERIMENTAL**
|
|
:param request_id: Identifier of XHR to replay.
|
'''
|
params: T_JSON_DICT = dict()
|
params['requestId'] = request_id.to_json()
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.replayXHR',
|
'params': params,
|
}
|
json = yield cmd_dict
|
|
|
def search_in_response_body(
|
request_id: RequestId,
|
query: str,
|
case_sensitive: typing.Optional[bool] = None,
|
is_regex: typing.Optional[bool] = None
|
) -> typing.Generator[T_JSON_DICT,T_JSON_DICT,typing.List[debugger.SearchMatch]]:
|
'''
|
Searches for given string in response content.
|
|
**EXPERIMENTAL**
|
|
:param request_id: Identifier of the network response to search.
|
:param query: String to search for.
|
:param case_sensitive: *(Optional)* If true, search is case sensitive.
|
:param is_regex: *(Optional)* If true, treats string parameter as regex.
|
:returns: List of search matches.
|
'''
|
params: T_JSON_DICT = dict()
|
params['requestId'] = request_id.to_json()
|
params['query'] = query
|
if case_sensitive is not None:
|
params['caseSensitive'] = case_sensitive
|
if is_regex is not None:
|
params['isRegex'] = is_regex
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.searchInResponseBody',
|
'params': params,
|
}
|
json = yield cmd_dict
|
return [debugger.SearchMatch.from_json(i) for i in json['result']]
|
|
|
def set_blocked_ur_ls(
|
urls: typing.List[str]
|
) -> typing.Generator[T_JSON_DICT,T_JSON_DICT,None]:
|
'''
|
Blocks URLs from loading.
|
|
**EXPERIMENTAL**
|
|
:param urls: URL patterns to block. Wildcards ('*') are allowed.
|
'''
|
params: T_JSON_DICT = dict()
|
params['urls'] = [i for i in urls]
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.setBlockedURLs',
|
'params': params,
|
}
|
json = yield cmd_dict
|
|
|
def set_bypass_service_worker(
|
bypass: bool
|
) -> typing.Generator[T_JSON_DICT,T_JSON_DICT,None]:
|
'''
|
Toggles ignoring of service worker for each request.
|
|
**EXPERIMENTAL**
|
|
:param bypass: Bypass service worker and load from network.
|
'''
|
params: T_JSON_DICT = dict()
|
params['bypass'] = bypass
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.setBypassServiceWorker',
|
'params': params,
|
}
|
json = yield cmd_dict
|
|
|
def set_cache_disabled(
|
cache_disabled: bool
|
) -> typing.Generator[T_JSON_DICT,T_JSON_DICT,None]:
|
'''
|
Toggles ignoring cache for each request. If ``true``, cache will not be used.
|
|
:param cache_disabled: Cache disabled state.
|
'''
|
params: T_JSON_DICT = dict()
|
params['cacheDisabled'] = cache_disabled
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.setCacheDisabled',
|
'params': params,
|
}
|
json = yield cmd_dict
|
|
|
def set_cookie(
|
name: str,
|
value: str,
|
url: typing.Optional[str] = None,
|
domain: typing.Optional[str] = None,
|
path: typing.Optional[str] = None,
|
secure: typing.Optional[bool] = None,
|
http_only: typing.Optional[bool] = None,
|
same_site: typing.Optional[CookieSameSite] = None,
|
expires: typing.Optional[TimeSinceEpoch] = None,
|
priority: typing.Optional[CookiePriority] = None
|
) -> typing.Generator[T_JSON_DICT,T_JSON_DICT,bool]:
|
'''
|
Sets a cookie with the given cookie data; may overwrite equivalent cookies if they exist.
|
|
:param name: Cookie name.
|
:param value: Cookie value.
|
:param url: *(Optional)* The request-URI to associate with the setting of the cookie. This value can affect the default domain and path values of the created cookie.
|
:param domain: *(Optional)* Cookie domain.
|
:param path: *(Optional)* Cookie path.
|
:param secure: *(Optional)* True if cookie is secure.
|
:param http_only: *(Optional)* True if cookie is http-only.
|
:param same_site: *(Optional)* Cookie SameSite type.
|
:param expires: *(Optional)* Cookie expiration date, session cookie if not set
|
:param priority: **(EXPERIMENTAL)** *(Optional)* Cookie Priority type.
|
:returns: True if successfully set cookie.
|
'''
|
params: T_JSON_DICT = dict()
|
params['name'] = name
|
params['value'] = value
|
if url is not None:
|
params['url'] = url
|
if domain is not None:
|
params['domain'] = domain
|
if path is not None:
|
params['path'] = path
|
if secure is not None:
|
params['secure'] = secure
|
if http_only is not None:
|
params['httpOnly'] = http_only
|
if same_site is not None:
|
params['sameSite'] = same_site.to_json()
|
if expires is not None:
|
params['expires'] = expires.to_json()
|
if priority is not None:
|
params['priority'] = priority.to_json()
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.setCookie',
|
'params': params,
|
}
|
json = yield cmd_dict
|
return bool(json['success'])
|
|
|
def set_cookies(
|
cookies: typing.List[CookieParam]
|
) -> typing.Generator[T_JSON_DICT,T_JSON_DICT,None]:
|
'''
|
Sets given cookies.
|
|
:param cookies: Cookies to be set.
|
'''
|
params: T_JSON_DICT = dict()
|
params['cookies'] = [i.to_json() for i in cookies]
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.setCookies',
|
'params': params,
|
}
|
json = yield cmd_dict
|
|
|
def set_data_size_limits_for_test(
|
max_total_size: int,
|
max_resource_size: int
|
) -> typing.Generator[T_JSON_DICT,T_JSON_DICT,None]:
|
'''
|
For testing.
|
|
**EXPERIMENTAL**
|
|
:param max_total_size: Maximum total buffer size.
|
:param max_resource_size: Maximum per-resource size.
|
'''
|
params: T_JSON_DICT = dict()
|
params['maxTotalSize'] = max_total_size
|
params['maxResourceSize'] = max_resource_size
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.setDataSizeLimitsForTest',
|
'params': params,
|
}
|
json = yield cmd_dict
|
|
|
def set_extra_http_headers(
|
headers: Headers
|
) -> typing.Generator[T_JSON_DICT,T_JSON_DICT,None]:
|
'''
|
Specifies whether to always send extra HTTP headers with the requests from this page.
|
|
:param headers: Map with extra HTTP headers.
|
'''
|
params: T_JSON_DICT = dict()
|
params['headers'] = headers.to_json()
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.setExtraHTTPHeaders',
|
'params': params,
|
}
|
json = yield cmd_dict
|
|
|
def set_request_interception(
|
patterns: typing.List[RequestPattern]
|
) -> typing.Generator[T_JSON_DICT,T_JSON_DICT,None]:
|
'''
|
Sets the requests to intercept that match the provided patterns and optionally resource types.
|
Deprecated, please use Fetch.enable instead.
|
|
**EXPERIMENTAL**
|
|
:param patterns: Requests matching any of these patterns will be forwarded and wait for the corresponding continueInterceptedRequest call.
|
'''
|
params: T_JSON_DICT = dict()
|
params['patterns'] = [i.to_json() for i in patterns]
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.setRequestInterception',
|
'params': params,
|
}
|
json = yield cmd_dict
|
|
|
def set_user_agent_override(
|
user_agent: str,
|
accept_language: typing.Optional[str] = None,
|
platform: typing.Optional[str] = None,
|
user_agent_metadata: typing.Optional[emulation.UserAgentMetadata] = None
|
) -> typing.Generator[T_JSON_DICT,T_JSON_DICT,None]:
|
'''
|
Allows overriding user agent with the given string.
|
|
:param user_agent: User agent to use.
|
:param accept_language: *(Optional)* Browser langugage to emulate.
|
:param platform: *(Optional)* The platform navigator.platform should return.
|
:param user_agent_metadata: **(EXPERIMENTAL)** *(Optional)* To be sent in Sec-CH-UA-* headers and returned in navigator.userAgentData
|
'''
|
params: T_JSON_DICT = dict()
|
params['userAgent'] = user_agent
|
if accept_language is not None:
|
params['acceptLanguage'] = accept_language
|
if platform is not None:
|
params['platform'] = platform
|
if user_agent_metadata is not None:
|
params['userAgentMetadata'] = user_agent_metadata.to_json()
|
cmd_dict: T_JSON_DICT = {
|
'method': 'Network.setUserAgentOverride',
|
'params': params,
|
}
|
json = yield cmd_dict
|
|
|
@event_class('Network.dataReceived')
|
@dataclass
|
class DataReceived:
|
'''
|
Fired when data chunk was received over the network.
|
'''
|
#: Request identifier.
|
request_id: RequestId
|
#: Timestamp.
|
timestamp: MonotonicTime
|
#: Data chunk length.
|
data_length: int
|
#: Actual bytes received (might be less than dataLength for compressed encodings).
|
encoded_data_length: int
|
|
@classmethod
|
def from_json(cls, json: T_JSON_DICT) -> DataReceived:
|
return cls(
|
request_id=RequestId.from_json(json['requestId']),
|
timestamp=MonotonicTime.from_json(json['timestamp']),
|
data_length=int(json['dataLength']),
|
encoded_data_length=int(json['encodedDataLength'])
|
)
|
|
|
@event_class('Network.eventSourceMessageReceived')
|
@dataclass
|
class EventSourceMessageReceived:
|
'''
|
Fired when EventSource message is received.
|
'''
|
#: Request identifier.
|
request_id: RequestId
|
#: Timestamp.
|
timestamp: MonotonicTime
|
#: Message type.
|
event_name: str
|
#: Message identifier.
|
event_id: str
|
#: Message content.
|
data: str
|
|
@classmethod
|
def from_json(cls, json: T_JSON_DICT) -> EventSourceMessageReceived:
|
return cls(
|
request_id=RequestId.from_json(json['requestId']),
|
timestamp=MonotonicTime.from_json(json['timestamp']),
|
event_name=str(json['eventName']),
|
event_id=str(json['eventId']),
|
data=str(json['data'])
|
)
|
|
|
@event_class('Network.loadingFailed')
|
@dataclass
|
class LoadingFailed:
|
'''
|
Fired when HTTP request has failed to load.
|
'''
|
#: Request identifier.
|
request_id: RequestId
|
#: Timestamp.
|
timestamp: MonotonicTime
|
#: Resource type.
|
type_: ResourceType
|
#: User friendly error message.
|
error_text: str
|
#: True if loading was canceled.
|
canceled: typing.Optional[bool]
|
#: The reason why loading was blocked, if any.
|
blocked_reason: typing.Optional[BlockedReason]
|
|
@classmethod
|
def from_json(cls, json: T_JSON_DICT) -> LoadingFailed:
|
return cls(
|
request_id=RequestId.from_json(json['requestId']),
|
timestamp=MonotonicTime.from_json(json['timestamp']),
|
type_=ResourceType.from_json(json['type']),
|
error_text=str(json['errorText']),
|
canceled=bool(json['canceled']) if 'canceled' in json else None,
|
blocked_reason=BlockedReason.from_json(json['blockedReason']) if 'blockedReason' in json else None
|
)
|
|
|
@event_class('Network.loadingFinished')
|
@dataclass
|
class LoadingFinished:
|
'''
|
Fired when HTTP request has finished loading.
|
'''
|
#: Request identifier.
|
request_id: RequestId
|
#: Timestamp.
|
timestamp: MonotonicTime
|
#: Total number of bytes received for this request.
|
encoded_data_length: float
|
#: Set when 1) response was blocked by Cross-Origin Read Blocking and also
|
#: 2) this needs to be reported to the DevTools console.
|
should_report_corb_blocking: typing.Optional[bool]
|
|
@classmethod
|
def from_json(cls, json: T_JSON_DICT) -> LoadingFinished:
|
return cls(
|
request_id=RequestId.from_json(json['requestId']),
|
timestamp=MonotonicTime.from_json(json['timestamp']),
|
encoded_data_length=float(json['encodedDataLength']),
|
should_report_corb_blocking=bool(json['shouldReportCorbBlocking']) if 'shouldReportCorbBlocking' in json else None
|
)
|
|
|
@event_class('Network.requestIntercepted')
|
@dataclass
|
class RequestIntercepted:
|
'''
|
**EXPERIMENTAL**
|
|
Details of an intercepted HTTP request, which must be either allowed, blocked, modified or
|
mocked.
|
Deprecated, use Fetch.requestPaused instead.
|
'''
|
#: Each request the page makes will have a unique id, however if any redirects are encountered
|
#: while processing that fetch, they will be reported with the same id as the original fetch.
|
#: Likewise if HTTP authentication is needed then the same fetch id will be used.
|
interception_id: InterceptionId
|
request: Request
|
#: The id of the frame that initiated the request.
|
frame_id: page.FrameId
|
#: How the requested resource will be used.
|
resource_type: ResourceType
|
#: Whether this is a navigation request, which can abort the navigation completely.
|
is_navigation_request: bool
|
#: Set if the request is a navigation that will result in a download.
|
#: Only present after response is received from the server (i.e. HeadersReceived stage).
|
is_download: typing.Optional[bool]
|
#: Redirect location, only sent if a redirect was intercepted.
|
redirect_url: typing.Optional[str]
|
#: Details of the Authorization Challenge encountered. If this is set then
|
#: continueInterceptedRequest must contain an authChallengeResponse.
|
auth_challenge: typing.Optional[AuthChallenge]
|
#: Response error if intercepted at response stage or if redirect occurred while intercepting
|
#: request.
|
response_error_reason: typing.Optional[ErrorReason]
|
#: Response code if intercepted at response stage or if redirect occurred while intercepting
|
#: request or auth retry occurred.
|
response_status_code: typing.Optional[int]
|
#: Response headers if intercepted at the response stage or if redirect occurred while
|
#: intercepting request or auth retry occurred.
|
response_headers: typing.Optional[Headers]
|
#: If the intercepted request had a corresponding requestWillBeSent event fired for it, then
|
#: this requestId will be the same as the requestId present in the requestWillBeSent event.
|
request_id: typing.Optional[RequestId]
|
|
@classmethod
|
def from_json(cls, json: T_JSON_DICT) -> RequestIntercepted:
|
return cls(
|
interception_id=InterceptionId.from_json(json['interceptionId']),
|
request=Request.from_json(json['request']),
|
frame_id=page.FrameId.from_json(json['frameId']),
|
resource_type=ResourceType.from_json(json['resourceType']),
|
is_navigation_request=bool(json['isNavigationRequest']),
|
is_download=bool(json['isDownload']) if 'isDownload' in json else None,
|
redirect_url=str(json['redirectUrl']) if 'redirectUrl' in json else None,
|
auth_challenge=AuthChallenge.from_json(json['authChallenge']) if 'authChallenge' in json else None,
|
response_error_reason=ErrorReason.from_json(json['responseErrorReason']) if 'responseErrorReason' in json else None,
|
response_status_code=int(json['responseStatusCode']) if 'responseStatusCode' in json else None,
|
response_headers=Headers.from_json(json['responseHeaders']) if 'responseHeaders' in json else None,
|
request_id=RequestId.from_json(json['requestId']) if 'requestId' in json else None
|
)
|
|
|
@event_class('Network.requestServedFromCache')
|
@dataclass
|
class RequestServedFromCache:
|
'''
|
Fired if request ended up loading from cache.
|
'''
|
#: Request identifier.
|
request_id: RequestId
|
|
@classmethod
|
def from_json(cls, json: T_JSON_DICT) -> RequestServedFromCache:
|
return cls(
|
request_id=RequestId.from_json(json['requestId'])
|
)
|
|
|
@event_class('Network.requestWillBeSent')
|
@dataclass
|
class RequestWillBeSent:
|
'''
|
Fired when page is about to send HTTP request.
|
'''
|
#: Request identifier.
|
request_id: RequestId
|
#: Loader identifier. Empty string if the request is fetched from worker.
|
loader_id: LoaderId
|
#: URL of the document this request is loaded for.
|
document_url: str
|
#: Request data.
|
request: Request
|
#: Timestamp.
|
timestamp: MonotonicTime
|
#: Timestamp.
|
wall_time: TimeSinceEpoch
|
#: Request initiator.
|
initiator: Initiator
|
#: Redirect response data.
|
redirect_response: typing.Optional[Response]
|
#: Type of this resource.
|
type_: typing.Optional[ResourceType]
|
#: Frame identifier.
|
frame_id: typing.Optional[page.FrameId]
|
#: Whether the request is initiated by a user gesture. Defaults to false.
|
has_user_gesture: typing.Optional[bool]
|
|
@classmethod
|
def from_json(cls, json: T_JSON_DICT) -> RequestWillBeSent:
|
return cls(
|
request_id=RequestId.from_json(json['requestId']),
|
loader_id=LoaderId.from_json(json['loaderId']),
|
document_url=str(json['documentURL']),
|
request=Request.from_json(json['request']),
|
timestamp=MonotonicTime.from_json(json['timestamp']),
|
wall_time=TimeSinceEpoch.from_json(json['wallTime']),
|
initiator=Initiator.from_json(json['initiator']),
|
redirect_response=Response.from_json(json['redirectResponse']) if 'redirectResponse' in json else None,
|
type_=ResourceType.from_json(json['type']) if 'type' in json else None,
|
frame_id=page.FrameId.from_json(json['frameId']) if 'frameId' in json else None,
|
has_user_gesture=bool(json['hasUserGesture']) if 'hasUserGesture' in json else None
|
)
|
|
|
@event_class('Network.resourceChangedPriority')
|
@dataclass
|
class ResourceChangedPriority:
|
'''
|
**EXPERIMENTAL**
|
|
Fired when resource loading priority is changed
|
'''
|
#: Request identifier.
|
request_id: RequestId
|
#: New priority
|
new_priority: ResourcePriority
|
#: Timestamp.
|
timestamp: MonotonicTime
|
|
@classmethod
|
def from_json(cls, json: T_JSON_DICT) -> ResourceChangedPriority:
|
return cls(
|
request_id=RequestId.from_json(json['requestId']),
|
new_priority=ResourcePriority.from_json(json['newPriority']),
|
timestamp=MonotonicTime.from_json(json['timestamp'])
|
)
|
|
|
@event_class('Network.signedExchangeReceived')
|
@dataclass
|
class SignedExchangeReceived:
|
'''
|
**EXPERIMENTAL**
|
|
Fired when a signed exchange was received over the network
|
'''
|
#: Request identifier.
|
request_id: RequestId
|
#: Information about the signed exchange response.
|
info: SignedExchangeInfo
|
|
@classmethod
|
def from_json(cls, json: T_JSON_DICT) -> SignedExchangeReceived:
|
return cls(
|
request_id=RequestId.from_json(json['requestId']),
|
info=SignedExchangeInfo.from_json(json['info'])
|
)
|
|
|
@event_class('Network.responseReceived')
|
@dataclass
|
class ResponseReceived:
|
'''
|
Fired when HTTP response is available.
|
'''
|
#: Request identifier.
|
request_id: RequestId
|
#: Loader identifier. Empty string if the request is fetched from worker.
|
loader_id: LoaderId
|
#: Timestamp.
|
timestamp: MonotonicTime
|
#: Resource type.
|
type_: ResourceType
|
#: Response data.
|
response: Response
|
#: Frame identifier.
|
frame_id: typing.Optional[page.FrameId]
|
|
@classmethod
|
def from_json(cls, json: T_JSON_DICT) -> ResponseReceived:
|
return cls(
|
request_id=RequestId.from_json(json['requestId']),
|
loader_id=LoaderId.from_json(json['loaderId']),
|
timestamp=MonotonicTime.from_json(json['timestamp']),
|
type_=ResourceType.from_json(json['type']),
|
response=Response.from_json(json['response']),
|
frame_id=page.FrameId.from_json(json['frameId']) if 'frameId' in json else None
|
)
|
|
|
@event_class('Network.webSocketClosed')
|
@dataclass
|
class WebSocketClosed:
|
'''
|
Fired when WebSocket is closed.
|
'''
|
#: Request identifier.
|
request_id: RequestId
|
#: Timestamp.
|
timestamp: MonotonicTime
|
|
@classmethod
|
def from_json(cls, json: T_JSON_DICT) -> WebSocketClosed:
|
return cls(
|
request_id=RequestId.from_json(json['requestId']),
|
timestamp=MonotonicTime.from_json(json['timestamp'])
|
)
|
|
|
@event_class('Network.webSocketCreated')
|
@dataclass
|
class WebSocketCreated:
|
'''
|
Fired upon WebSocket creation.
|
'''
|
#: Request identifier.
|
request_id: RequestId
|
#: WebSocket request URL.
|
url: str
|
#: Request initiator.
|
initiator: typing.Optional[Initiator]
|
|
@classmethod
|
def from_json(cls, json: T_JSON_DICT) -> WebSocketCreated:
|
return cls(
|
request_id=RequestId.from_json(json['requestId']),
|
url=str(json['url']),
|
initiator=Initiator.from_json(json['initiator']) if 'initiator' in json else None
|
)
|
|
|
@event_class('Network.webSocketFrameError')
|
@dataclass
|
class WebSocketFrameError:
|
'''
|
Fired when WebSocket message error occurs.
|
'''
|
#: Request identifier.
|
request_id: RequestId
|
#: Timestamp.
|
timestamp: MonotonicTime
|
#: WebSocket error message.
|
error_message: str
|
|
@classmethod
|
def from_json(cls, json: T_JSON_DICT) -> WebSocketFrameError:
|
return cls(
|
request_id=RequestId.from_json(json['requestId']),
|
timestamp=MonotonicTime.from_json(json['timestamp']),
|
error_message=str(json['errorMessage'])
|
)
|
|
|
@event_class('Network.webSocketFrameReceived')
|
@dataclass
|
class WebSocketFrameReceived:
|
'''
|
Fired when WebSocket message is received.
|
'''
|
#: Request identifier.
|
request_id: RequestId
|
#: Timestamp.
|
timestamp: MonotonicTime
|
#: WebSocket response data.
|
response: WebSocketFrame
|
|
@classmethod
|
def from_json(cls, json: T_JSON_DICT) -> WebSocketFrameReceived:
|
return cls(
|
request_id=RequestId.from_json(json['requestId']),
|
timestamp=MonotonicTime.from_json(json['timestamp']),
|
response=WebSocketFrame.from_json(json['response'])
|
)
|
|
|
@event_class('Network.webSocketFrameSent')
|
@dataclass
|
class WebSocketFrameSent:
|
'''
|
Fired when WebSocket message is sent.
|
'''
|
#: Request identifier.
|
request_id: RequestId
|
#: Timestamp.
|
timestamp: MonotonicTime
|
#: WebSocket response data.
|
response: WebSocketFrame
|
|
@classmethod
|
def from_json(cls, json: T_JSON_DICT) -> WebSocketFrameSent:
|
return cls(
|
request_id=RequestId.from_json(json['requestId']),
|
timestamp=MonotonicTime.from_json(json['timestamp']),
|
response=WebSocketFrame.from_json(json['response'])
|
)
|
|
|
@event_class('Network.webSocketHandshakeResponseReceived')
|
@dataclass
|
class WebSocketHandshakeResponseReceived:
|
'''
|
Fired when WebSocket handshake response becomes available.
|
'''
|
#: Request identifier.
|
request_id: RequestId
|
#: Timestamp.
|
timestamp: MonotonicTime
|
#: WebSocket response data.
|
response: WebSocketResponse
|
|
@classmethod
|
def from_json(cls, json: T_JSON_DICT) -> WebSocketHandshakeResponseReceived:
|
return cls(
|
request_id=RequestId.from_json(json['requestId']),
|
timestamp=MonotonicTime.from_json(json['timestamp']),
|
response=WebSocketResponse.from_json(json['response'])
|
)
|
|
|
@event_class('Network.webSocketWillSendHandshakeRequest')
|
@dataclass
|
class WebSocketWillSendHandshakeRequest:
|
'''
|
Fired when WebSocket is about to initiate handshake.
|
'''
|
#: Request identifier.
|
request_id: RequestId
|
#: Timestamp.
|
timestamp: MonotonicTime
|
#: UTC Timestamp.
|
wall_time: TimeSinceEpoch
|
#: WebSocket request data.
|
request: WebSocketRequest
|
|
@classmethod
|
def from_json(cls, json: T_JSON_DICT) -> WebSocketWillSendHandshakeRequest:
|
return cls(
|
request_id=RequestId.from_json(json['requestId']),
|
timestamp=MonotonicTime.from_json(json['timestamp']),
|
wall_time=TimeSinceEpoch.from_json(json['wallTime']),
|
request=WebSocketRequest.from_json(json['request'])
|
)
|
|
|
@event_class('Network.requestWillBeSentExtraInfo')
|
@dataclass
|
class RequestWillBeSentExtraInfo:
|
'''
|
**EXPERIMENTAL**
|
|
Fired when additional information about a requestWillBeSent event is available from the
|
network stack. Not every requestWillBeSent event will have an additional
|
requestWillBeSentExtraInfo fired for it, and there is no guarantee whether requestWillBeSent
|
or requestWillBeSentExtraInfo will be fired first for the same request.
|
'''
|
#: Request identifier. Used to match this information to an existing requestWillBeSent event.
|
request_id: RequestId
|
#: A list of cookies potentially associated to the requested URL. This includes both cookies sent with
|
#: the request and the ones not sent; the latter are distinguished by having blockedReason field set.
|
associated_cookies: typing.List[BlockedCookieWithReason]
|
#: Raw request headers as they will be sent over the wire.
|
headers: Headers
|
|
@classmethod
|
def from_json(cls, json: T_JSON_DICT) -> RequestWillBeSentExtraInfo:
|
return cls(
|
request_id=RequestId.from_json(json['requestId']),
|
associated_cookies=[BlockedCookieWithReason.from_json(i) for i in json['associatedCookies']],
|
headers=Headers.from_json(json['headers'])
|
)
|
|
|
@event_class('Network.responseReceivedExtraInfo')
|
@dataclass
|
class ResponseReceivedExtraInfo:
|
'''
|
**EXPERIMENTAL**
|
|
Fired when additional information about a responseReceived event is available from the network
|
stack. Not every responseReceived event will have an additional responseReceivedExtraInfo for
|
it, and responseReceivedExtraInfo may be fired before or after responseReceived.
|
'''
|
#: Request identifier. Used to match this information to another responseReceived event.
|
request_id: RequestId
|
#: A list of cookies which were not stored from the response along with the corresponding
|
#: reasons for blocking. The cookies here may not be valid due to syntax errors, which
|
#: are represented by the invalid cookie line string instead of a proper cookie.
|
blocked_cookies: typing.List[BlockedSetCookieWithReason]
|
#: Raw response headers as they were received over the wire.
|
headers: Headers
|
#: Raw response header text as it was received over the wire. The raw text may not always be
|
#: available, such as in the case of HTTP/2 or QUIC.
|
headers_text: typing.Optional[str]
|
|
@classmethod
|
def from_json(cls, json: T_JSON_DICT) -> ResponseReceivedExtraInfo:
|
return cls(
|
request_id=RequestId.from_json(json['requestId']),
|
blocked_cookies=[BlockedSetCookieWithReason.from_json(i) for i in json['blockedCookies']],
|
headers=Headers.from_json(json['headers']),
|
headers_text=str(json['headersText']) if 'headersText' in json else None
|
)
|
