implement UDP streams (mitmproxy#6557)

based on mitmproxy/mitmproxy_rs#127

mitmproxy/addons/keepserving.py

@@ -3,6 +3,7 @@
 import asyncio
 
 from mitmproxy import ctx
+from mitmproxy.utils import asyncio_utils
 
 
 class KeepServing:
@@ -44,4 +45,6 @@ def running(self):
             ctx.options.rfile,
         ]
         if any(opts) and not ctx.options.keepserving:
-            self._watch_task = asyncio.get_running_loop().create_task(self.watch())
+            self._watch_task = asyncio_utils.create_task(
+                self.watch(), name="keepserving"
+            )

mitmproxy/addons/proxyserver.py

@@ -33,6 +33,7 @@
 from mitmproxy.proxy.mode_servers import ProxyConnectionHandler
 from mitmproxy.proxy.mode_servers import ServerInstance
 from mitmproxy.proxy.mode_servers import ServerManager
+from mitmproxy.utils import asyncio_utils
 from mitmproxy.utils import human
 from mitmproxy.utils import signals
 
@@ -276,7 +277,9 @@ def configure(self, updated) -> None:
                     )
 
             if self.is_running:
-                self._update_task = asyncio.create_task(self.servers.update(modes))
+                self._update_task = asyncio_utils.create_task(
+                    self.servers.update(modes), name="update servers"
+                )
 
     async def setup_servers(self) -> bool:
         """Setup proxy servers. This may take an indefinite amount of time to complete (e.g. on permission prompts)."""

mitmproxy/addons/readfile.py

@@ -11,6 +11,8 @@
 from mitmproxy import flowfilter
 from mitmproxy import io
 
+logger = logging.getLogger(__name__)
+
 
 class ReadFile:
     """
@@ -68,7 +70,7 @@ async def doread(self, rfile: str) -> None:
         try:
             await self.load_flows_from_path(rfile)
         except exceptions.FlowReadException as e:
-            raise exceptions.OptionsError(e) from e
+            logger.exception(f"Failed to read {ctx.options.rfile}: {e}")
         finally:
             self._read_task = None
 

mitmproxy/addons/script.py

@@ -128,6 +128,9 @@ def loadscript(self):
                 ctx.master.addons.invoke_addon_sync(self.ns, hooks.RunningHook())
 
     async def watcher(self):
+        # Script loading is terminally confused at the moment.
+        # This here is a stopgap workaround to defer loading.
+        await asyncio.sleep(0)
         last_mtime = 0.0
         while True:
             try:

mitmproxy/certs.py

@@ -100,16 +100,20 @@ def issuer(self) -> list[tuple[str, str]]:
 
     @property
     def notbefore(self) -> datetime.datetime:
+        # TODO: Use self._cert.not_valid_before_utc once cryptography 42 hits.
         # x509.Certificate.not_valid_before is a naive datetime in UTC
         return self._cert.not_valid_before.replace(tzinfo=datetime.timezone.utc)
 
     @property
     def notafter(self) -> datetime.datetime:
+        # TODO: Use self._cert.not_valid_after_utc once cryptography 42 hits.
         # x509.Certificate.not_valid_after is a naive datetime in UTC
         return self._cert.not_valid_after.replace(tzinfo=datetime.timezone.utc)
 
     def has_expired(self) -> bool:
-        return datetime.datetime.utcnow() > self._cert.not_valid_after
+        if sys.version_info < (3, 11):  # pragma: no cover
+            return datetime.datetime.utcnow() > self._cert.not_valid_after
+        return datetime.datetime.now(datetime.UTC) > self.notafter
 
     @property
     def subject(self) -> list[tuple[str, str]]:

mitmproxy/master.py

@@ -4,6 +4,7 @@
 from . import ctx as mitmproxy_ctx
 from .addons import termlog
 from .proxy.mode_specs import ReverseMode
+from .utils import asyncio_utils
 from mitmproxy import addonmanager
 from mitmproxy import command
 from mitmproxy import eventsequence
@@ -51,9 +52,10 @@ def __init__(
         mitmproxy_ctx.options = self.options
 
     async def run(self) -> None:
-        old_handler = self.event_loop.get_exception_handler()
-        self.event_loop.set_exception_handler(self._asyncio_exception_handler)
-        try:
+        with (
+            asyncio_utils.install_exception_handler(self._asyncio_exception_handler),
+            asyncio_utils.set_eager_task_factory(),
+        ):
             self.should_exit.clear()
 
             if ec := self.addons.get("errorcheck"):
@@ -67,17 +69,15 @@ async def run(self) -> None:
                     ],
                     return_when=asyncio.FIRST_COMPLETED,
                 )
+            await self.running()
             if ec := self.addons.get("errorcheck"):
                 await ec.shutdown_if_errored()
                 ec.finish()
-            await self.running()
             try:
                 await self.should_exit.wait()
             finally:
                 # .wait might be cancelled (e.g. by sys.exit)
                 await self.done()
-        finally:
-            self.event_loop.set_exception_handler(old_handler)
 
     def shutdown(self):
         """