forked from llimllib/chrome-control
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Network.py
461 lines (363 loc) · 17.2 KB
/
Network.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
from enum import Enum
from typing import Any, List
from base import ChromeCommand
import Runtime
import Page
import Security
# Unique loader identifier.
LoaderId = str
# Unique request identifier.
RequestId = str
# Number of seconds since epoch.
Timestamp = float
class Headers: pass
ConnectionType = Enum("ConnectionType", "none cellular2g cellular3g cellular4g bluetooth ethernet wifi wimax other")
ConnectionType.__doc__ = "Loading priority of a resource request."
CookieSameSite = Enum("CookieSameSite", "Strict Lax")
CookieSameSite.__doc__ = "Represents the cookie's 'SameSite' status: https://tools.ietf.org/html/draft-west-first-party-cookies"
class ResourceTiming:
"""Timing information for the request."""
def __init__(self, requestTime: float, proxyStart: float, proxyEnd: float, dnsStart: float, dnsEnd: float, connectStart: float, connectEnd: float, sslStart: float, sslEnd: float, workerStart: float, workerReady: float, sendStart: float, sendEnd: float, pushStart: float, pushEnd: float, receiveHeadersEnd: float):
# Timing's requestTime is a baseline in seconds, while the other numbers are ticks in milliseconds relatively to this requestTime.
self.requestTime = requestTime
# Started resolving proxy.
self.proxyStart = proxyStart
# Finished resolving proxy.
self.proxyEnd = proxyEnd
# Started DNS address resolve.
self.dnsStart = dnsStart
# Finished DNS address resolve.
self.dnsEnd = dnsEnd
# Started connecting to the remote host.
self.connectStart = connectStart
# Connected to the remote host.
self.connectEnd = connectEnd
# Started SSL handshake.
self.sslStart = sslStart
# Finished SSL handshake.
self.sslEnd = sslEnd
# Started running ServiceWorker.
self.workerStart = workerStart
# Finished Starting ServiceWorker.
self.workerReady = workerReady
# Started sending request.
self.sendStart = sendStart
# Finished sending request.
self.sendEnd = sendEnd
# Time the server started pushing request.
self.pushStart = pushStart
# Time the server finished pushing request.
self.pushEnd = pushEnd
# Finished receiving response headers.
self.receiveHeadersEnd = receiveHeadersEnd
ResourcePriority = Enum("ResourcePriority", "VeryLow Low Medium High VeryHigh")
ResourcePriority.__doc__ = "Loading priority of a resource request."
class Request:
"""HTTP request data."""
def __init__(self, url: str, method: str, headers: "Headers", initialPriority: "ResourcePriority", referrerPolicy: str, postData: str=None, mixedContentType: str=None):
# Request URL.
self.url = url
# HTTP request method.
self.method = method
# HTTP request headers.
self.headers = headers
# Priority of the resource request at the time request is sent.
self.initialPriority = initialPriority
# The referrer policy of the request, as defined in https://www.w3.org/TR/referrer-policy/
self.referrerPolicy = referrerPolicy
# HTTP POST request data.
self.postData = postData
# The mixed content status of the request, as defined in http://www.w3.org/TR/mixed-content/
self.mixedContentType = mixedContentType
class SignedCertificateTimestamp:
"""Details of a signed certificate timestamp (SCT)."""
def __init__(self, status: str, origin: str, logDescription: str, logId: str, timestamp: "Timestamp", hashAlgorithm: str, signatureAlgorithm: str, signatureData: str):
# Validation status.
self.status = status
# Origin.
self.origin = origin
# Log name / description.
self.logDescription = logDescription
# Log ID.
self.logId = logId
# Issuance date.
self.timestamp = timestamp
# Hash algorithm.
self.hashAlgorithm = hashAlgorithm
# Signature algorithm.
self.signatureAlgorithm = signatureAlgorithm
# Signature data.
self.signatureData = signatureData
class SecurityDetails:
"""Security details about a request."""
def __init__(self, protocol: str, keyExchange: str, cipher: str, certificateId: "Security.CertificateId", subjectName: str, sanList: List, issuer: str, validFrom: "Timestamp", validTo: "Timestamp", signedCertificateTimestampList: List, keyExchangeGroup: str=None, mac: str=None):
# Protocol name (e.g. "TLS 1.2" or "QUIC").
self.protocol = protocol
# Key Exchange used by the connection, or the empty string if not applicable.
self.keyExchange = keyExchange
# Cipher name.
self.cipher = cipher
# Certificate ID value.
self.certificateId = certificateId
# Certificate subject name.
self.subjectName = subjectName
# Subject Alternative Name (SAN) DNS names and IP addresses.
self.sanList = sanList
# Name of the issuing CA.
self.issuer = issuer
# Certificate valid from date.
self.validFrom = validFrom
# Certificate valid to (expiration) date
self.validTo = validTo
# List of signed certificate timestamps (SCTs).
self.signedCertificateTimestampList = signedCertificateTimestampList
# (EC)DH group used by the connection, if applicable.
self.keyExchangeGroup = keyExchangeGroup
# TLS MAC. Note that AEAD ciphers do not have separate MACs.
self.mac = mac
BlockedReason = Enum("BlockedReason", "csp mixed-content origin inspector subresource-filter other")
BlockedReason.__doc__ = "The reason why request was blocked."
class Response:
"""HTTP response data."""
def __init__(self, url: str, status: float, statusText: str, headers: "Headers", mimeType: str, connectionReused: bool, connectionId: float, encodedDataLength: float, securityState: "Security.SecurityState", headersText: str=None, requestHeaders: "Headers"=None, requestHeadersText: str=None, remoteIPAddress: str=None, remotePort: int=None, fromDiskCache: bool=None, fromServiceWorker: bool=None, timing: "ResourceTiming"=None, protocol: str=None, securityDetails: "SecurityDetails"=None):
# Response URL. This URL can be different from CachedResource.url in case of redirect.
self.url = url
# HTTP response status code.
self.status = status
# HTTP response status text.
self.statusText = statusText
# HTTP response headers.
self.headers = headers
# Resource mimeType as determined by the browser.
self.mimeType = mimeType
# Specifies whether physical connection was actually reused for this request.
self.connectionReused = connectionReused
# Physical connection id that was actually used for this request.
self.connectionId = connectionId
# Total number of bytes received for this request so far.
self.encodedDataLength = encodedDataLength
# Security state of the request resource.
self.securityState = securityState
# HTTP response headers text.
self.headersText = headersText
# Refined HTTP request headers that were actually transmitted over the network.
self.requestHeaders = requestHeaders
# HTTP request headers text.
self.requestHeadersText = requestHeadersText
# Remote IP address.
self.remoteIPAddress = remoteIPAddress
# Remote port.
self.remotePort = remotePort
# Specifies that the request was served from the disk cache.
self.fromDiskCache = fromDiskCache
# Specifies that the request was served from the ServiceWorker.
self.fromServiceWorker = fromServiceWorker
# Timing information for the given request.
self.timing = timing
# Protocol used to fetch this request.
self.protocol = protocol
# Security details for the request.
self.securityDetails = securityDetails
class WebSocketRequest:
"""WebSocket request data."""
def __init__(self, headers: "Headers"):
# HTTP request headers.
self.headers = headers
class WebSocketResponse:
"""WebSocket response data."""
def __init__(self, status: float, statusText: str, headers: "Headers", headersText: str=None, requestHeaders: "Headers"=None, requestHeadersText: str=None):
# HTTP response status code.
self.status = status
# HTTP response status text.
self.statusText = statusText
# HTTP response headers.
self.headers = headers
# HTTP response headers text.
self.headersText = headersText
# HTTP request headers.
self.requestHeaders = requestHeaders
# HTTP request headers text.
self.requestHeadersText = requestHeadersText
class WebSocketFrame:
"""WebSocket frame data."""
def __init__(self, opcode: float, mask: bool, payloadData: str):
# WebSocket frame opcode.
self.opcode = opcode
# WebSocke frame mask.
self.mask = mask
# WebSocke frame payload data.
self.payloadData = payloadData
class CachedResource:
"""Information about the cached resource."""
def __init__(self, url: str, type: "Page.ResourceType", bodySize: float, response: "Response"=None):
# Resource URL. This is the url of the original network request.
self.url = url
# Type of this resource.
self.type = type
# Cached response body size.
self.bodySize = bodySize
# Cached response data.
self.response = response
class Initiator:
"""Information about the request initiator."""
def __init__(self, type: str, stack: "Runtime.StackTrace"=None, url: str=None, lineNumber: float=None):
# Type of this initiator.
self.type = type
# Initiator JavaScript stack trace, set for Script only.
self.stack = stack
# Initiator URL, set for Parser type only.
self.url = url
# Initiator line number, set for Parser type only (0-based).
self.lineNumber = lineNumber
class Cookie:
"""Cookie object"""
def __init__(self, name: str, value: str, domain: str, path: str, expires: float, size: int, httpOnly: bool, secure: bool, session: bool, sameSite: "CookieSameSite"=None):
# Cookie name.
self.name = name
# Cookie value.
self.value = value
# Cookie domain.
self.domain = domain
# Cookie path.
self.path = path
# Cookie expiration date as the number of seconds since the UNIX epoch.
self.expires = expires
# Cookie size.
self.size = size
# True if cookie is http-only.
self.httpOnly = httpOnly
# True if cookie is secure.
self.secure = secure
# True in case of session cookie.
self.session = session
# Cookie SameSite type.
self.sameSite = sameSite
class enable(ChromeCommand):
"""Enables network tracking, network events will now be delivered to the client."""
def __init__(self, maxTotalBufferSize: int=None, maxResourceBufferSize: int=None):
# Buffer size in bytes to use when preserving network payloads (XHRs, etc).
self.maxTotalBufferSize = maxTotalBufferSize
# Per-resource buffer size in bytes to use when preserving network payloads (XHRs, etc).
self.maxResourceBufferSize = maxResourceBufferSize
class disable(ChromeCommand):
"""Disables network tracking, prevents network events from being sent to the client."""
def __init__(self): pass
class setUserAgentOverride(ChromeCommand):
"""Allows overriding user agent with the given string."""
def __init__(self, userAgent: str):
# User agent to use.
self.userAgent = userAgent
class setExtraHTTPHeaders(ChromeCommand):
"""Specifies whether to always send extra HTTP headers with the requests from this page."""
def __init__(self, headers: "Headers"):
# Map with extra HTTP headers.
self.headers = headers
class getResponseBody(ChromeCommand):
"""Returns content served for the given request."""
def __init__(self, requestId: "RequestId"):
# Identifier of the network request to get content for.
self.requestId = requestId
class addBlockedURL(ChromeCommand):
"""Blocks specific URL from loading."""
def __init__(self, url: str):
# URL to block.
self.url = url
class removeBlockedURL(ChromeCommand):
"""Cancels blocking of a specific URL from loading."""
def __init__(self, url: str):
# URL to stop blocking.
self.url = url
class replayXHR(ChromeCommand):
"""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."""
def __init__(self, requestId: "RequestId"):
# Identifier of XHR to replay.
self.requestId = requestId
class setMonitoringXHREnabled(ChromeCommand):
"""Toggles monitoring of XMLHttpRequest. If <code>true</code>, console will receive messages upon each XHR issued."""
def __init__(self, enabled: bool):
# Monitoring enabled state.
self.enabled = enabled
class canClearBrowserCache(ChromeCommand):
"""Tells whether clearing browser cache is supported."""
def __init__(self): pass
class clearBrowserCache(ChromeCommand):
"""Clears browser cache."""
def __init__(self): pass
class canClearBrowserCookies(ChromeCommand):
"""Tells whether clearing browser cookies is supported."""
def __init__(self): pass
class clearBrowserCookies(ChromeCommand):
"""Clears browser cookies."""
def __init__(self): pass
class getCookies(ChromeCommand):
"""Returns all browser cookies for the current URL. Depending on the backend support, will return detailed cookie information in the <code>cookies</code> field."""
def __init__(self): pass
class getAllCookies(ChromeCommand):
"""Returns all browser cookies. Depending on the backend support, will return detailed cookie information in the <code>cookies</code> field."""
def __init__(self): pass
class deleteCookie(ChromeCommand):
"""Deletes browser cookie with given name, domain and path."""
def __init__(self, cookieName: str, url: str):
# Name of the cookie to remove.
self.cookieName = cookieName
# URL to match cooke domain and path.
self.url = url
class setCookie(ChromeCommand):
"""Sets a cookie with the given cookie data; may overwrite equivalent cookies if they exist."""
def __init__(self, url: str, name: str, value: str, domain: str=None, path: str=None, secure: bool=None, httpOnly: bool=None, sameSite: "CookieSameSite"=None, expirationDate: "Timestamp"=None):
# 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.
self.url = url
# The name of the cookie.
self.name = name
# The value of the cookie.
self.value = value
# If omitted, the cookie becomes a host-only cookie.
self.domain = domain
# Defaults to the path portion of the url parameter.
self.path = path
# Defaults ot false.
self.secure = secure
# Defaults to false.
self.httpOnly = httpOnly
# Defaults to browser default behavior.
self.sameSite = sameSite
# If omitted, the cookie becomes a session cookie.
self.expirationDate = expirationDate
class canEmulateNetworkConditions(ChromeCommand):
"""Tells whether emulation of network conditions is supported."""
def __init__(self): pass
class emulateNetworkConditions(ChromeCommand):
"""Activates emulation of network conditions."""
def __init__(self, offline: bool, latency: float, downloadThroughput: float, uploadThroughput: float, connectionType: "ConnectionType"=None):
# True to emulate internet disconnection.
self.offline = offline
# Additional latency (ms).
self.latency = latency
# Maximal aggregated download throughput.
self.downloadThroughput = downloadThroughput
# Maximal aggregated upload throughput.
self.uploadThroughput = uploadThroughput
# Connection type if known.
self.connectionType = connectionType
class setCacheDisabled(ChromeCommand):
"""Toggles ignoring cache for each request. If <code>true</code>, cache will not be used."""
def __init__(self, cacheDisabled: bool):
# Cache disabled state.
self.cacheDisabled = cacheDisabled
class setBypassServiceWorker(ChromeCommand):
"""Toggles ignoring of service worker for each request."""
def __init__(self, bypass: bool):
# Bypass service worker and load from network.
self.bypass = bypass
class setDataSizeLimitsForTest(ChromeCommand):
"""For testing."""
def __init__(self, maxTotalSize: int, maxResourceSize: int):
# Maximum total buffer size.
self.maxTotalSize = maxTotalSize
# Maximum per-resource size.
self.maxResourceSize = maxResourceSize
class getCertificate(ChromeCommand):
"""Returns the DER-encoded certificate."""
def __init__(self, origin: str):
# Origin to get certificate for.
self.origin = origin