feat(security): servername debounce

stapler/cert_manager.py

@@ -1,3 +1,4 @@
+import datetime
 import logging
 import pathlib
 import shutil
@@ -21,13 +22,14 @@ class CertManager:
     __slots__ = [
         "certbot_conf",
         "certbot_www",
-        "last_file_change",
+        "last_servername",
         "logger",
         "self_signed_path",
         "with_certbot",
     ]
 
     SELF_SIGNED_DAYS = 30
+    DEBOUCE_SECONDS = 5
     CRT_FILE = "fullchain.pem"
     KEY_FILE = "privkey.pem"
 
@@ -37,7 +39,7 @@ class CertManager:
         self.certbot_www: pathlib.Path = pathlib.Path(params.certbot_www)
         self.self_signed_path: pathlib.Path = pathlib.Path(params.self_signed_path)
         self.with_certbot: bool = params.with_certbot
-        self.last_file_change: int | float = 0
+        self.last_servername: datetime.datetime = datetime.datetime.now(tz=datetime.UTC)
 
     def init(self) -> None:
         self.logger.debug("Initializing...")
@@ -75,6 +77,12 @@ class CertManager:
             return True
         return created or self.__create_self_signed(host)
 
+    def debounce(self) -> bool:
+        last = self.last_servername
+        now = datetime.datetime.now(tz=datetime.UTC)
+        self.last_servername = now
+        return (now - last).total_seconds() > self.DEBOUCE_SECONDS
+
     def get_cert(self, host: str) -> pathlib.Path:
         if self.__exists_certbot(host):
             return self.__certbot_file(host, self.CRT_FILE)
@@ -211,7 +219,9 @@ class CertManager:
             return None
         self.logger.debug("servername callback: %s", host)
         if not self.exists(host) and (
-            not self.valid_host(host) or not self.create_or_update(host)
+            not self.debounce()
+            or not self.valid_host(host)
+            or not self.create_or_update(host)
         ):
             return None
         cert_file = self.get_cert(host)

tests/test_cert_manager.py

@@ -1,3 +1,4 @@
+import datetime
 import logging
 import ssl
 import subprocess
@@ -27,6 +28,9 @@ class TestRegistry(BaseTestCase):
             )
         )
         self.cert_manager.logger = unittest.mock.Mock(logging.Logger)
+        self.cert_manager.last_servername = datetime.datetime.now(
+            tz=datetime.UTC
+        ) - datetime.timedelta(seconds=self.cert_manager.DEBOUCE_SECONDS)
         self.context_mock = unittest.mock.Mock(ssl.SSLContext)
         self.socket_mock = unittest.mock.Mock(ssl.SSLObject)
         unittest.mock.patch("subprocess.check_output")
@@ -183,6 +187,16 @@ class TestRegistry(BaseTestCase):
                 self.socket_mock, "example.fr", self.context_mock
             )
 
+    def test_servername_callback_fail_debouce(self) -> None:
+        self._make_self_signed("example.com")
+        self.cert_manager.last_servername = datetime.datetime.now(tz=datetime.UTC)
+        with (
+            self.patch("ssl.create_default_context", count=0),
+        ):
+            self.cert_manager.servername_callback(
+                self.socket_mock, "example.fr", self.context_mock
+            )
+
     def test_servername_callback_fail_no_binaries(self) -> None:
         self._make_self_signed("example.com")
         response = requests.Response()