fix(tesla): make PEM key handling non-blocking and configurable

docs/energy_local_control.md

@@ -33,7 +33,17 @@ scripts, REPLs, `python -c`, and notebooks without an entry-point guard. If the
 subprocess cannot be used (including in a frozen application), exits
 unsuccessfully, or returns empty or invalid key data, generation falls back to
 the current process and logs a warning that the asyncio event loop may be
-blocked. Loading an existing key file does not start a subprocess.
+blocked. Loading and validating an existing key runs in a worker thread and
+does not start a subprocess.
+
+By default, existing keys receive cryptography's full RSA consistency check.
+For an existing key that your application already trusts, pass
+`skip_rsa_key_validation=True` to skip that potentially expensive check. This
+option does not weaken validation when loading the key's PEM structure. Pass
+`async_rsa_key_creation=False` only when subprocess spawning is unavailable or
+undesired: new-key generation then runs in the current process and blocks the
+event loop. Both options affect only RSA key handling; their defaults preserve
+the non-blocking and fully validated behavior described above.
 
 ## 2. Register the key with the gateway, over the cloud
 

tesla_fleet_api/tesla/tesla.py

@@ -119,15 +119,19 @@ async def _generate_rsa_private_key_pem_isolated(key_size: int) -> bytes:
 async def _deserialize_rsa_pem(pem: bytes) -> rsa.RSAPrivateKey:
     """Deserialize a freshly generated RSA PEM off the event loop.
 
-    A malformed PEM (e.g. from a wrong-but-zero-exit isolated subprocess)
-    raises `ValueError` here rather than being trusted - the caller treats
-    that the same as any other isolation failure and falls back.
+    Skips cryptography's RSA consistency check (`unsafe_skip_rsa_key_validation`)
+    since this PEM was just generated by us and can't be mathematically
+    malformed - only structurally invalid (e.g. a wrong-but-zero-exit
+    isolated subprocess), which still raises `ValueError` here since that
+    check is independent of PEM parsing itself; the caller treats that the
+    same as any other isolation failure and falls back.
     """
     value = await asyncio.to_thread(
         serialization.load_pem_private_key,
         pem,
         password=None,
         backend=default_backend(),
+        unsafe_skip_rsa_key_validation=True,
     )
     if not isinstance(value, rsa.RSAPrivateKey):
         raise AssertionError("Generated key is not an RSAPrivateKey")
@@ -180,14 +184,29 @@ def _owner_only_opener(file: str, flags: int) -> int:
     return fd
 
 
-async def _load_pem_private_key(path: str, retry_invalid: bool = False) -> object:
+async def _load_pem_private_key(
+    path: str,
+    retry_invalid: bool = False,
+    unsafe_skip_rsa_key_validation: bool = False,
+) -> object:
+    """Read and deserialize a PEM key file off the event loop.
+
+    Deserialization (in particular RSA's consistency check) runs in a thread
+    via `asyncio.to_thread` since it can take ~200ms for a 4096-bit key.
+    `unsafe_skip_rsa_key_validation` is forwarded to cryptography to skip
+    that check for a caller-trusted RSA PEM; it's a no-op for EC keys.
+    """
     deadline = time.monotonic() + _KEY_READ_RETRY_TIMEOUT
     while True:
         async with aiofiles.open(path, "rb") as key_file:
             key_data = await key_file.read()
         try:
-            return serialization.load_pem_private_key(
-                key_data, password=None, backend=default_backend()
+            return await asyncio.to_thread(
+                serialization.load_pem_private_key,
+                key_data,
+                password=None,
+                backend=default_backend(),
+                unsafe_skip_rsa_key_validation=unsafe_skip_rsa_key_validation,
             )
         except ValueError:
             if not retry_invalid or time.monotonic() >= deadline:
@@ -285,7 +304,11 @@ def public_uncompressed_point(self) -> str:
         )
 
     async def get_rsa_private_key(
-        self, path: str = "tedapi_rsa_private.pem", key_size: int = 4096
+        self,
+        path: str = "tedapi_rsa_private.pem",
+        key_size: int = 4096,
+        skip_rsa_key_validation: bool = False,
+        async_rsa_key_creation: bool = True,
     ) -> rsa.RSAPrivateKey:
         """Get or create an RSA private key for energy gateway client registration.
 
@@ -294,9 +317,21 @@ async def get_rsa_private_key(
         PEM file. A newly created key file is opened with O_EXCL so it is born
         at mode 0o600 with no world-readable window. If another process wins
         the create race, its file is read instead of raising.
+
+        `skip_rsa_key_validation` skips cryptography's RSA consistency check
+        when reading a PEM from disk - safe only for a key you trust wasn't
+        tampered with or corrupted, since it bypasses the check that would
+        otherwise catch that. `async_rsa_key_creation` (default `True`,
+        matching prior behavior) generates a new key off the event loop via
+        an isolated subprocess with an in-process fallback; set `False` to
+        force the simpler, blocking in-process generator instead.
         """
         if not exists(path):
-            value, pem = await _generate_rsa_private_key(key_size)
+            if async_rsa_key_creation:
+                value, pem = await _generate_rsa_private_key(key_size)
+            else:
+                pem = _generate_rsa_private_key_pem(key_size)
+                value = await _deserialize_rsa_pem(pem)
             self.rsa_private_key = value
             try:
                 async with aiofiles.open(
@@ -305,13 +340,21 @@ async def get_rsa_private_key(
                     await key_file.write(pem)
                 return self.rsa_private_key
             except FileExistsError:
-                value = await _load_pem_private_key(path, retry_invalid=True)
+                value = await _load_pem_private_key(
+                    path,
+                    retry_invalid=True,
+                    unsafe_skip_rsa_key_validation=skip_rsa_key_validation,
+                )
                 if not isinstance(value, rsa.RSAPrivateKey):
                     raise AssertionError("Loaded key is not an RSAPrivateKey")
                 self.rsa_private_key = value
                 return self.rsa_private_key
 
-        value = await _load_pem_private_key(path, retry_invalid=True)
+        value = await _load_pem_private_key(
+            path,
+            retry_invalid=True,
+            unsafe_skip_rsa_key_validation=skip_rsa_key_validation,
+        )
         if not isinstance(value, rsa.RSAPrivateKey):
             raise AssertionError("Loaded key is not an RSAPrivateKey")
         self.rsa_private_key = value

tests/test_tesla_private_key.py

@@ -586,3 +586,132 @@ async def fake_create_subprocess_exec(
             self.assertTrue(
                 any("isolated subprocess" in message for message in logs.output)
             )
+
+
+class GetRsaPrivateKeyOptionalValidationTests(IsolatedAsyncioTestCase):
+    async def test_existing_key_read_validates_by_default(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            path = str(Path(tmp_dir) / "tedapi_rsa_private.pem")
+            await Tesla().get_rsa_private_key(path, key_size=1024)
+
+            with mock.patch(
+                "tesla_fleet_api.tesla.tesla.serialization.load_pem_private_key",
+                wraps=serialization.load_pem_private_key,
+            ) as load:
+                await Tesla().get_rsa_private_key(path, key_size=1024)
+
+            load.assert_called_once()
+            self.assertFalse(load.call_args.kwargs["unsafe_skip_rsa_key_validation"])
+
+    async def test_skip_rsa_key_validation_forwarded_on_existing_key_read(
+        self,
+    ) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            path = str(Path(tmp_dir) / "tedapi_rsa_private.pem")
+            created = await Tesla().get_rsa_private_key(path, key_size=1024)
+
+            with mock.patch(
+                "tesla_fleet_api.tesla.tesla.serialization.load_pem_private_key",
+                wraps=serialization.load_pem_private_key,
+            ) as load:
+                key = await Tesla().get_rsa_private_key(
+                    path, key_size=1024, skip_rsa_key_validation=True
+                )
+
+            load.assert_called_once()
+            self.assertTrue(load.call_args.kwargs["unsafe_skip_rsa_key_validation"])
+            self.assertEqual(_rsa_pem(key), _rsa_pem(created))
+
+    async def test_internally_generated_pem_deserialization_skips_validation(
+        self,
+    ) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            path = str(Path(tmp_dir) / "tedapi_rsa_private.pem")
+
+            with mock.patch(
+                "tesla_fleet_api.tesla.tesla.serialization.load_pem_private_key",
+                wraps=serialization.load_pem_private_key,
+            ) as load:
+                await Tesla().get_rsa_private_key(path, key_size=1024)
+
+            load.assert_called_once()
+            self.assertTrue(load.call_args.kwargs["unsafe_skip_rsa_key_validation"])
+
+    async def test_async_rsa_key_creation_true_uses_isolated_subprocess(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            path = str(Path(tmp_dir) / "tedapi_rsa_private.pem")
+
+            with mock.patch(
+                "tesla_fleet_api.tesla.tesla.asyncio.create_subprocess_exec",
+                wraps=asyncio.create_subprocess_exec,
+            ) as create_subprocess_exec:
+                key = await Tesla().get_rsa_private_key(
+                    path, key_size=1024, async_rsa_key_creation=True
+                )
+
+            create_subprocess_exec.assert_called_once()
+            self.assertIsInstance(key, rsa.RSAPrivateKey)
+
+    async def test_async_rsa_key_creation_false_skips_isolated_subprocess(
+        self,
+    ) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            path = str(Path(tmp_dir) / "tedapi_rsa_private.pem")
+
+            with mock.patch(
+                "tesla_fleet_api.tesla.tesla.asyncio.create_subprocess_exec",
+            ) as create_subprocess_exec:
+                key = await Tesla().get_rsa_private_key(
+                    path, key_size=1024, async_rsa_key_creation=False
+                )
+
+            create_subprocess_exec.assert_not_called()
+            self.assertIsInstance(key, rsa.RSAPrivateKey)
+            mode = stat.S_IMODE(Path(path).stat().st_mode)
+            self.assertEqual(mode, 0o600)
+
+    async def test_existing_key_read_does_not_block_event_loop(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            path = str(Path(tmp_dir) / "tedapi_rsa_private.pem")
+            real_key = rsa.generate_private_key(public_exponent=65537, key_size=1024)
+            Path(path).write_bytes(_rsa_pem(real_key))
+            os.chmod(path, 0o600)
+
+            def slow_load_pem_private_key(
+                *args: object, **kwargs: object
+            ) -> rsa.RSAPrivateKey:
+                import time as time_module
+
+                time_module.sleep(0.15)
+                return real_key
+
+            ticks = 0
+
+            async def heartbeat() -> None:
+                nonlocal ticks
+                while True:
+                    await asyncio.sleep(0.01)
+                    ticks += 1
+
+            with mock.patch(
+                "tesla_fleet_api.tesla.tesla.serialization.load_pem_private_key",
+                side_effect=slow_load_pem_private_key,
+            ):
+                heart = asyncio.create_task(heartbeat())
+                try:
+                    key = await Tesla().get_rsa_private_key(path, key_size=1024)
+                finally:
+                    heart.cancel()
+
+            self.assertGreater(ticks, 10)
+            self.assertEqual(_rsa_pem(key), _rsa_pem(real_key))
+
+    async def test_defaults_unchanged_for_existing_rsa_key_read(self) -> None:
+        """No new params behaves identically to before this change."""
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            path = str(Path(tmp_dir) / "tedapi_rsa_private.pem")
+            created = await Tesla().get_rsa_private_key(path, key_size=1024)
+
+            read_back = await Tesla().get_rsa_private_key(path, key_size=1024)
+
+            self.assertEqual(_rsa_pem(read_back), _rsa_pem(created))