feat: add methods to retrieve network response body and logs

Author: thalissonvs

pydoll/browser/tab.py

@@ -32,13 +32,15 @@
     IFrameNotFound,
     InvalidFileExtension,
     InvalidIFrame,
+    NetworkEventsNotEnabled,
     NoDialogPresent,
     NotAnIFrame,
     PageLoadTimeout,
     WaitElementTimeout,
 )
 from pydoll.protocol.base import Response
-from pydoll.protocol.network.types import Cookie, CookieParam
+from pydoll.protocol.network.responses import GetResponseBodyResponse
+from pydoll.protocol.network.types import Cookie, CookieParam, NetworkLog
 from pydoll.protocol.page.events import PageEvent
 from pydoll.protocol.page.responses import CaptureScreenshotResponse, PrintToPDFResponse
 from pydoll.protocol.runtime.responses import EvaluateResponse
@@ -313,6 +315,50 @@ async def get_cookies(self) -> list[Cookie]:
         )
         return response['result']['cookies']
 
+    async def get_network_response_body(self, request_id: str) -> str:
+        """
+        Get the response body for a given request ID.
+
+        Args:
+            request_id: Request ID to get the response body for.
+
+        Returns:
+            The response body for the given request ID.
+
+        Raises:
+            NetworkEventsNotEnabled: If network events are not enabled.
+        """
+        if not self.network_events_enabled:
+            raise NetworkEventsNotEnabled('Network events must be enabled to get response body')
+
+        response: GetResponseBodyResponse = await self._execute_command(
+            NetworkCommands.get_response_body(request_id)
+        )
+        return response['result']['body']
+
+    async def get_network_logs(self, filter: Optional[str] = None) -> list[NetworkLog]:
+        """
+        Get network logs.
+
+        Args:
+            filter: Filter to apply to the network logs.
+
+        Returns:
+            The network logs.
+
+        Raises:
+            NetworkEventsNotEnabled: If network events are not enabled.
+        """
+        if not self.network_events_enabled:
+            raise NetworkEventsNotEnabled('Network events must be enabled to get network logs')
+
+        logs = self._connection_handler.network_logs
+        if filter:
+            logs = [
+                log for log in logs if filter in log['params'].get('request', {}).get('url', '')
+            ]
+        return logs
+
     async def set_cookies(self, cookies: list[CookieParam]):
         """
         Set multiple cookies for current page.

pydoll/constants.py

@@ -913,3 +913,63 @@ class DialogType(str, Enum):
     CONFIRM = 'confirm'
     PROMPT = 'prompt'
     BEFORE_UNLOAD = 'beforeunload'
+
+
+class InitiatorType(str, Enum):
+    PARSER = 'parser'
+    SCRIPT = 'script'
+    PRELOAD = 'preload'
+    SIGNED_EXCHANGE = 'SignedExchange'
+    PREFLIGHT = 'preflight'
+    OTHER = 'other'
+
+
+class NetworkServiceWorkerRouterSourceType(str, Enum):
+    """Network service worker router source types."""
+
+    NETWORK = 'network'
+    CACHE = 'cache'
+    FETCH_EVENT = 'fetch-event'
+    RACE_NETWORK = 'race-network'
+    RACE_NETWORK_AND_FETCH_HANDLER = 'race-network-and-fetch-handler'
+    RACE_NETWORK_AND_CACHE = 'race-network-and-cache'
+
+
+class NetworkServiceWorkerResponseSource(str, Enum):
+    """Network service worker response source types."""
+
+    CACHE_STORAGE = 'cache-storage'
+    HTTP_CACHE = 'http-cache'
+    FALLBACK_CODE = 'fallback-code'
+    NETWORK = 'network'
+
+
+class AlternateProtocolUsage(str, Enum):
+    """Alternate protocol usage types."""
+
+    ALTERNATIVE_JOB_WON_WITHOUT_RACE = 'alternativeJobWonWithoutRace'
+    ALTERNATIVE_JOB_WON_RACE = 'alternativeJobWonRace'
+    MAIN_JOB_WON_RACE = 'mainJobWonRace'
+    MAPPING_MISSING = 'mappingMissing'
+    BROKEN = 'broken'
+    DNS_ALPN_H3_JOB_WON_WITHOUT_RACE = 'dnsAlpnH3JobWonWithoutRace'
+    DNS_ALPN_H3_JOB_WON_RACE = 'dnsAlpnH3JobWonRace'
+    UNSPECIFIED_REASON = 'unspecifiedReason'
+
+
+class SecurityState(str, Enum):
+    """Security state types."""
+
+    UNKNOWN = 'unknown'
+    NEUTRAL = 'neutral'
+    INSECURE = 'insecure'
+    INFO = 'info'
+    INSECURE_BROKEN = 'insecure-broken'
+
+
+class CertificateTransparencyCompliance(str, Enum):
+    """Certificate transparency compliance types."""
+
+    UNKNOWN = 'unknown'
+    NOT_COMPLIANT = 'not-compliant'
+    COMPLIANT = 'compliant'

pydoll/exceptions.py

@@ -233,3 +233,9 @@ class IFrameNotFound(PydollException):
     """Raised when an iframe is not found."""
 
     message = 'The iframe was not found'
+
+
+class NetworkEventsNotEnabled(PydollException):
+    """Raised when network events are not enabled."""
+
+    message = 'Network events not enabled'

pydoll/protocol/network/types.py

@@ -1,16 +1,24 @@
 from typing import NotRequired, TypedDict
 
 from pydoll.constants import (
+    AlternateProtocolUsage,
+    CertificateTransparencyCompliance,
     ContentSecurityPolicySource,
     CookiePriority,
     CookieSameSite,
     CookieSourceScheme,
+    InitiatorType,
     MixedContentType,
+    NetworkServiceWorkerResponseSource,
+    NetworkServiceWorkerRouterSourceType,
     ReferrerPolicy,
     RefreshPolicy,
     ResourcePriority,
+    ResourceType,
+    SecurityState,
     TrustTokenOperationType,
 )
+from pydoll.protocol.runtime.types import StackTrace
 
 
 class SearchMatch(TypedDict):
@@ -150,3 +158,131 @@ class RequestPausedEventParams(TypedDict):
 class RequestPausedEvent(TypedDict):
     method: str
     params: RequestPausedEventParams
+
+
+class Initiator(TypedDict):
+    type: InitiatorType
+    stack: NotRequired[StackTrace]
+    url: NotRequired[str]
+    lineNumber: NotRequired[int]
+    columnNumber: NotRequired[int]
+    requestId: NotRequired[str]
+
+
+class ServiceWorkerRouterInfo(TypedDict):
+    """Service worker router info object."""
+
+    ruleIdMatched: NotRequired[int]
+    matchedSourceType: NotRequired[NetworkServiceWorkerRouterSourceType]
+    actualSourceType: NotRequired[NetworkServiceWorkerRouterSourceType]
+
+
+class ResourceTiming(TypedDict):
+    """Resource timing object."""
+
+    requestTime: float
+    proxyStart: float
+    proxyEnd: float
+    dnsStart: float
+    dnsEnd: float
+    connectStart: float
+    connectEnd: float
+    sslStart: float
+    sslEnd: float
+    workerStart: float
+    workerReady: float
+    workerFetchStart: float
+    workerRespondWithSettled: float
+    workerRouterEvaluationStart: NotRequired[float]
+    workerCacheLookupStart: NotRequired[float]
+    sendStart: float
+    sendEnd: float
+    pushStart: float
+    pushEnd: float
+    receiveHeadersStart: float
+    receiveHeadersEnd: float
+
+
+class SignedCertificateTimestamp(TypedDict):
+    """Signed certificate timestamp object."""
+
+    status: str
+    origin: str
+    logDescription: str
+    logId: str
+    timestamp: float
+    hashAlgorithm: str
+    signatureAlgorithm: str
+    signatureData: str
+
+
+class SecurityDetails(TypedDict):
+    """Security details object."""
+
+    protocol: str
+    keyExchange: str
+    keyExchangeGroup: NotRequired[str]
+    cipher: str
+    mac: NotRequired[str]
+    certificateId: int
+    subjectName: str
+    sanList: list[str]
+    issuer: str
+    validFrom: float
+    validTo: float
+    signedCertificateTimestampList: list[SignedCertificateTimestamp]
+    certificateTransparencyCompliance: CertificateTransparencyCompliance
+    serverSignatureAlgorithm: NotRequired[int]
+    encryptedClientHello: bool
+
+
+class Response(TypedDict):
+    url: str
+    status: int
+    statusText: str
+    headers: list[dict]
+    headersText: NotRequired[str]
+    mimeType: str
+    charset: str
+    requestHeaders: NotRequired[list[dict]]
+    requestHeadersText: NotRequired[str]
+    connectionReused: bool
+    connectionId: float
+    remoteIPAddress: NotRequired[str]
+    remotePort: NotRequired[int]
+    fromDiskCache: NotRequired[bool]
+    fromServiceWorker: NotRequired[bool]
+    fromPrefetchCache: NotRequired[bool]
+    fromEarlyHints: NotRequired[bool]
+    serviceWorkerRouterInfo: NotRequired[ServiceWorkerRouterInfo]
+    encodedDataLength: float
+    timing: NotRequired[ResourceTiming]
+    serviceWorkerResponseSource: NotRequired[NetworkServiceWorkerResponseSource]
+    responseTime: NotRequired[float]
+    cacheStorageCacheName: NotRequired[str]
+    protocol: NotRequired[str]
+    alternateProtocolUsage: NotRequired[AlternateProtocolUsage]
+    securityState: SecurityState
+    securityDetails: NotRequired[SecurityDetails]
+
+
+class NetworkLogParams(TypedDict):
+    requestId: str
+    loaderId: str
+    documentURL: str
+    request: Request
+    timestamp: float
+    wallTime: float
+    initiator: Initiator
+    redirectHasExtraInfo: bool
+    redirectResponse: NotRequired[Response]
+    type: NotRequired[ResourceType]
+    frameId: NotRequired[str]
+    hasUserGesture: NotRequired[bool]
+
+
+class NetworkLog(TypedDict):
+    """Network log object."""
+
+    method: str
+    params: RequestPausedEventParams