diff --git a/docs/energy_local_control.md b/docs/energy_local_control.md index 08fca15..ffbcdf9 100644 --- a/docs/energy_local_control.md +++ b/docs/energy_local_control.md @@ -26,6 +26,15 @@ inherits it - `TeslaFleetApi`, `Teslemetry`, `Tessie`) loads an existing RSA private key or creates a new 4096-bit unencrypted PEM key file. This is the key you will register with the gateway and later hand to `aiopowerwall`. +Creating a new key uses a plain `sys.executable -c` subprocess so RSA generation +does not block the asyncio event loop. The subprocess runs its own script as +`__main__` and never imports the caller's entry point, so this works from +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. + ## 2. Register the key with the gateway, over the cloud `EnergySite.add_authorized_client` registers the public half of that key with diff --git a/tesla_fleet_api/tesla/tesla.py b/tesla_fleet_api/tesla/tesla.py index 1c2f8c3..cc6f7eb 100644 --- a/tesla_fleet_api/tesla/tesla.py +++ b/tesla_fleet_api/tesla/tesla.py @@ -2,11 +2,14 @@ import base64 import asyncio +from contextlib import suppress import os +import sys import time from os.path import exists import aiofiles +from tesla_fleet_api.const import LOGGER from tesla_fleet_api.tesla.charging import Charging from tesla_fleet_api.tesla.energysite import EnergySites from tesla_fleet_api.tesla.partner import Partner @@ -22,6 +25,146 @@ _KEY_READ_RETRY_INTERVAL = 0.05 +def _generate_rsa_private_key_pem(key_size: int) -> bytes: + """Generate an RSA key and serialize it. + + cryptography's RSA keygen holds the GIL for its full duration, so a + thread (unlike a separate process) would still stall the caller's event + loop; used both as the isolated subprocess's script body and as the + in-process fallback. + """ + key = rsa.generate_private_key( + public_exponent=65537, + key_size=key_size, + backend=default_backend(), + ) + return key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.TraditionalOpenSSL, + encryption_algorithm=serialization.NoEncryption(), + ) + + +_RSA_KEYGEN_SUBPROCESS_SCRIPT = """ +import sys +from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import rsa + +key = rsa.generate_private_key( + public_exponent=65537, key_size=int(sys.argv[1]), backend=default_backend() +) +sys.stdout.buffer.write( + key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.TraditionalOpenSSL, + encryption_algorithm=serialization.NoEncryption(), + ) +) +""" + + +async def _generate_rsa_private_key_pem_isolated(key_size: int) -> bytes: + """Generate an RSA key's PEM in a plain, short-lived `sys.executable -c ...` subprocess. + + Unlike `multiprocessing`, a plain subprocess's `-c` script is its own + `__main__` - it never re-imports the caller's actual entry point, so + there is nothing to guard, nothing to pickle, no semaphore to create, and + no daemonic-process restriction. `asyncio.create_subprocess_exec` is + awaited natively, off the loop by construction, including cancellation: + killing and awaiting the child's exit are both async, so cancelling the + caller can't block the loop either. Raises on any exec, non-zero-exit, or + empty-output failure, so the caller can fall back to in-process + generation. + + In a frozen bundle (PyInstaller, cx_Freeze, py2exe), `sys.executable` is + the application itself, not a Python interpreter - `-c` would relaunch + the whole application rather than run this script, which can exit 0 + without ever emitting a PEM. `sys.frozen` is the de facto marker these + freezers all set, so that case is refused up front instead of relying on + the empty-output check alone to catch it after the fact. + """ + if getattr(sys, "frozen", False): + raise RuntimeError( + "sys.executable is a frozen application bundle, not a Python " + "interpreter; it cannot run the RSA keygen script" + ) + proc = await asyncio.create_subprocess_exec( + sys.executable, + "-c", + _RSA_KEYGEN_SUBPROCESS_SCRIPT, + str(key_size), + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + try: + stdout, stderr = await proc.communicate() + except asyncio.CancelledError: + with suppress(ProcessLookupError): + proc.kill() + await proc.wait() + raise + if proc.returncode != 0: + raise RuntimeError( + f"RSA key generation subprocess exited with code {proc.returncode}: " + f"{stderr.decode(errors='replace').strip()}" + ) + if not stdout: + raise RuntimeError( + "RSA key generation subprocess exited successfully but produced no output" + ) + return stdout + + +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. + """ + value = await asyncio.to_thread( + serialization.load_pem_private_key, + pem, + password=None, + backend=default_backend(), + ) + if not isinstance(value, rsa.RSAPrivateKey): + raise AssertionError("Generated key is not an RSAPrivateKey") + return value + + +async def _generate_rsa_private_key(key_size: int) -> tuple[rsa.RSAPrivateKey, bytes]: + """Generate an RSA key and return it with its PEM. + + Prefers a short-lived subprocess so keygen - which holds the GIL for its + full duration, unlike a thread - never stalls the caller's event loop. + Falls back to in-process generation, which does block the loop, whenever + that subprocess can't be used - the same environments this method could + already generate keys in before process isolation was introduced. The + fallback catches any exception from launching, running, or deserializing + the subprocess's output, deliberately not enumerated by type, since + process isolation is best-effort here and any way it can fail - including + producing a PEM that turns out not to deserialize - should degrade to the + working (if blocking) legacy path rather than propagate or return bad key + material. + """ + try: + pem = await _generate_rsa_private_key_pem_isolated(key_size) + value = await _deserialize_rsa_pem(pem) + except Exception as err: + LOGGER.warning( + "RSA key generation could not use an isolated subprocess " + "(%s: %s); falling back to in-process generation, which will " + "block the event loop for the duration of key generation.", + type(err).__name__, + err, + ) + pem = await asyncio.to_thread(_generate_rsa_private_key_pem, key_size) + value = await _deserialize_rsa_pem(pem) + return value, pem + + def _owner_only_opener(file: str, flags: int) -> int: """Open a new file exclusively, born at mode 0o600 with no chmod window.""" fd = os.open(file, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) @@ -153,16 +296,8 @@ async def get_rsa_private_key( the create race, its file is read instead of raising. """ if not exists(path): - self.rsa_private_key = rsa.generate_private_key( - public_exponent=65537, - key_size=key_size, - backend=default_backend(), - ) - pem = self.rsa_private_key.private_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.TraditionalOpenSSL, - encryption_algorithm=serialization.NoEncryption(), - ) + value, pem = await _generate_rsa_private_key(key_size) + self.rsa_private_key = value try: async with aiofiles.open( path, "wb", opener=_owner_only_opener diff --git a/tests/test_tesla_private_key.py b/tests/test_tesla_private_key.py index d9b25de..b528f77 100644 --- a/tests/test_tesla_private_key.py +++ b/tests/test_tesla_private_key.py @@ -10,8 +10,8 @@ from __future__ import annotations import asyncio -import stat import os +import stat import tempfile from pathlib import Path from unittest import IsolatedAsyncioTestCase, mock @@ -176,6 +176,11 @@ async def test_new_key_file_is_owner_only(self) -> None: self.assertEqual(mode, 0o600) async def test_new_key_file_is_owner_only_with_restrictive_umask(self) -> None: + # Unlike multiprocessing's semaphore file, a plain subprocess has no + # umask-sensitive IPC of its own, so this exercises the isolated + # subprocess path directly (no fallback needed) and still produces a + # correct owner-only key file via _owner_only_opener's explicit + # fchmod, which bypasses umask entirely. with tempfile.TemporaryDirectory() as tmp_dir: path = str(Path(tmp_dir) / "tedapi_rsa_private.pem") tesla = Tesla() @@ -258,3 +263,326 @@ async def finish_write() -> None: await writer self.assertEqual(_rsa_pem(key), _rsa_pem(winner_key)) + + async def test_generation_does_not_block_the_event_loop(self) -> None: + """A concurrent heartbeat must keep ticking during real RSA generation. + + Generation runs in a separate worker process (a plain + ``asyncio.to_thread`` would not do - cryptography's RSA keygen holds + the GIL for its full duration, so a thread still stalls the caller's + event loop), which is why this uses a real key rather than a mocked + stand-in: only genuine process isolation proves the loop stays free. + """ + with tempfile.TemporaryDirectory() as tmp_dir: + path = str(Path(tmp_dir) / "tedapi_rsa_private.pem") + + ticks = 0 + + async def heartbeat() -> None: + nonlocal ticks + while True: + await asyncio.sleep(0.01) + ticks += 1 + + heart = asyncio.create_task(heartbeat()) + try: + key = await Tesla().get_rsa_private_key(path, key_size=4096) + finally: + heart.cancel() + + self.assertGreater(ticks, 5) + self.assertEqual(_rsa_pem(key), Path(path).read_bytes()) + + async def test_deserialization_does_not_block_the_event_loop(self) -> None: + """The post-generation PEM deserialization also must not stall the loop. + + It runs in the main process (via ``asyncio.to_thread``), so unlike + generation it can be exercised with a mocked slow stand-in. + """ + 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) + + 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: + await Tesla().get_rsa_private_key(path, key_size=1024) + finally: + heart.cancel() + + self.assertGreater(ticks, 10) + + async def test_cancellation_kills_the_subprocess(self) -> None: + """Cancelling mid-generation kills the child rather than leaving it orphaned. + + Killing and awaiting the child's exit are both async operations, so + this also can't block the event loop the way a synchronous + `ProcessPoolExecutor.shutdown(wait=True)` could. + """ + with tempfile.TemporaryDirectory() as tmp_dir: + path = str(Path(tmp_dir) / "tedapi_rsa_private.pem") + generation_started = asyncio.Event() + killed = asyncio.Event() + + class _SlowProc: + returncode: int | None = None + + async def communicate(self) -> tuple[bytes, bytes]: + generation_started.set() + await asyncio.Event().wait() # never resolves on its own + raise AssertionError("unreachable") + + def kill(self) -> None: + killed.set() + + async def wait(self) -> int: + await killed.wait() + self.returncode = -9 + return self.returncode + + async def fake_create_subprocess_exec( + *args: object, **kwargs: object + ) -> _SlowProc: + return _SlowProc() + + with mock.patch( + "tesla_fleet_api.tesla.tesla.asyncio.create_subprocess_exec", + side_effect=fake_create_subprocess_exec, + ): + task = asyncio.create_task( + Tesla().get_rsa_private_key(path, key_size=1024) + ) + await generation_started.wait() + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await task + + self.assertTrue(killed.is_set()) + + async def test_cancellation_preserves_cancel_when_subprocess_already_exited( + self, + ) -> None: + generation_started = asyncio.Event() + waited = asyncio.Event() + + class _ExitedProc: + returncode = 0 + + async def communicate(self) -> tuple[bytes, bytes]: + generation_started.set() + await asyncio.Event().wait() + raise AssertionError("unreachable") + + def kill(self) -> None: + raise ProcessLookupError + + async def wait(self) -> int: + waited.set() + return self.returncode + + async def fake_create_subprocess_exec( + *args: object, **kwargs: object + ) -> _ExitedProc: + return _ExitedProc() + + with mock.patch( + "tesla_fleet_api.tesla.tesla.asyncio.create_subprocess_exec", + side_effect=fake_create_subprocess_exec, + ): + task = asyncio.create_task( + Tesla().get_rsa_private_key("unused.pem", key_size=1024) + ) + await generation_started.wait() + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await task + + self.assertTrue(waited.is_set()) + + async def test_subprocess_launch_failure_falls_back_to_in_process_generation( + self, + ) -> None: + """A caller whose environment can't launch the subprocess still succeeds. + + Falls back to in-process generation with a logged warning instead of + raising, matching the pre-process-isolation behavior. The fallback + deliberately isn't specific to any one launch-failure cause (a + restrictive umask, a REPL/notebook caller, a sandboxed environment + without `sys.executable`, ...) - any of them lands here the same way. + """ + 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", + side_effect=OSError("boom"), + ), + self.assertLogs("tesla_fleet_api", level="WARNING") as logs, + ): + key = await Tesla().get_rsa_private_key(path, key_size=1024) + + self.assertIsInstance(key, rsa.RSAPrivateKey) + mode = stat.S_IMODE(Path(path).stat().st_mode) + self.assertEqual(mode, 0o600) + self.assertTrue( + any("isolated subprocess" in message for message in logs.output) + ) + + async def test_subprocess_nonzero_exit_falls_back_to_in_process_generation( + self, + ) -> None: + """A subprocess that launches but exits non-zero also falls back.""" + with tempfile.TemporaryDirectory() as tmp_dir: + path = str(Path(tmp_dir) / "tedapi_rsa_private.pem") + + class _FailingProc: + returncode = 1 + + async def communicate(self) -> tuple[bytes, bytes]: + return b"", b"boom" + + async def fake_create_subprocess_exec( + *args: object, **kwargs: object + ) -> _FailingProc: + return _FailingProc() + + with ( + mock.patch( + "tesla_fleet_api.tesla.tesla.asyncio.create_subprocess_exec", + side_effect=fake_create_subprocess_exec, + ), + self.assertLogs("tesla_fleet_api", level="WARNING") as logs, + ): + key = await Tesla().get_rsa_private_key(path, key_size=1024) + + self.assertIsInstance(key, rsa.RSAPrivateKey) + mode = stat.S_IMODE(Path(path).stat().st_mode) + self.assertEqual(mode, 0o600) + self.assertTrue( + any("isolated subprocess" in message for message in logs.output) + ) + + async def test_frozen_bundle_falls_back_to_in_process_generation(self) -> None: + """In a PyInstaller/cx_Freeze/py2exe bundle, `sys.executable` is the app itself. + + `-c` would relaunch the whole application rather than run the keygen + script, so isolation is skipped up front (via `sys.frozen`, the de + facto marker these freezers all set) rather than relying on it to + exit non-zero or produce no output. + """ + with tempfile.TemporaryDirectory() as tmp_dir: + path = str(Path(tmp_dir) / "tedapi_rsa_private.pem") + + with ( + mock.patch("tesla_fleet_api.tesla.tesla.sys.frozen", True, create=True), + mock.patch( + "tesla_fleet_api.tesla.tesla.asyncio.create_subprocess_exec" + ) as create_subprocess_exec, + self.assertLogs("tesla_fleet_api", level="WARNING") as logs, + ): + key = await Tesla().get_rsa_private_key(path, key_size=1024) + + create_subprocess_exec.assert_not_called() + self.assertIsInstance(key, rsa.RSAPrivateKey) + mode = stat.S_IMODE(Path(path).stat().st_mode) + self.assertEqual(mode, 0o600) + self.assertTrue( + any("isolated subprocess" in message for message in logs.output) + ) + + async def test_empty_subprocess_output_falls_back_to_in_process_generation( + self, + ) -> None: + """A subprocess that exits 0 but writes no PEM is treated as a failure. + + Covers a wrong child (e.g. one that silently did nothing useful) + succeeding without ever emitting a key, which must not be trusted as + if it had. + """ + with tempfile.TemporaryDirectory() as tmp_dir: + path = str(Path(tmp_dir) / "tedapi_rsa_private.pem") + + class _EmptyOutputProc: + returncode = 0 + + async def communicate(self) -> tuple[bytes, bytes]: + return b"", b"" + + async def fake_create_subprocess_exec( + *args: object, **kwargs: object + ) -> _EmptyOutputProc: + return _EmptyOutputProc() + + with ( + mock.patch( + "tesla_fleet_api.tesla.tesla.asyncio.create_subprocess_exec", + side_effect=fake_create_subprocess_exec, + ), + self.assertLogs("tesla_fleet_api", level="WARNING") as logs, + ): + key = await Tesla().get_rsa_private_key(path, key_size=1024) + + self.assertIsInstance(key, rsa.RSAPrivateKey) + mode = stat.S_IMODE(Path(path).stat().st_mode) + self.assertEqual(mode, 0o600) + self.assertTrue( + any("isolated subprocess" in message for message in logs.output) + ) + + async def test_invalid_subprocess_pem_falls_back_to_in_process_generation( + self, + ) -> None: + """A subprocess that exits 0 with non-empty but malformed output also falls back. + + Covers a wrong-but-zero-exit child whose output isn't a usable PEM at + all, which must not be trusted as if it had produced one. + """ + with tempfile.TemporaryDirectory() as tmp_dir: + path = str(Path(tmp_dir) / "tedapi_rsa_private.pem") + + class _GarbageOutputProc: + returncode = 0 + + async def communicate(self) -> tuple[bytes, bytes]: + return b"not a pem", b"" + + async def fake_create_subprocess_exec( + *args: object, **kwargs: object + ) -> _GarbageOutputProc: + return _GarbageOutputProc() + + with ( + mock.patch( + "tesla_fleet_api.tesla.tesla.asyncio.create_subprocess_exec", + side_effect=fake_create_subprocess_exec, + ), + self.assertLogs("tesla_fleet_api", level="WARNING") as logs, + ): + key = await Tesla().get_rsa_private_key(path, key_size=1024) + + self.assertIsInstance(key, rsa.RSAPrivateKey) + mode = stat.S_IMODE(Path(path).stat().st_mode) + self.assertEqual(mode, 0o600) + self.assertTrue( + any("isolated subprocess" in message for message in logs.output) + )