wip: initial attempt at adding webtransport to grizzly

.codecov.yml

@@ -1,3 +1,5 @@
+ignore:
+  - "grizzly/services/webtransport/wpt_h3_server"
 codecov:
   ci:
     - community-tc.services.mozilla.com

grizzly/conftest.py

@@ -14,6 +14,7 @@
 def session_setup(mocker):
     mocker.patch("grizzly.main.FuzzManagerReporter", autospec=True)
     mocker.patch("grizzly.main.Sapphire", autospec_set=True)
+    mocker.patch("grizzly.main.WebServices", autospec_set=True)
     adapter_cls = mocker.Mock(spec_set=Adapter)
     adapter_cls.return_value.RELAUNCH = Adapter.RELAUNCH
     adapter_cls.return_value.TIME_LIMIT = Adapter.TIME_LIMIT

grizzly/main.py

@@ -23,6 +23,7 @@
     package_version,
     time_limits,
 )
+from .services import WebServices
 from .session import LogRate, Session
 from .target import Target, TargetLaunchError, TargetLaunchTimeout
 
@@ -52,6 +53,7 @@ def main(args: Namespace) -> int:
     adapter: Optional[Adapter] = None
     certs: Optional[CertificateBundle] = None
     complete_with_results = False
+    ext_services = None
     target: Optional[Target] = None
     try:
         LOG.debug("initializing Adapter %r", args.adapter)
@@ -121,6 +123,9 @@ def main(args: Namespace) -> int:
         # launch http server used to serve test cases
         LOG.debug("starting Sapphire server")
         with Sapphire(auto_close=1, timeout=timeout, certs=certs) as server:
+            if certs is not None:
+                ext_services = WebServices.start_services(certs.host, certs.key)
+
             target.reverse(server.port, server.port)
             LOG.debug("initializing the Session")
             with Session(
@@ -149,6 +154,7 @@ def main(args: Namespace) -> int:
                     log_rate=log_rate,
                     launch_attempts=args.launch_attempts,
                     post_launch_delay=args.post_launch_delay,
+                    services=ext_services,
                 )
                 complete_with_results = session.status.results.total > 0
 
@@ -169,6 +175,8 @@ def main(args: Namespace) -> int:
         if adapter is not None:
             LOG.debug("calling adapter.cleanup()")
             adapter.cleanup()
+        if ext_services is not None:
+            ext_services.cleanup()
         if certs is not None:
             certs.cleanup()
         LOG.info("Done.")

grizzly/reduce/core.py

@@ -37,6 +37,7 @@
     time_limits,
 )
 from ..replay import ReplayManager, ReplayResult
+from ..services import WebServices
 from ..target import AssetManager, Target, TargetLaunchError, TargetLaunchTimeout
 from .exceptions import GrizzlyReduceBaseException, NotReproducible
 from .strategies import STRATEGIES
@@ -90,6 +91,7 @@ def __init__(
         relaunch: int = 1,
         report_period: Optional[int] = None,
         report_to_fuzzmanager: bool = False,
+        services=None,
         signature: Optional[CrashSignature] = None,
         signature_desc: Optional[str] = None,
         static_timeout: bool = False,
@@ -115,6 +117,7 @@ def __init__(
                       Target should be relaunched.
             report_period: Periodically report best results for long-running strategies.
             report_to_fuzzmanager: Report to FuzzManager rather than filesystem.
+            services (WebServices): WebServices instance.
             signature: Signature for accepting crashes.
             signature_desc: Short description of the given signature.
             static_timeout: Use only specified timeouts (`--timeout` and
@@ -153,6 +156,7 @@ def __init__(
         )
         self._use_analysis = use_analysis
         self._use_harness = use_harness
+        self._services = services
 
     def __enter__(self) -> "ReduceManager":
         return self
@@ -322,6 +326,7 @@ def run_reliability_analysis(self) -> Tuple[int, int]:
                     idle_delay=self._idle_delay,
                     idle_threshold=self._idle_threshold,
                     on_iteration_cb=self._on_replay_iteration,
+                    services=self._services,
                 )
                 try:
                     crashes = sum(x.count for x in results if x.expected)
@@ -525,6 +530,7 @@ def run(
                                     repeat=repeat,
                                     on_iteration_cb=self._on_replay_iteration,
                                     post_launch_delay=post_launch_delay,
+                                    services=self._services,
                                 )
                                 self._status.attempts += 1
                                 self.update_timeout(results)
@@ -777,6 +783,7 @@ def main(cls, args: Namespace) -> int:
 
         asset_mgr: Optional[AssetManager] = None
         certs = None
+        ext_services = None
         signature = None
         signature_desc = None
         target: Optional[Target] = None
@@ -846,6 +853,10 @@ def main(cls, args: Namespace) -> int:
             LOG.debug("starting sapphire server")
             # launch HTTP server used to serve test cases
             with Sapphire(auto_close=1, timeout=timeout, certs=certs) as server:
+                if certs is not None:
+                    LOG.debug("starting additional web services")
+                    ext_services = WebServices.start_services(certs.host, certs.key)
+
                 target.reverse(server.port, server.port)
                 with ReduceManager(
                     set(args.ignore),
@@ -868,6 +879,7 @@ def main(cls, args: Namespace) -> int:
                     tool=args.tool,
                     use_analysis=not args.no_analysis,
                     use_harness=not args.no_harness,
+                    services=ext_services,
                 ) as mgr:
                     return_code = mgr.run(
                         repeat=args.repeat,
@@ -910,4 +922,6 @@ def main(cls, args: Namespace) -> int:
                 asset_mgr.cleanup()
             if certs is not None:
                 certs.cleanup()
+            if ext_services is not None:
+                ext_services.cleanup()
             LOG.info("Done.")

grizzly/replay/replay.py

@@ -33,6 +33,7 @@
     package_version,
     time_limits,
 )
+from ..services import WebServices
 from ..target import (
     AssetManager,
     Result,
@@ -294,6 +295,7 @@ def run(
         launch_attempts: int = 3,
         on_iteration_cb: Optional[Callable[[], None]] = None,
         post_launch_delay: int = -1,
+        services=None,
     ) -> List[ReplayResult]:
         """Run testcase replay.
 
@@ -349,6 +351,9 @@ def harness_fn(_: str) -> bytes:  # pragma: no cover
             )
             server_map.set_redirect("grz_start", "grz_harness", required=False)
 
+        if services:
+            services.map_locations(server_map)
+
         # track unprocessed results
         reports: Dict[str, ReplayResult] = {}
         try:
@@ -629,6 +634,7 @@ def main(cls, args: Namespace) -> int:
         certs = None
         results: Optional[List[ReplayResult]] = None
         target: Optional[Target] = None
+        ext_services = None
         try:
             # check if hangs are expected
             expect_hang = cls.expect_hang(args.ignore, signature, testcases)
@@ -682,6 +688,10 @@ def main(cls, args: Namespace) -> int:
             LOG.debug("starting sapphire server")
             # launch HTTP server used to serve test cases
             with Sapphire(auto_close=1, timeout=timeout, certs=certs) as server:
+                if certs is not None:
+                    LOG.debug("starting additional web services")
+                    ext_services = WebServices.start_services(certs.host, certs.key)
+
                 target.reverse(server.port, server.port)
                 with cls(
                     set(args.ignore),
@@ -702,6 +712,7 @@ def main(cls, args: Namespace) -> int:
                         min_results=args.min_crashes,
                         post_launch_delay=args.post_launch_delay,
                         repeat=repeat,
+                        services=ext_services,
                     )
             # handle results
             success = any(x.expected for x in results)
@@ -754,4 +765,6 @@ def main(cls, args: Namespace) -> int:
                 asset_mgr.cleanup()
             if certs is not None:
                 certs.cleanup()
+            if ext_services is not None:
+                ext_services.cleanup()
             LOG.info("Done.")

grizzly/replay/test_main.py

@@ -36,6 +36,7 @@ def test_main_01(mocker, server, tmp_path):
     # Of the four attempts only the first and third will 'reproduce' the result
     # and the forth attempt should be skipped.
     mocker.patch("grizzly.common.runner.sleep", autospec=True)
+    mocker.patch("grizzly.replay.replay.WebServices", autospec=True)
     server.serve_path.return_value = (Served.ALL, {"test.html": "/fake/path"})
     # setup Target
     load_target = mocker.patch("grizzly.replay.replay.load_plugin", autospec=True)
@@ -151,6 +152,7 @@ def test_main_02(mocker, server, tmp_path, repro_results):
         test_index=[],
         time_limit=10,
         timeout=None,
+        use_https=False,
         valgrind=False,
     )
     assert ReplayManager.main(args) == Exit.FAILURE
@@ -215,6 +217,7 @@ def test_main_03(mocker, load_plugin, load_testcases, signature, result):
         test_index=[],
         time_limit=10,
         timeout=None,
+        use_https=False,
         valgrind=False,
     )
     asset_mgr = load_testcases[1] if isinstance(load_testcases, tuple) else None
@@ -256,6 +259,7 @@ def test_main_04(mocker, tmp_path):
         test_index=[],
         time_limit=10,
         timeout=None,
+        use_https=False,
         valgrind=False,
     )
     # target launch error
@@ -317,6 +321,7 @@ def test_main_05(mocker, server, tmp_path):
         test_index=[],
         time_limit=1,
         timeout=None,
+        use_https=False,
         valgrind=False,
     )
     # build a test case
@@ -385,6 +390,7 @@ def test_main_06(
         test_index=[],
         time_limit=10,
         timeout=None,
+        use_https=False,
         valgrind=valgrind,
     )
     # maximum one debugger allowed at a time
@@ -438,6 +444,7 @@ def test_main_07(mocker, server, tmp_path):
         time_limit=10,
         timeout=None,
         tool=None,
+        use_https=False,
         valgrind=False,
     )
     assert ReplayManager.main(args) == Exit.SUCCESS

grizzly/services/__init__.py

@@ -0,0 +1,11 @@
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+__all__ = (
+    "ServiceName",
+    "WebServices",
+    "WebTransportServer",
+)
+
+from .core import ServiceName, WebServices
+from .webtransport.core import WebTransportServer

grizzly/services/base.py

@@ -0,0 +1,30 @@
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at https://mozilla.org/MPL/2.0/.
+from abc import ABC, abstractmethod
+
+
+class BaseService(ABC):
+    """Base service class"""
+
+    @property
+    @abstractmethod
+    def location(self):
+        """Location to use with Sapphire.set_dynamic_response"""
+
+    @property
+    @abstractmethod
+    def port(self):
+        """The port on which the server is listening"""
+
+    @abstractmethod
+    def url(self, _query):
+        """Returns the URL of the server."""
+
+    @abstractmethod
+    async def is_ready(self):
+        """Wait until the service is ready"""
+
+    @abstractmethod
+    def cleanup(self):
+        """Stop the server."""

grizzly/services/core.py

@@ -0,0 +1,96 @@
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+import asyncio
+from enum import Enum
+from logging import getLogger
+from typing import Dict
+
+from sapphire import create_listening_socket
+
+from .base import BaseService
+from .webtransport.core import WebTransportServer
+
+LOG = getLogger(__name__)
+
+
+class ServiceName(Enum):
+    """Enum for listing available services"""
+
+    WEB_TRANSPORT = 1
+
+
+class WebServices:
+    """Class for running additional web services"""
+
+    def __init__(self, services: Dict[ServiceName, BaseService]):
+        """Initialize new WebServices instance
+
+        Args:
+            services (dict of ServiceName: BaseService): Collection of services.
+        """
+        self.services = services
+
+    @staticmethod
+    def get_free_port():
+        """Returns an open port"""
+        sock = create_listening_socket()
+        port = sock.getsockname()[1]
+        sock.close()
+
+        return port
+
+    async def is_running(self, timeout=20):
+        """Polls all available services to ensure they are running and accessible.
+
+        Args:
+            timeout (int): Total time to wait.
+
+        Returns:
+            bool: Indicates if all services started successfully.
+        """
+        tasks = {}
+        for name, service in self.services.items():
+            task = asyncio.create_task(service.is_ready())
+            tasks[name] = task
+
+        try:
+            await asyncio.wait_for(asyncio.gather(*tasks.values()), timeout)
+        except asyncio.TimeoutError:
+            for name, task in tasks.items():
+                if not task.done():
+                    LOG.warning("Failed to start service (%s)", ServiceName(name).name)
+            return False
+
+        return True
+
+    def cleanup(self):
+        """Stops all running services and join's the service thread"""
+        for service in self.services.values():
+            service.cleanup()
+
+    def map_locations(self, server_map):
+        """Configure server map"""
+        for service in self.services.values():
+            server_map.set_dynamic_response(
+                service.location, service.url, mime_type="text/plain", required=False
+            )
+
+    @classmethod
+    def start_services(cls, cert, key):
+        """Start all available services
+
+        Args:
+            cert (Path): Path to the certificate file
+            key (Path): Path to the certificate's private key
+        """
+        services = {}
+        # Start WebTransport service
+        wt_port = cls.get_free_port()
+        services[ServiceName.WEB_TRANSPORT] = WebTransportServer(wt_port, cert, key)
+        services[ServiceName.WEB_TRANSPORT].start()
+
+        ext_services = cls(services)
+        assert asyncio.run(ext_services.is_running())
+
+        return ext_services