diff --git a/README.md b/README.md index 5ed1b72..358a45f 100644 --- a/README.md +++ b/README.md @@ -172,10 +172,15 @@ 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 `LibraryError`, not a `TeslaFleetError` - it's a local key-file failure, +not an upstream Fleet API error) 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 diff --git a/docs/bluetooth_vehicles.md b/docs/bluetooth_vehicles.md index 5cffcca..612c1cf 100644 --- a/docs/bluetooth_vehicles.md +++ b/docs/bluetooth_vehicles.md @@ -39,11 +39,9 @@ async def main(): asyncio.run(main()) ``` -`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. +See the [README](../README.md#bluetooth-for-vehicles) for +`get_private_key(path)`'s create/load semantics and the `PrivateKeyError` +raised for an unusable existing key file. ## Keeping the Connection Alive (`keepalive_interval`) diff --git a/docs/fleet_api_energy_sites.md b/docs/fleet_api_energy_sites.md index 53ea27e..01f2437 100644 --- a/docs/fleet_api_energy_sites.md +++ b/docs/fleet_api_energy_sites.md @@ -397,10 +397,9 @@ an empty or `null` response even when it is not useful for proving local key readiness. `get_rsa_private_key(path)` loads an existing RSA private key for gateway -client registration 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. +client registration or creates a new unencrypted PEM key file; see the +[README](../README.md#bluetooth-for-vehicles) for its create/load semantics +and the `PrivateKeyError` raised for an unusable existing key file. ### Available Commands diff --git a/tesla_fleet_api/exceptions.py b/tesla_fleet_api/exceptions.py index b61d582..e9a478f 100644 --- a/tesla_fleet_api/exceptions.py +++ b/tesla_fleet_api/exceptions.py @@ -425,6 +425,25 @@ def __init__(self) -> None: ) +class PrivateKeyError(LibraryError): + """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). A local key-file + failure, not an upstream Fleet API error, so this subclasses + ``LibraryError`` rather than ``TeslaFleetError``. + """ + + def __init__(self, reason: str, message: str) -> None: + self.reason = reason + self.message = message + super().__init__(message) + + class SignedCommandRequired(TeslaFleetError): """The requested action requires a signed command; the unsigned cloud API cannot actuate it. diff --git a/tesla_fleet_api/tesla/tesla.py b/tesla_fleet_api/tesla/tesla.py index b2218d5..33e0e8d 100644 --- a/tesla_fleet_api/tesla/tesla.py +++ b/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 diff --git a/tests/test_tesla_private_key.py b/tests/test_tesla_private_key.py index 9546f9e..e90d422 100644 --- a/tests/test_tesla_private_key.py +++ b/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)