typing list, tuple, dict

Author: cle-b

httpdbg/args.py

@@ -1,12 +1,11 @@
 # -*- coding: utf-8 -*-
 import argparse
 from pathlib import Path
-from typing import List, Tuple
 
 from httpdbg.log import LogLevel
 
 
-def read_args(args: List[str]) -> Tuple[argparse.Namespace, List[str]]:
+def read_args(args: list[str]) -> tuple[argparse.Namespace, list[str]]:
     httpdbg_args = args
     client_args = []
     for action in ["--console", "--module", "-m", "--script"]:

httpdbg/hooks/all.py

@@ -1,8 +1,6 @@
 # -*- coding: utf-8 -*-
 from contextlib import contextmanager
 from typing import Generator
-from typing import List
-from typing import Tuple
 from typing import Union
 
 from httpdbg.hooks.aiohttp import hook_aiohttp
@@ -26,10 +24,10 @@
 @contextmanager
 def httprecord(
     records: HTTPRecords = None,
-    initiators: Union[List[str], None] = None,
+    initiators: Union[list[str], None] = None,
     client: bool = True,
     server: bool = False,
-    ignore: Tuple[Tuple[str, int], ...] = (),
+    ignore: tuple[tuple[str, int], ...] = (),
     multiprocess: bool = True,
 ) -> Generator[HTTPRecords, None, None]:
     if records is None:

httpdbg/hooks/external.py

@@ -7,7 +7,6 @@
 import tempfile
 import time
 import threading
-from typing import List
 from typing import Union
 
 from httpdbg.env import HTTPDBG_MULTIPROCESS_DIR
@@ -18,7 +17,7 @@
 @contextmanager
 def watcher_external(
     records: HTTPRecords,
-    initiators: Union[List[str], None] = None,
+    initiators: Union[list[str], None] = None,
     server: bool = False,
 ) -> Generator[HTTPRecords, None, None]:
     if HTTPDBG_MULTIPROCESS_DIR not in os.environ:

httpdbg/hooks/fastapi.py

@@ -3,7 +3,6 @@
 from contextlib import contextmanager
 import functools
 from functools import wraps
-from typing import Dict
 from typing import Generator
 from typing import Union
 
@@ -37,7 +36,7 @@ def hook(*args, **kwargs):
 def set_hook_fastapi_apirouter_add_api_route(
     records: HTTPRecords,
     method: Callable,
-    already_mapped: Union[Dict[Callable, Callable], None] = None,
+    already_mapped: Union[dict[Callable, Callable], None] = None,
 ):
 
     @wraps(method)
@@ -123,7 +122,7 @@ def hook_fastapi(records: HTTPRecords) -> Generator[None, None, None]:
     try:
         import fastapi.routing
 
-        already_mapped: Dict[Callable, Callable] = {}
+        already_mapped: dict[Callable, Callable] = {}
 
         set_hook_fastapi_apirouter_add_api_route_with_already_mapped = (
             functools.partial(

httpdbg/hooks/flask.py

@@ -3,7 +3,6 @@
 from contextlib import contextmanager
 import functools
 from functools import wraps
-from typing import Dict
 from typing import Generator
 from typing import Union
 
@@ -36,7 +35,7 @@ def hook(*args, **kwargs):
 def set_hook_flask_add_url_rule(
     records: HTTPRecords,
     method: Callable,
-    already_mapped: Union[Dict[Callable, Callable], None] = None,
+    already_mapped: Union[dict[Callable, Callable], None] = None,
 ):
 
     def hook(*args, **kwargs):
@@ -67,7 +66,7 @@ def hook(*args, **kwargs):
 def set_hook_flask_register_error_handler(
     records: HTTPRecords,
     method: Callable,
-    already_mapped: Union[Dict[Callable, Callable], None] = None,
+    already_mapped: Union[dict[Callable, Callable], None] = None,
 ):
 
     def hook(*args, **kwargs):
@@ -103,7 +102,7 @@ def hook_flask(records: HTTPRecords) -> Generator[None, None, None]:
 
         # we must not apply the hook more than once on a mapped endpoint function
         # AssertionError: View function mapping is overwriting an existing endpoint function: xxxx
-        already_mapped: Dict[Callable, Callable] = {}
+        already_mapped: dict[Callable, Callable] = {}
 
         set_hook_flask_add_url_rule_with_already_mapped = functools.partial(
             set_hook_flask_add_url_rule, already_mapped=already_mapped

httpdbg/hooks/generic.py

@@ -9,7 +9,6 @@
 from types import ModuleType
 from typing import Any
 from typing import Generator
-from typing import List
 from typing import Union
 
 from httpdbg.hooks.utils import decorate
@@ -49,13 +48,13 @@ def hook(*args, **kwargs):
 
 @contextmanager
 def hook_generic(
-    records: HTTPRecords, initiators: Union[List[str], None] = None
+    records: HTTPRecords, initiators: Union[list[str], None] = None
 ) -> Generator[None, None, None]:
     if initiators:
         # we add a hook for a generic initiator only if the module is imported
 
-        hooks: List[Callable] = []
-        already_hooked: List[str] = []
+        hooks: list[Callable] = []
+        already_hooked: list[str] = []
 
         original_builtin_import = builtins.__import__
 
@@ -130,7 +129,7 @@ def __hook__import__(
                 fnc = undecorate(fnc)
 
 
-def list_callables_from_module(records: HTTPRecords, module: str) -> List[Any]:
+def list_callables_from_module(records: HTTPRecords, module: str) -> list[Any]:
     callables = []
 
     try:
@@ -180,7 +179,7 @@ def list_callables_from_class(
     imported_module: ModuleType,
     module: str,
     classname: str,
-) -> List[Any]:
+) -> list[Any]:
     callables = []
 
     try:

httpdbg/hooks/h2.py

@@ -0,0 +1,117 @@
+# -*- coding: utf-8 -*-
+from collections.abc import Callable
+from contextlib import contextmanager
+import traceback
+from typing import Generator
+
+from httpdbg.hooks.utils import getcallargs
+from httpdbg.hooks.utils import decorate
+from httpdbg.hooks.utils import undecorate
+from httpdbg.initiator import httpdbg_initiator
+from httpdbg.records import HTTPRecords
+
+
+def set_hook_for_h2_send_headers(records: HTTPRecords, method: Callable):
+    def hook(*args, **kwargs):
+        callargs = getcallargs(method, *args, **kwargs)
+        print(callargs)
+        initiator_and_group = None
+        try:
+            with httpdbg_initiator(
+                records, traceback.extract_stack(), method, *args, **kwargs
+            ) as initiator_and_group:
+                ret = method(*args, **kwargs)
+            return ret
+        except Exception as ex:
+            raise
+            # callargs = getcallargs(method, *args, **kwargs)
+
+            # if "url" in callargs:
+            #     if initiator_and_group:
+            #         initiator, group, is_new = initiator_and_group
+            #         if is_new:
+            #             records.add_new_record_exception(
+            #                 initiator, group, str(callargs["url"]), ex
+            #             )
+            # raise
+
+    return hook
+
+
+def set_hook_for_h2_send_data(records: HTTPRecords, method: Callable):
+    def hook(*args, **kwargs):
+        callargs = getcallargs(method, *args, **kwargs)
+        print(callargs)
+        initiator_and_group = None
+        try:
+            with httpdbg_initiator(
+                records, traceback.extract_stack(), method, *args, **kwargs
+            ) as initiator_and_group:
+                ret = method(*args, **kwargs)
+            return ret
+        except Exception as ex:
+            raise
+            # callargs = getcallargs(method, *args, **kwargs)
+
+            # if "url" in callargs:
+            #     if initiator_and_group:
+            #         initiator, group, is_new = initiator_and_group
+            #         if is_new:
+            #             records.add_new_record_exception(
+            #                 initiator, group, str(callargs["url"]), ex
+            #             )
+            # raise
+
+    return hook
+
+
+def set_hook_for_h2_receive_data(records: HTTPRecords, method: Callable):
+    def hook(*args, **kwargs):
+        # callargs = getcallargs(method, *args, **kwargs)
+        # print(callargs)
+        initiator_and_group = None
+        try:
+            with httpdbg_initiator(
+                records, traceback.extract_stack(), method, *args, **kwargs
+            ) as initiator_and_group:
+                ret = method(*args, **kwargs)
+            for event in ret:
+                print(type(event))
+                print(event)
+            return ret
+        except Exception as ex:
+            raise
+            # callargs = getcallargs(method, *args, **kwargs)
+
+            # if "url" in callargs:
+            #     if initiator_and_group:
+            #         initiator, group, is_new = initiator_and_group
+            #         if is_new:
+            #             records.add_new_record_exception(
+            #                 initiator, group, str(callargs["url"]), ex
+            #             )
+            # raise
+
+    return hook
+
+
+@contextmanager
+def hook_h2(records: HTTPRecords) -> Generator[None, None, None]:
+    hooks = False
+    try:
+        import h2.connection
+
+        h2.connection.H2Connection.send_headers = decorate(records, h2.connection.H2Connection.send_headers, set_hook_for_h2_send_headers)  # type: ignore[arg-type]
+        h2.connection.H2Connection.send_data = decorate(records, h2.connection.H2Connection.send_data, set_hook_for_h2_send_data)  # type: ignore[arg-type]
+        h2.connection.H2Connection.receive_data = decorate(records, h2.connection.H2Connection.receive_data, set_hook_for_h2_receive_data)  # type: ignore[arg-type]
+
+        hooks = True
+    except ImportError:
+        pass
+
+    yield
+
+    if hooks:
+        h2.connection.H2Connection.send_headers = undecorate(h2.connection.H2Connection.send_headers)  # type: ignore[arg-type]
+        h2.connection.H2Connection.send_data = undecorate(h2.connection.H2Connection.send_data)  # type: ignore[arg-type]
+        h2.connection.H2Connection.receive_data = undecorate(h2.connection.H2Connection.receive_data)  # type: ignore[arg-type]

httpdbg/hooks/record.py

@@ -1,8 +1,6 @@
 # -*- coding: utf-8 -*-
 from abc import ABC, abstractmethod
 import datetime
-from typing import List
-from typing import Tuple
 from typing import Union
 
 from httpdbg.utils import get_new_uuid
@@ -23,7 +21,7 @@ def rawheaders(self) -> bytes:
 
     @property
     @abstractmethod
-    def headers(self) -> List[HTTPDBGHeader]:
+    def headers(self) -> list[HTTPDBGHeader]:
         pass
 
     @property
@@ -51,7 +49,7 @@ class HTTPRecordRequest(HTTPRecordReqResp, ABC):
 
     @property
     @abstractmethod
-    def cookies(self) -> List[HTTPDBGCookie]:
+    def cookies(self) -> list[HTTPDBGCookie]:
         pass
 
     @property
@@ -74,7 +72,7 @@ class HTTPRecordResponse(HTTPRecordReqResp, ABC):
 
     @property
     @abstractmethod
-    def cookies(self) -> List[HTTPDBGCookie]:
+    def cookies(self) -> list[HTTPDBGCookie]:
         pass
 
     @property
@@ -107,7 +105,7 @@ def __init__(
         is_client: bool = True,
     ) -> None:
         self.id = get_new_uuid()
-        self.address: Tuple[str, int] = ("", 0)
+        self.address: tuple[str, int] = ("", 0)
         self._url: Union[str, None] = None
         self.initiator_id: Union[str, None] = initiator_id
         self.exception: Union[Exception, None] = None

httpdbg/hooks/recordhttp1.py

@@ -1,6 +1,5 @@
 import datetime
 from urllib.parse import urlparse
-from typing import List
 
 from httpdbg.preview import generate_preview
 from httpdbg.hooks.record import HTTPRecord
@@ -18,7 +17,7 @@ class HTTP1RecordReqResp(HTTPRecordReqResp):
     def __init__(self) -> None:
         self._rawdata: bytes = bytes()
         self._rawheaders: bytes = bytes()
-        self._headers: List[HTTPDBGHeader] = []
+        self._headers: list[HTTPDBGHeader] = []
         self.last_update: datetime.datetime = datetime.datetime.now(
             datetime.timezone.utc
         )
@@ -38,7 +37,7 @@ def rawheaders(self) -> bytes:
         return self._rawheaders
 
     @property
-    def headers(self) -> List[HTTPDBGHeader]:
+    def headers(self) -> list[HTTPDBGHeader]:
         if not self._headers:
             if self.rawheaders:
                 for header in self.rawheaders[self.rawheaders.find(b"\r\n") :].split(
@@ -98,7 +97,7 @@ def __init__(self) -> None:
         self._protocol = bytes()
 
     @property
-    def cookies(self) -> List[HTTPDBGCookie]:
+    def cookies(self) -> list[HTTPDBGCookie]:
         return list_cookies_headers_request_simple_cookies(self.headers)
 
     def _parse_first_line(self) -> None:
@@ -133,7 +132,7 @@ def __init__(self):
         self._message = bytes()
 
     @property
-    def cookies(self) -> List[HTTPDBGCookie]:
+    def cookies(self) -> list[HTTPDBGCookie]:
         return list_cookies_headers_response_simple_cookies(self.headers)
 
     def _parse_first_line(self) -> None:

httpdbg/hooks/socket.py

@@ -10,7 +10,7 @@
 import sys
 import traceback
 from typing import Generator
-from typing import Dict, Tuple, Union
+from typing import Union
 
 from httpdbg.initiator import httpdbg_initiator
 from httpdbg.log import logger
@@ -25,9 +25,9 @@
 class SocketRawData(object):
     """Store the request data without encryption, even when using an SSLSocket."""
 
-    def __init__(self, id: int, address: Tuple[str, int], ssl: bool) -> None:
+    def __init__(self, id: int, address: tuple[str, int], ssl: bool) -> None:
         self.id: int = id
-        self.address: Tuple[str, int] = address
+        self.address: tuple[str, int] = address
         self.ssl: bool = ssl
         self._rawdata: bytes = bytes()
         self.record: Union[HTTP1Record, None] = None
@@ -66,10 +66,10 @@ class TracerHTTP1:
 
     def __init__(
         self,
-        ignore: Tuple[Tuple[str, int], ...] = (),
+        ignore: tuple[tuple[str, int], ...] = (),
     ):
-        self.sockets: Dict[int, SocketRawData] = {}
-        self.ignore: Tuple[Tuple[str, int], ...] = ignore
+        self.sockets: dict[int, SocketRawData] = {}
+        self.ignore: tuple[tuple[str, int], ...] = ignore
 
     def get_socket_data(
         self, obj, extra_sock=None, force_new=False, request=None, is_uvicorn=False