feat(tesla): raise PrivateKeyError for unusable existing key files

Home Assistant has to catch OSError/ValueError/TypeError/AssertionError at
multiple sites to handle an unreadable, malformed, encrypted, or wrong-type
key file. Give get_private_key/get_rsa_private_key one typed exception with
a reason instead.

Claude-Session: https://claude.ai/code/session_01D7girHWFLWBu78by9NL3qE

Author: Bre77

README.md

@@ -172,10 +172,14 @@ asyncio.run(main())
 For more detailed examples, see [Bluetooth for Vehicles](docs/bluetooth_vehicles.md).
 
 `get_private_key(path)` loads an existing EC private key or creates a new
-unencrypted PEM key file. Newly created key files are created owner-readable
-and owner-writable only (`0600`) from the start, with no write-then-chmod
-window, and concurrent creators fall back to reading the file that won the
-create race.
+unencrypted PEM key file, and `get_rsa_private_key(path)` does the same for an
+RSA key. Newly created key files are created owner-readable and
+owner-writable only (`0600`) from the start, with no write-then-chmod window,
+and concurrent creators fall back to reading the file that won the create
+race. If an existing key file can't be read, isn't valid PEM, is
+password-encrypted, or is the wrong key type, both raise `PrivateKeyError`
+(a `TeslaFleetError`) with a `reason` of `"unreadable"`, `"malformed"`,
+`"encrypted"`, or `"wrong_type"`.
 
 `VehicleBluetooth` keeps a held BLE connection alive during idle periods by
 default with a passive GATT read about every 20 seconds. Pass

tesla_fleet_api/exceptions.py

@@ -404,6 +404,23 @@ class DeviceUnexpectedResponse(TeslaFleetError):
     status = 540
 
 
+class PrivateKeyError(TeslaFleetError):
+    """An existing private key file could not be loaded as a usable key.
+
+    Raised by ``Tesla.get_private_key``/``get_rsa_private_key`` only for a
+    known-existing key file's read/parse failure - key generation and the
+    O_EXCL create-race fallback keep raising their original exceptions.
+    ``reason`` is one of ``"unreadable"`` (I/O failure), ``"malformed"`` (not
+    valid PEM), ``"encrypted"`` (PEM requires a passphrase), or
+    ``"wrong_type"`` (loaded key is not the expected type).
+    """
+
+    def __init__(self, reason: str, message: str) -> None:
+        self.reason = reason
+        self.message = message
+        super().__init__()
+
+
 class LibraryError(Exception):
     """Errors related to this library."""
 

tesla_fleet_api/tesla/tesla.py

@@ -7,9 +7,11 @@
 import sys
 import time
 from os.path import exists
+from typing import TypeVar
 import aiofiles
 
 from tesla_fleet_api.const import LOGGER
+from tesla_fleet_api.exceptions import PrivateKeyError
 from tesla_fleet_api.tesla.charging import Charging
 from tesla_fleet_api.tesla.energysite import EnergySites
 from tesla_fleet_api.tesla.partner import Partner
@@ -214,6 +216,46 @@ async def _load_pem_private_key(
             await asyncio.sleep(_KEY_READ_RETRY_INTERVAL)
 
 
+_KeyT = TypeVar("_KeyT", ec.EllipticCurvePrivateKey, rsa.RSAPrivateKey)
+
+
+async def _load_existing_private_key(
+    path: str,
+    expected_type: type[_KeyT],
+    unsafe_skip_rsa_key_validation: bool = False,
+) -> _KeyT:
+    """Read and parse a key file already known to exist, raising ``PrivateKeyError`` for every failure shape.
+
+    Only covers a known-existing file's read/parse - key generation and the
+    O_EXCL create-race fallback are separate call sites that keep raising
+    their original exceptions.
+    """
+    try:
+        value = await _load_pem_private_key(
+            path,
+            retry_invalid=True,
+            unsafe_skip_rsa_key_validation=unsafe_skip_rsa_key_validation,
+        )
+    except OSError as err:
+        raise PrivateKeyError(
+            "unreadable", f"Could not read private key file at {path}"
+        ) from err
+    except TypeError as err:
+        raise PrivateKeyError(
+            "encrypted", f"Private key file at {path} is encrypted"
+        ) from err
+    except ValueError as err:
+        raise PrivateKeyError(
+            "malformed", f"Private key file at {path} is not a valid PEM private key"
+        ) from err
+    if not isinstance(value, expected_type):
+        raise PrivateKeyError(
+            "wrong_type",
+            f"Private key file at {path} is not a {expected_type.__name__}",
+        )
+    return value
+
+
 class Tesla:
     """Base class describing interactions with Tesla products."""
 
@@ -258,15 +300,7 @@ async def get_private_key(
                 self.private_key = value
                 return self.private_key
 
-        try:
-            value = await _load_pem_private_key(path, retry_invalid=True)
-        except FileNotFoundError:
-            raise FileNotFoundError(f"Private key file not found at {path}")
-        except PermissionError:
-            raise PermissionError(f"Permission denied when trying to read {path}")
-
-        if not isinstance(value, ec.EllipticCurvePrivateKey):
-            raise AssertionError("Loaded key is not an EllipticCurvePrivateKey")
+        value = await _load_existing_private_key(path, ec.EllipticCurvePrivateKey)
         self.private_key = value
         return self.private_key
 
@@ -350,13 +384,9 @@ async def get_rsa_private_key(
                 self.rsa_private_key = value
                 return self.rsa_private_key
 
-        value = await _load_pem_private_key(
-            path,
-            retry_invalid=True,
-            unsafe_skip_rsa_key_validation=skip_rsa_key_validation,
+        value = await _load_existing_private_key(
+            path, rsa.RSAPrivateKey, 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
 

tests/test_tesla_private_key.py

@@ -19,6 +19,7 @@
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric import ec, rsa
 
+from tesla_fleet_api.exceptions import PrivateKeyError
 from tesla_fleet_api.tesla.tesla import Tesla
 
 
@@ -715,3 +716,121 @@ async def test_defaults_unchanged_for_existing_rsa_key_read(self) -> None:
             read_back = await Tesla().get_rsa_private_key(path, key_size=1024)
 
             self.assertEqual(_rsa_pem(read_back), _rsa_pem(created))
+
+
+class PrivateKeyErrorTests(IsolatedAsyncioTestCase):
+    """An existing-but-unusable key file must raise ``PrivateKeyError`` with the right reason."""
+
+    async def test_ec_loader_unreadable(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            path = str(Path(tmp_dir) / "private_key.pem")
+            os.mkdir(path)
+
+            with self.assertRaises(PrivateKeyError) as ctx:
+                await Tesla().get_private_key(path)
+
+            self.assertEqual(ctx.exception.reason, "unreadable")
+            self.assertIsInstance(ctx.exception.__cause__, OSError)
+            self.assertIn(path, ctx.exception.message)
+
+    async def test_ec_loader_malformed(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            path = str(Path(tmp_dir) / "private_key.pem")
+            Path(path).write_bytes(b"not a pem file")
+
+            with self.assertRaises(PrivateKeyError) as ctx:
+                await Tesla().get_private_key(path)
+
+            self.assertEqual(ctx.exception.reason, "malformed")
+            self.assertIsInstance(ctx.exception.__cause__, ValueError)
+            self.assertIn(path, ctx.exception.message)
+
+    async def test_ec_loader_encrypted(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            path = str(Path(tmp_dir) / "private_key.pem")
+            key = ec.generate_private_key(ec.SECP256R1())
+            pem = key.private_bytes(
+                encoding=serialization.Encoding.PEM,
+                format=serialization.PrivateFormat.TraditionalOpenSSL,
+                encryption_algorithm=serialization.BestAvailableEncryption(
+                    b"correct horse battery staple"
+                ),
+            )
+            Path(path).write_bytes(pem)
+
+            with self.assertRaises(PrivateKeyError) as ctx:
+                await Tesla().get_private_key(path)
+
+            self.assertEqual(ctx.exception.reason, "encrypted")
+            self.assertIsInstance(ctx.exception.__cause__, TypeError)
+            self.assertIn(path, ctx.exception.message)
+
+    async def test_ec_loader_wrong_type(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            path = str(Path(tmp_dir) / "private_key.pem")
+            key = rsa.generate_private_key(public_exponent=65537, key_size=1024)
+            Path(path).write_bytes(_rsa_pem(key))
+
+            with self.assertRaises(PrivateKeyError) as ctx:
+                await Tesla().get_private_key(path)
+
+            self.assertEqual(ctx.exception.reason, "wrong_type")
+            self.assertIsNone(ctx.exception.__cause__)
+            self.assertIn(path, ctx.exception.message)
+
+    async def test_rsa_loader_unreadable(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            path = str(Path(tmp_dir) / "tedapi_rsa_private.pem")
+            os.mkdir(path)
+
+            with self.assertRaises(PrivateKeyError) as ctx:
+                await Tesla().get_rsa_private_key(path, key_size=1024)
+
+            self.assertEqual(ctx.exception.reason, "unreadable")
+            self.assertIsInstance(ctx.exception.__cause__, OSError)
+            self.assertIn(path, ctx.exception.message)
+
+    async def test_rsa_loader_malformed(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            path = str(Path(tmp_dir) / "tedapi_rsa_private.pem")
+            Path(path).write_bytes(b"not a pem file")
+
+            with self.assertRaises(PrivateKeyError) as ctx:
+                await Tesla().get_rsa_private_key(path, key_size=1024)
+
+            self.assertEqual(ctx.exception.reason, "malformed")
+            self.assertIsInstance(ctx.exception.__cause__, ValueError)
+            self.assertIn(path, ctx.exception.message)
+
+    async def test_rsa_loader_encrypted(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            path = str(Path(tmp_dir) / "tedapi_rsa_private.pem")
+            key = rsa.generate_private_key(public_exponent=65537, key_size=1024)
+            pem = key.private_bytes(
+                encoding=serialization.Encoding.PEM,
+                format=serialization.PrivateFormat.TraditionalOpenSSL,
+                encryption_algorithm=serialization.BestAvailableEncryption(
+                    b"correct horse battery staple"
+                ),
+            )
+            Path(path).write_bytes(pem)
+
+            with self.assertRaises(PrivateKeyError) as ctx:
+                await Tesla().get_rsa_private_key(path, key_size=1024)
+
+            self.assertEqual(ctx.exception.reason, "encrypted")
+            self.assertIsInstance(ctx.exception.__cause__, TypeError)
+            self.assertIn(path, ctx.exception.message)
+
+    async def test_rsa_loader_wrong_type(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            path = str(Path(tmp_dir) / "tedapi_rsa_private.pem")
+            key = ec.generate_private_key(ec.SECP256R1())
+            Path(path).write_bytes(_ec_pem(key))
+
+            with self.assertRaises(PrivateKeyError) as ctx:
+                await Tesla().get_rsa_private_key(path, key_size=1024)
+
+            self.assertEqual(ctx.exception.reason, "wrong_type")
+            self.assertIsNone(ctx.exception.__cause__)
+            self.assertIn(path, ctx.exception.message)