From 735873514a16c2ccda3e32612c4ee2845131e9e3 Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Thu, 29 Oct 2026 07:17:29 +1000 Subject: [PATCH 1/2] Add client-signed energy device_command transport EnergySite and TeslemetryEnergySite take an optional RSA private_key and gateway din. With a key, commands that need a signature (set_island_mode, go_off_grid, reconnect_grid, and get_backup_events/get_teg_config, which return 400 unsigned) are RSA-signed and sent through device_command; every other command stays on the unsigned path. Without a key nothing changes. TeslemetryEnergySite.set_island_mode calls super() when keyed, and goes to a _server_signed_command hook otherwise; the hook raises SignedCommandRequired until Teslemetry signs on its server. --- AGENTS.md | 7 +- docs/energy_local_control.md | 31 +- docs/fleet_api_energy_sites.md | 23 +- tesla_fleet_api/exceptions.py | 11 +- tesla_fleet_api/tesla/energy_signing.py | 129 ++++++++ tesla_fleet_api/tesla/energysite.py | 157 ++++++++-- tesla_fleet_api/teslemetry/energysite.py | 82 ++++- tests/test_energysite_island_mode.py | 10 +- tests/test_energysite_signed_transport.py | 362 ++++++++++++++++++++++ 9 files changed, 756 insertions(+), 56 deletions(-) create mode 100644 tesla_fleet_api/tesla/energy_signing.py create mode 100644 tests/test_energysite_signed_transport.py diff --git a/AGENTS.md b/AGENTS.md index 7f0865e..d5f7706 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -91,9 +91,10 @@ top-level workflow, never `workflow_call` — the trusted publisher is bound to per-backend health matrix. - Failover can double-execute a non-idempotent command that failed mid-flight. `BluetoothUnconfirmedCommand` must propagate without replay. -- `set_island_mode`/`go_off_grid`/`reconnect_grid` **must keep raising - `SignedCommandRequired`**: the unsigned `grpc_command` is acknowledged but does - not move the contactor. Only the signed local path actuates; verify state after. +- `set_island_mode`/`go_off_grid`/`reconnect_grid` **must never send the unsigned + `grpc_command`** and keep raising `SignedCommandRequired` without a signing key: + the unsigned command is acknowledged but does not move the contactor. Only a + signed path (local, or cloud `device_command`) actuates; verify state after. ## ObservationFunnel (`funnel.py`) diff --git a/docs/energy_local_control.md b/docs/energy_local_control.md index 26afcae..4aae28c 100644 --- a/docs/energy_local_control.md +++ b/docs/energy_local_control.md @@ -245,23 +245,24 @@ general `Router` semantics (per-command failover, the double-execution caveat, and the `health` check). > **Warning: island mode / off-grid actuation is not guaranteed, and the -> cloud path cannot do it at all.** +> unsigned cloud path cannot do it at all.** > -> This repo's cloud `EnergySite.set_island_mode()` / `go_off_grid()` / -> `reconnect_grid()` can only send an **unsigned** `grpc_command`, and -> gateways have been observed accepting that command without physically -> operating the grid contactor. Rather than ship that as a silent no-op with -> an OK-looking response, these three methods unconditionally raise -> `tesla_fleet_api.exceptions.SignedCommandRequired` - there is no cloud-only -> way to actuate the contactor. +> Gateways have been observed accepting the **unsigned** cloud +> `grpc_command` for `set_island_mode()` / `go_off_grid()` / +> `reconnect_grid()` without physically operating the grid contactor. So a +> cloud `EnergySite` built without a signing key raises +> `tesla_fleet_api.exceptions.SignedCommandRequired` for these three methods +> instead of returning a false success. Built with `private_key` and `din` +> (see [Client-signed device commands](fleet_api_energy_sites.md#client-signed-device-commands)), +> it signs them and sends them through the cloud `device_command` endpoint. > -> Routing through `EnergySiteRouter` with an `aiopowerwall` local primary is -> the intended way to actually operate the contactor, since it sends a -> signed request over the LAN. If the local primary fails and `Router` fails -> over to the cloud secondary, the call now raises `SignedCommandRequired` -> from the cloud side instead of returning a false success - the router -> semantics don't change (the local backend is still tried first), but a -> total failure now surfaces as a clear error. `aiopowerwall`'s own docs also +> Routing through `EnergySiteRouter` with an `aiopowerwall` local primary +> sends a signed request over the LAN first. If the local primary fails and +> `Router` fails over to the cloud secondary, a keyed cloud site sends the +> signed command, and a site without a key raises `SignedCommandRequired`. +> A command that failed mid-flight on the LAN may already have run, so the +> cloud retry can run it twice; island mode is absolute, so a second run sets +> the same mode. `aiopowerwall`'s own docs also > carry an unresolved, firmware-dependent caveat for its local > `set_island_mode` - some gateways acknowledge it without acting on it, and > `trigger_islanding()` (the explicit black-start command) may be needed as diff --git a/docs/fleet_api_energy_sites.md b/docs/fleet_api_energy_sites.md index 50d4435..d638f6a 100644 --- a/docs/fleet_api_energy_sites.md +++ b/docs/fleet_api_energy_sites.md @@ -388,7 +388,28 @@ tariff lacks either `seasons` or `energy_charges`. Energy gateways (Powerwalls, etc.) support gRPC commands sent via `POST /api/1/energy_sites/{id}/command`. These are undocumented Tesla API endpoints that communicate directly with the gateway hardware. All device command methods require the `energy_cmds` scope. -These commands are unsigned and cloud-only. The actuating `set_island_mode`/`go_off_grid`/`reconnect_grid` methods on `EnergySite` raise `SignedCommandRequired` unconditionally - the gateway can accept that unsigned command without physically taking effect, so there is no cloud-only way to actuate it. For registering a key and pairing it with a signed local LAN control path via the sibling `aiopowerwall` library, see [Energy: Local Control](energy_local_control.md). +By default these commands are unsigned. The actuating `set_island_mode`/`go_off_grid`/`reconnect_grid` methods need a signature: the gateway can accept the unsigned command without physically taking effect, so without a signing key they raise `SignedCommandRequired` before any request. For registering a key and pairing it with a signed local LAN control path via the sibling `aiopowerwall` library, see [Energy: Local Control](energy_local_control.md). + +### Client-signed device commands + +Once an RSA key is `VERIFIED` on the gateway (`add_authorized_client`, then the breaker toggle), pass it with the gateway DIN (the energy site's `gateway_id` in `products()`) to sign commands through the cloud: + +```python +site = api.energySites.create(energy_site_id, private_key=rsa_key, din=gateway_id) +await site.go_off_grid() +``` + +The site signs only commands that need a signature and posts them to the undocumented `POST /api/1/energy_sites/{id}/device_command`. Every other command keeps the unsigned path, because a command the gateway accepts unsigned is less likely to fail that way. + +| Method | With a signing key | Without a key | +|---|---|---| +| `set_island_mode`, `go_off_grid`, `reconnect_grid` | signed | raise `SignedCommandRequired` | +| `get_backup_events`, `get_teg_config` (400 unsigned on observed gateways) | signed | unsigned | +| every other device command | unsigned | unsigned | + +`None` means unsigned: there is no fallback to `api.rsa_private_key`. The signature expires after 30 seconds and has no counter, so the same bytes resent inside that window execute again; the library never retries a signed command. The reply is returned unchanged. A success-shaped reply does not prove the contactor moved: read `island_status` from `live_status()` after each step. + +On `Teslemetry`, the same signed commands go through Teslemetry's `device_command` route. Without a client key, `TeslemetryEnergySite.set_island_mode` goes to the `_server_signed_command` hook for Teslemetry to sign on its server; until that route signs, it raises `SignedCommandRequired`. Use `list_authorized_clients()` only as a secondary, best-effort cloud check while pairing a local key. The reliable verification is a successful signed diff --git a/tesla_fleet_api/exceptions.py b/tesla_fleet_api/exceptions.py index 85ce49b..a1a05a4 100644 --- a/tesla_fleet_api/exceptions.py +++ b/tesla_fleet_api/exceptions.py @@ -469,15 +469,18 @@ class SignedCommandRequired(TeslaFleetError): Energy gateways have been observed acknowledging this class of unsigned ``grpc_command`` without physically operating the grid contactor - see ``EnergySite.set_island_mode``. Pair an RSA key with - ``EnergySite.add_authorized_client`` and issue the command through a - signed local LAN backend (composed via ``EnergySiteRouter``) instead. + ``EnergySite.add_authorized_client``, then either build the energy site + with that ``private_key`` and its ``din`` to sign it through the cloud, or + issue it through a signed local LAN backend (composed via + ``EnergySiteRouter``). """ message = ( "This command cannot actuate via the unsigned cloud API - gateways " "acknowledge it without operating the contactor. Pair a key with " - "add_authorized_client and issue it through a signed local control " - "path (EnergySiteRouter) instead." + "add_authorized_client, then create the energy site with that " + "private_key and its din, or use a signed local control path " + "(EnergySiteRouter)." ) diff --git a/tesla_fleet_api/tesla/energy_signing.py b/tesla_fleet_api/tesla/energy_signing.py new file mode 100644 index 0000000..914438c --- /dev/null +++ b/tesla_fleet_api/tesla/energy_signing.py @@ -0,0 +1,129 @@ +"""RSA signing for energy device commands. + +One stateless scheme serves both the local TEDapi ``/tedapi/v1r`` endpoint and +Tesla's cloud ``device_command`` endpoint: a ``MessageEnvelope`` is wrapped in a +``RoutableMessage`` signed with RSA PKCS#1 v1.5 / SHA-512 over a TLV header +personalised to the gateway DIN. It matches aiopowerwall's +``V1rTransport._build_signed_request`` byte for byte; the cloud path was proven +live on 2026-09-21 with a 30 second TTL. + +There is no session, counter or handshake: the same bytes execute again if +resent before ``expires_at``. Keep the TTL short. +""" + +from __future__ import annotations + +import math +import struct +import time +import uuid +from typing import Any + +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import padding, rsa +from google.protobuf import json_format +from tesla_protocol.energy_device import ( + authorization_types_pb2, + signed_message_pb2, + transport_pb2, +) + +# The cloud round trip needs more than aiopowerwall's 12 s LAN TTL; 30 s +# worked live through ``device_command``. +ENERGY_SIGNATURE_TTL_SECONDS = 30 + + +def _tlv(tag: int, value: bytes) -> bytes: + """Encode one tag-length-value entry.""" + return bytes([tag, len(value)]) + value + + +class EnergyDeviceSigner: + """Builds RSA-signed ``RoutableMessage`` bytes for one energy device key. + + The key must be an RSA key registered on the gateway with + ``add_authorized_client`` and in state ``VERIFIED``. An ECC key cannot sign + this scheme: the ``SignatureData`` oneof has only ``rsa_data``. + """ + + def __init__( + self, + private_key: rsa.RSAPrivateKey, + ttl: int = ENERGY_SIGNATURE_TTL_SECONDS, + ) -> None: + if not isinstance(private_key, rsa.RSAPrivateKey): # pyright: ignore[reportUnnecessaryIsInstance] + raise TypeError( + "Energy device signing requires an RSA private key, " + f"got {type(private_key).__name__}" + ) + self._private_key = private_key + self.public_key_der = private_key.public_key().public_bytes( + serialization.Encoding.DER, serialization.PublicFormat.PKCS1 + ) + self.ttl = ttl + + @staticmethod + def envelope( + din: str, + category: str, + command: str, + params: dict[str, Any] | None = None, + ) -> bytes: + """Serialize a ``MessageEnvelope`` addressed to ``din``. + + ``category``, ``command`` and ``params`` are the same proto field names + ``EnergySite._command`` sends unsigned, so every gRPC helper can be + signed without per-command code. ``delivery_channel`` stays unset, as in + the envelope proven on the cloud path. + """ + envelope = json_format.ParseDict( + {category: {command: params or {}}}, transport_pb2.MessageEnvelope() + ) + envelope.sender.authorized_client = ( + authorization_types_pb2.AUTHORIZED_CLIENT_TYPE_CUSTOMER_MOBILE_APP + ) + envelope.recipient.din = din + return envelope.SerializeToString() + + def sign( + self, + envelope: bytes, + din: str, + *, + now: float | None = None, + message_uuid: bytes | None = None, + ) -> bytes: + """Wrap ``envelope`` in a signed ``RoutableMessage`` and serialize it. + + ``din`` must be the same gateway DIN the envelope and the request + target. ``now`` and ``message_uuid`` exist for deterministic tests. + """ + expires_at = math.ceil(time.time() if now is None else now) + self.ttl + signed_data = b"".join( + ( + _tlv( + signed_message_pb2.TAG_SIGNATURE_TYPE, + bytes([signed_message_pb2.SIGNATURE_TYPE_RSA]), + ), + _tlv( + signed_message_pb2.TAG_DOMAIN, + bytes([signed_message_pb2.DOMAIN_ENERGY_DEVICE]), + ), + _tlv(signed_message_pb2.TAG_PERSONALIZATION, din.encode("ascii")), + _tlv(signed_message_pb2.TAG_EXPIRES_AT, struct.pack(">I", expires_at)), + bytes([signed_message_pb2.TAG_END]), + envelope, + ) + ) + signature = self._private_key.sign( + signed_data, padding.PKCS1v15(), hashes.SHA512() + ) + + routable = signed_message_pb2.RoutableMessage() + routable.to_destination.domain = signed_message_pb2.DOMAIN_ENERGY_DEVICE + routable.protobuf_message_as_bytes = envelope + routable.uuid = message_uuid if message_uuid is not None else uuid.uuid4().bytes + routable.signature_data.signer_identity.public_key = self.public_key_der + routable.signature_data.rsa_data.expires_at = expires_at + routable.signature_data.rsa_data.signature = signature + return routable.SerializeToString() diff --git a/tesla_fleet_api/tesla/energysite.py b/tesla_fleet_api/tesla/energysite.py index 1ea78ed..abc8dc3 100644 --- a/tesla_fleet_api/tesla/energysite.py +++ b/tesla_fleet_api/tesla/energysite.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio import base64 import math from typing import TYPE_CHECKING, Any @@ -15,8 +16,11 @@ TeslaEnergyPeriod, ) from tesla_fleet_api.exceptions import SignedCommandRequired +from tesla_fleet_api.tesla.energy_signing import EnergyDeviceSigner if TYPE_CHECKING: + from cryptography.hazmat.primitives.asymmetric import rsa + from tesla_fleet_api.tesla.fleet import TeslaFleetApi # Upper bound on a ``dispatch_battery_power`` setpoint duration (one day), @@ -43,13 +47,35 @@ def validate_dispatch_battery_power(power_watts: float, duration_seconds: int) - class EnergySite: - """Class describing the Tesla Fleet API partner endpoints""" + """Class describing the Tesla Fleet API partner endpoints + + Pass ``private_key`` and ``din`` to send client-signed device commands + through the cloud ``device_command`` endpoint. Only commands that need a + signature use it; every other command keeps the unsigned path. The key + must be an RSA key that is ``VERIFIED`` on this gateway (see + ``add_authorized_client``), and ``din`` is the gateway DIN (the energy + site's ``gateway_id`` in ``products``). ``None`` means unsigned: unlike + vehicles, there is no fallback to the parent's key. + """ energy_site_id: int + din: str | None - def __init__(self, parent: TeslaFleetApi, energy_site_id: int) -> None: + def __init__( + self, + parent: TeslaFleetApi, + energy_site_id: int, + private_key: rsa.RSAPrivateKey | None = None, + din: str | None = None, + ) -> None: self._request = parent._request # pyright: ignore[reportPrivateUsage] self.energy_site_id = energy_site_id + self.din = din + self._signer: EnergyDeviceSigner | None = None + if private_key is not None: + if not din: + raise ValueError("A signing private_key requires the gateway din.") + self._signer = EnergyDeviceSigner(private_key) # Energy device gRPC commands based on research from # https://github.com/jasonacox/pypowerwall (MIT licensed) @@ -75,6 +101,63 @@ async def _command( }, ) + def _device_command_body( + self, + routable_message: bytes, + target_id: str, + identifier_type: EnergyDeviceIdentifierType | int, + ) -> dict[str, Any]: + """Build the flat ``device_command`` body around a signed message.""" + return { + "identifier_type": int(identifier_type), + "routable_message": base64.b64encode(routable_message).decode("ascii"), + "target_id": target_id, + } + + async def _device_command( + self, + routable_message: bytes, + target_id: str, + identifier_type: EnergyDeviceIdentifierType + | int = EnergyDeviceIdentifierType.GATEWAY_DIN, + ) -> dict[str, Any]: + """Send a signed ``RoutableMessage`` to the energy device. + + Tesla requires the body wrapped in ``data`` and ``identifier_type`` as + an integer. The reply is returned unchanged: its shape has not been + captured live yet, so it is not decoded here. + """ + return await self._request( + Method.POST, + f"api/1/energy_sites/{self.energy_site_id}/device_command", + json={ + "data": self._device_command_body( + routable_message, target_id, identifier_type + ) + }, + ) + + async def _signed_command( + self, + category: str, + command: str, + params: dict[str, Any] | None = None, + ) -> dict[str, Any]: + """Sign a gRPC command with the constructor key and send it to the gateway. + + Takes the same arguments as :meth:`_command`. Raises + ``SignedCommandRequired`` when the site has no signing key. A resend + of the same signed bytes inside the TTL executes again, so this never + retries. + """ + signer, din = self._signer, self.din + if signer is None or din is None: + raise SignedCommandRequired + envelope = signer.envelope(din, category, command, params) + # RSA-4096 with SHA-512 takes about 10 ms; keep it off the event loop. + routable_message = await asyncio.to_thread(signer.sign, envelope, din) + return await self._device_command(routable_message, din) + async def get_system_info(self) -> dict[str, Any]: """Get energy device system information including firmware version, device type, part number, serial number, and DIN.""" return await self._command("common", "get_system_info_request") @@ -229,7 +312,13 @@ async def get_signed_commands_public_key(self) -> dict[str, Any]: ) async def get_backup_events(self) -> dict[str, Any]: - """Get backup events from the energy gateway. May timeout on some firmware versions.""" + """Get backup events from the energy gateway. May timeout on some firmware versions. + + The unsigned request returned 400 on observed gateways, so a site + with a signing key sends it signed. + """ + if self._signer is not None: + return await self._signed_command("teg", "get_backup_events_request") return await self._command("teg", "get_backup_events_request") async def schedule_backup_event( @@ -268,7 +357,13 @@ async def cancel_backup_event(self) -> dict[str, Any]: return await self._command("teg", "cancel_manual_backup_event_request") async def get_teg_config(self) -> dict[str, Any]: - """Get Tesla Energy Gateway configuration from the TEG service.""" + """Get Tesla Energy Gateway configuration from the TEG service. + + The unsigned request returned 400 on observed gateways, so a site + with a signing key sends it signed. + """ + if self._signer is not None: + return await self._signed_command("teg", "get_config_request") return await self._command("teg", "get_config_request") async def dispatch_battery_power( @@ -325,32 +420,37 @@ async def set_island_mode( ) -> dict[str, Any]: """Set the island mode on the energy gateway. - Always raises ``SignedCommandRequired``: this cloud method can only - send an unsigned ``grpc_command``, and gateways have been observed - accepting that request without physically operating the grid - contactor. For a signed local LAN control path, pair an RSA key with - ``add_authorized_client`` and compose an aiopowerwall local backend - with ``EnergySiteRouter``. + Signed only: gateways have been observed accepting the unsigned + ``grpc_command`` without physically operating the grid contactor. + With a signing key this sends ``teg.set_island_mode_request`` signed + through ``device_command``. Without one it raises + ``SignedCommandRequired`` before any request. For a signed local LAN + path, compose an aiopowerwall local backend with ``EnergySiteRouter``. + + A success-shaped reply does not prove the contactor moved: confirm + ``island_status`` with a status read before relying on it. Args: - mode: Unused - the call raises before inspecting it. Kept for - signature compatibility. - force: Unused - the call raises before inspecting it. Kept for - signature compatibility. + mode: ``EnergyIslandMode.OFF_GRID`` (6) or ``ON_GRID`` (1). + force: Sent when not ``None``. Some gateways need ``True`` to + open the contactor. """ - raise SignedCommandRequired + params: dict[str, Any] = {"mode": int(mode)} + if force is not None: + params["force"] = force + return await self._signed_command("teg", "set_island_mode_request", params) async def go_off_grid(self) -> dict[str, Any]: - """Request off-grid mode through the cloud ``grpc_command`` path. + """Request off-grid mode, with ``force=True`` as aiopowerwall does. - Always raises ``SignedCommandRequired`` - see ``set_island_mode``. + Signed only - see ``set_island_mode``. """ - return await self.set_island_mode(EnergyIslandMode.OFF_GRID) + return await self.set_island_mode(EnergyIslandMode.OFF_GRID, force=True) async def reconnect_grid(self) -> dict[str, Any]: - """Request on-grid mode through the cloud ``grpc_command`` path. + """Request on-grid mode. - Always raises ``SignedCommandRequired`` - see ``set_island_mode``. + Signed only - see ``set_island_mode``. """ return await self.set_island_mode(EnergyIslandMode.ON_GRID) @@ -497,7 +597,18 @@ class EnergySites(dict[int, EnergySite]): def __init__(self, parent: TeslaFleetApi) -> None: self._parent = parent - def create(self, energy_site_id: int) -> EnergySite: - """Create a specific energy site.""" - self[energy_site_id] = self.Site(self._parent, energy_site_id) + def create( + self, + energy_site_id: int, + private_key: rsa.RSAPrivateKey | None = None, + din: str | None = None, + ) -> EnergySite: + """Create a specific energy site. + + Pass ``private_key`` and ``din`` to enable client-signed device + commands; see :class:`EnergySite`. + """ + self[energy_site_id] = self.Site( + self._parent, energy_site_id, private_key=private_key, din=din + ) return self[energy_site_id] diff --git a/tesla_fleet_api/teslemetry/energysite.py b/tesla_fleet_api/teslemetry/energysite.py index ad476ec..866100a 100644 --- a/tesla_fleet_api/teslemetry/energysite.py +++ b/tesla_fleet_api/teslemetry/energysite.py @@ -8,7 +8,7 @@ import warnings from collections.abc import Awaitable, Callable from dataclasses import dataclass -from typing import Any, cast +from typing import TYPE_CHECKING, Any, cast from tesla_fleet_api.const import ( AuthorizationRole, @@ -16,12 +16,15 @@ AuthorizedClientState, AuthorizedClientType, AuthorizedVerificationType, + EnergyDeviceIdentifierType, + EnergyIslandMode, Method, ) from tesla_fleet_api.exceptions import ( AuthorizedClientPairingTimedOut, AuthorizedClientWaitExpired, InvalidResponse, + SignedCommandRequired, TeslaFleetError, ) from tesla_fleet_api.tesla.energysite import ( @@ -30,6 +33,9 @@ validate_dispatch_battery_power, ) +if TYPE_CHECKING: + from cryptography.hazmat.primitives.asymmetric import rsa + DEFAULT_PAIRING_TIMEOUT = 600.0 DEFAULT_PAIRING_POLL_INTERVAL = 5.0 @@ -328,7 +334,64 @@ def _parse_gateway_address(payload: Any) -> str | None: class TeslemetryEnergySite(EnergySite): - """Teslemetry specific energy site.""" + """Teslemetry specific energy site. + + With a signing ``private_key`` (see :class:`EnergySite`), a command that + needs a signature calls ``super()`` to sign it with that key and sends it + through Teslemetry's ``device_command`` route. Without a key, such a + command goes to :meth:`_server_signed_command`, for Teslemetry to sign. + """ + + async def _device_command( + self, + routable_message: bytes, + target_id: str, + identifier_type: EnergyDeviceIdentifierType + | int = EnergyDeviceIdentifierType.GATEWAY_DIN, + ) -> dict[str, Any]: + """Send a signed ``RoutableMessage`` through Teslemetry's ``device_command`` route. + + Teslemetry takes the flat body and adds Tesla's ``data`` wrapper itself. + """ + return await self._request( + Method.POST, + f"api/1/energy_sites/{self.energy_site_id}/device_command", + json=self._device_command_body( + routable_message, target_id, identifier_type + ), + ) + + async def _server_signed_command( + self, route: str, data: dict[str, Any] + ) -> dict[str, Any]: + """Send a command that needs a signature for Teslemetry to sign on the server. + + This is the hook for a site without a client key. The target is + ``POST api/1/energy_sites/{id}/command/{route}`` with ``data`` as the + body. Today that route sends the unsigned ``grpc_command``, which the + gateway acknowledges without acting for these commands, so this raises + ``SignedCommandRequired`` before any request instead of returning a + false success. Send ``data`` to the route once Teslemetry signs there. + """ + raise SignedCommandRequired + + async def set_island_mode( + self, + mode: EnergyIslandMode | int, + force: bool | None = None, + ) -> dict[str, Any]: + """Set the island mode on the energy gateway. + + With a signing key this signs the command with that key; without one + it goes to :meth:`_server_signed_command`. See + :meth:`EnergySite.set_island_mode`. + """ + if self._signer is not None: + return await super().set_island_mode(mode, force) + data: dict[str, Any] = {"mode": int(mode)} + if force is not None: + data["force"] = force + return await self._server_signed_command("set_island_mode", data) async def add_authorized_client( self, @@ -612,8 +675,17 @@ class TeslemetryEnergySites(EnergySites): Site = TeslemetryEnergySite - def create(self, energy_site_id: int) -> TeslemetryEnergySite: - """Create a specific energy site.""" - site = self.Site(self._parent, energy_site_id) + def create( + self, + energy_site_id: int, + private_key: rsa.RSAPrivateKey | None = None, + din: str | None = None, + ) -> TeslemetryEnergySite: + """Create a specific energy site. + + Pass ``private_key`` and ``din`` to enable client-signed device + commands; see :class:`EnergySite`. + """ + site = self.Site(self._parent, energy_site_id, private_key=private_key, din=din) self[energy_site_id] = site return site diff --git a/tests/test_energysite_island_mode.py b/tests/test_energysite_island_mode.py index afab6c5..6865a7b 100644 --- a/tests/test_energysite_island_mode.py +++ b/tests/test_energysite_island_mode.py @@ -1,11 +1,11 @@ """Tests for EnergySite.set_island_mode/go_off_grid/reconnect_grid. -These cloud-only methods can only send an unsigned ``grpc_command`` - -gateways have been observed acknowledging that request without physically -operating the grid contactor. They must raise +Without a signing key these methods could only send an unsigned +``grpc_command`` - gateways have been observed acknowledging that request +without physically operating the grid contactor. They must raise :class:`~tesla_fleet_api.exceptions.SignedCommandRequired` instead of -shipping a silent no-op; the signed local control path -(``add_authorized_client`` + ``EnergySiteRouter``) is unaffected. +shipping a silent no-op. The signed paths are covered in +``test_energysite_signed_transport.py``. """ from __future__ import annotations diff --git a/tests/test_energysite_signed_transport.py b/tests/test_energysite_signed_transport.py new file mode 100644 index 0000000..19a8cb8 --- /dev/null +++ b/tests/test_energysite_signed_transport.py @@ -0,0 +1,362 @@ +"""Tests for the client-signed energy ``device_command`` transport. + +A site built with ``private_key`` and ``din`` signs only the commands that +need a signature (``set_island_mode`` and the reads that return 400 unsigned) +and posts them to ``device_command``. Every other command stays on the +unsigned ``grpc_command`` path, and a site without a key behaves as before. +""" + +from __future__ import annotations + +import base64 +import struct +from typing import Any +from unittest import IsolatedAsyncioTestCase, TestCase +from unittest.mock import AsyncMock + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ec, padding, rsa +from tesla_protocol.energy_device import ( + authorization_types_pb2, + signed_message_pb2, + transport_pb2, +) + +from tesla_fleet_api.const import EnergyIslandMode, Method +from tesla_fleet_api.exceptions import SignedCommandRequired +from tesla_fleet_api.router import EnergySiteRouter +from tesla_fleet_api.tesla.energy_signing import ( + ENERGY_SIGNATURE_TTL_SECONDS, + EnergyDeviceSigner, +) +from tesla_fleet_api.tesla.energysite import EnergySite, EnergySites +from tesla_fleet_api.teslemetry.energysite import ( + TeslemetryEnergySite, + TeslemetryEnergySites, +) + +SITE_ID = 12345 +DIN = "1232100-00-E--TG123456789ABC" +NOW = 1_790_000_000.25 +MESSAGE_UUID = bytes(range(16)) + +# A test key only; it is not registered on any gateway. +TEST_KEY = rsa.generate_private_key(public_exponent=65537, key_size=2048) + + +def _expected_signed_data(din: str, expires_at: int, envelope: bytes) -> bytes: + """The TLV header from the protocol, written out byte by byte.""" + personalization = din.encode("ascii") + return ( + bytes([0, 1, 7]) # TAG_SIGNATURE_TYPE = SIGNATURE_TYPE_RSA + + bytes([1, 1, 7]) # TAG_DOMAIN = DOMAIN_ENERGY_DEVICE + + bytes([2, len(personalization)]) + + personalization # TAG_PERSONALIZATION = DIN + + bytes([4, 4]) + + struct.pack(">I", expires_at) # TAG_EXPIRES_AT + + bytes([255]) # TAG_END + + envelope + ) + + +def _decode(routable_message: bytes) -> signed_message_pb2.RoutableMessage: + message = signed_message_pb2.RoutableMessage() + message.ParseFromString(routable_message) + return message + + +def _verify(message: signed_message_pb2.RoutableMessage, din: str) -> None: + """Raise ``InvalidSignature`` unless the message is signed by TEST_KEY for ``din``.""" + TEST_KEY.public_key().verify( + message.signature_data.rsa_data.signature, + _expected_signed_data( + din, + message.signature_data.rsa_data.expires_at, + message.protobuf_message_as_bytes, + ), + padding.PKCS1v15(), + hashes.SHA512(), + ) + + +def _envelope( + message: signed_message_pb2.RoutableMessage, +) -> transport_pb2.MessageEnvelope: + envelope = transport_pb2.MessageEnvelope() + envelope.ParseFromString(message.protobuf_message_as_bytes) + return envelope + + +def _make_site( + cls: type[EnergySite], private_key: rsa.RSAPrivateKey | None = TEST_KEY +) -> tuple[EnergySite, AsyncMock]: + request_mock: AsyncMock = AsyncMock(return_value={"response": "AA=="}) + parent = AsyncMock() + parent._request = request_mock + din = DIN if private_key is not None else None + return cls(parent, SITE_ID, private_key=private_key, din=din), request_mock + + +def _sent(request_mock: AsyncMock) -> tuple[str, dict[str, Any]]: + request_mock.assert_awaited_once() + call = request_mock.await_args + assert call is not None + method, path = call.args + assert method == Method.POST + return path, call.kwargs["json"] + + +class EnergyDeviceSignerTests(TestCase): + def test_signed_message_layout(self) -> None: + signer = EnergyDeviceSigner(TEST_KEY) + envelope = signer.envelope( + DIN, "teg", "set_island_mode_request", {"mode": 6, "force": True} + ) + message = _decode( + signer.sign(envelope, DIN, now=NOW, message_uuid=MESSAGE_UUID) + ) + + self.assertEqual( + message.to_destination.domain, signed_message_pb2.DOMAIN_ENERGY_DEVICE + ) + self.assertEqual(message.uuid, MESSAGE_UUID) + self.assertEqual(message.protobuf_message_as_bytes, envelope) + # expires_at rounds the clock up, then adds the 30 s TTL. + self.assertEqual(ENERGY_SIGNATURE_TTL_SECONDS, 30) + self.assertEqual(message.signature_data.rsa_data.expires_at, 1_790_000_001 + 30) + self.assertEqual( + message.signature_data.signer_identity.public_key, + TEST_KEY.public_key().public_bytes( + serialization.Encoding.DER, serialization.PublicFormat.PKCS1 + ), + ) + _verify(message, DIN) + + def test_signature_is_personalised_to_the_din(self) -> None: + signer = EnergyDeviceSigner(TEST_KEY) + envelope = signer.envelope(DIN, "teg", "get_backup_events_request") + message = _decode(signer.sign(envelope, DIN, now=NOW)) + with self.assertRaises(InvalidSignature): + _verify(message, "1232100-00-E--TG000000000000") + + def test_envelope_header_and_payload(self) -> None: + envelope = transport_pb2.MessageEnvelope() + envelope.ParseFromString( + EnergyDeviceSigner.envelope( + DIN, "teg", "set_island_mode_request", {"mode": 6, "force": True} + ) + ) + self.assertEqual( + envelope.sender.authorized_client, + authorization_types_pb2.AUTHORIZED_CLIENT_TYPE_CUSTOMER_MOBILE_APP, + ) + self.assertEqual(envelope.recipient.din, DIN) + # The envelope proven on the cloud path left delivery_channel unset. + self.assertEqual( + envelope.delivery_channel, transport_pb2.DELIVERY_CHANNEL_INVALID + ) + self.assertEqual(envelope.teg.set_island_mode_request.mode, 6) + self.assertTrue(envelope.teg.set_island_mode_request.force) + + def test_empty_request_sets_the_payload(self) -> None: + envelope = transport_pb2.MessageEnvelope() + envelope.ParseFromString( + EnergyDeviceSigner.envelope(DIN, "teg", "get_config_request") + ) + self.assertEqual(envelope.WhichOneof("payload"), "teg") + self.assertEqual(envelope.teg.WhichOneof("message"), "get_config_request") + + def test_each_message_gets_a_fresh_uuid(self) -> None: + signer = EnergyDeviceSigner(TEST_KEY) + envelope = signer.envelope(DIN, "teg", "get_config_request") + self.assertNotEqual( + _decode(signer.sign(envelope, DIN)).uuid, + _decode(signer.sign(envelope, DIN)).uuid, + ) + + def test_rejects_an_ecc_key(self) -> None: + with self.assertRaises(TypeError): + EnergyDeviceSigner(ec.generate_private_key(ec.SECP256R1())) # type: ignore[arg-type] + + +class SigningKeyConstructorTests(TestCase): + def test_key_requires_din(self) -> None: + parent = AsyncMock() + with self.assertRaises(ValueError): + EnergySite(parent, SITE_ID, private_key=TEST_KEY) + + def test_create_passes_key_and_din(self) -> None: + parent = AsyncMock() + for sites in (EnergySites(parent), TeslemetryEnergySites(parent)): + with self.subTest(sites=type(sites).__name__): + site = sites.create(SITE_ID, private_key=TEST_KEY, din=DIN) + self.assertEqual(site.din, DIN) + self.assertIsNotNone(site._signer) # pyright: ignore[reportPrivateUsage] + self.assertIsNone(sites.create(SITE_ID + 1)._signer) # pyright: ignore[reportPrivateUsage] + + +class FleetSignedTransportTests(IsolatedAsyncioTestCase): + async def test_set_island_mode_posts_signed_device_command(self) -> None: + site, request_mock = _make_site(EnergySite) + await site.set_island_mode(EnergyIslandMode.OFF_GRID, force=True) + + path, body = _sent(request_mock) + self.assertEqual(path, f"api/1/energy_sites/{SITE_ID}/device_command") + # Tesla requires the body inside "data" and identifier_type as an int. + self.assertEqual(list(body), ["data"]) + data = body["data"] + self.assertEqual(data["identifier_type"], 1) + self.assertIsInstance(data["identifier_type"], int) + self.assertEqual(data["target_id"], DIN) + message = _decode(base64.b64decode(data["routable_message"])) + _verify(message, DIN) + request = _envelope(message).teg.set_island_mode_request + self.assertEqual(request.mode, EnergyIslandMode.OFF_GRID) + self.assertTrue(request.force) + + async def test_go_off_grid_forces_and_reconnect_does_not(self) -> None: + site, request_mock = _make_site(EnergySite) + await site.go_off_grid() + message = _decode( + base64.b64decode(_sent(request_mock)[1]["data"]["routable_message"]) + ) + request = _envelope(message).teg.set_island_mode_request + self.assertEqual(request.mode, EnergyIslandMode.OFF_GRID) + self.assertTrue(request.force) + + request_mock.reset_mock() + await site.reconnect_grid() + message = _decode( + base64.b64decode(_sent(request_mock)[1]["data"]["routable_message"]) + ) + request = _envelope(message).teg.set_island_mode_request + self.assertEqual(request.mode, EnergyIslandMode.ON_GRID) + self.assertFalse(request.force) + + async def test_island_mode_without_key_still_raises(self) -> None: + site, request_mock = _make_site(EnergySite, private_key=None) + for call in ( + lambda: site.set_island_mode(EnergyIslandMode.OFF_GRID), + site.go_off_grid, + site.reconnect_grid, + ): + with self.assertRaises(SignedCommandRequired): + await call() + request_mock.assert_not_called() + + async def test_reads_that_fail_unsigned_go_signed_with_a_key(self) -> None: + for name, request_field in ( + ("get_backup_events", "get_backup_events_request"), + ("get_teg_config", "get_config_request"), + ): + with self.subTest(name=name): + site, request_mock = _make_site(EnergySite) + await getattr(site, name)() + path, body = _sent(request_mock) + self.assertEqual(path, f"api/1/energy_sites/{SITE_ID}/device_command") + message = _decode(base64.b64decode(body["data"]["routable_message"])) + _verify(message, DIN) + self.assertEqual( + _envelope(message).teg.WhichOneof("message"), request_field + ) + + site, request_mock = _make_site(EnergySite, private_key=None) + await getattr(site, name)() + path, body = _sent(request_mock) + self.assertEqual(path, f"api/1/energy_sites/{SITE_ID}/command") + self.assertEqual( + body["command_properties"]["message"], {"teg": {request_field: {}}} + ) + + async def test_unsigned_commands_stay_unsigned_with_a_key(self) -> None: + calls: list[tuple[str, tuple[Any, ...]]] = [ + ("get_system_info", ()), + ("list_authorized_clients", ()), + ("get_signed_commands_public_key", ()), + ("schedule_backup_event", ()), + ("cancel_backup_event", ()), + ("dispatch_battery_power", (5000, 60)), + ("resume_battery_operation", ()), + ] + for name, args in calls: + with self.subTest(name=name): + site, request_mock = _make_site(EnergySite) + await getattr(site, name)(*args) + path, body = _sent(request_mock) + self.assertEqual(path, f"api/1/energy_sites/{SITE_ID}/command") + self.assertEqual(body["command_type"], "grpc_command") + + +class TeslemetrySignedTransportTests(IsolatedAsyncioTestCase): + async def test_set_island_mode_with_key_signs_through_super(self) -> None: + site, request_mock = _make_site(TeslemetryEnergySite) + await site.go_off_grid() + + path, body = _sent(request_mock) + self.assertEqual(path, f"api/1/energy_sites/{SITE_ID}/device_command") + # Teslemetry takes the flat body and adds Tesla's "data" wrapper. + self.assertEqual( + sorted(body), ["identifier_type", "routable_message", "target_id"] + ) + self.assertEqual(body["identifier_type"], 1) + self.assertEqual(body["target_id"], DIN) + message = _decode(base64.b64decode(body["routable_message"])) + _verify(message, DIN) + request = _envelope(message).teg.set_island_mode_request + self.assertEqual(request.mode, EnergyIslandMode.OFF_GRID) + self.assertTrue(request.force) + + async def test_set_island_mode_without_key_uses_server_signing_hook(self) -> None: + site, request_mock = _make_site(TeslemetryEnergySite, private_key=None) + assert isinstance(site, TeslemetryEnergySite) + hook = AsyncMock(return_value={"response": {}}) + site._server_signed_command = hook # type: ignore[method-assign] + await site.go_off_grid() + hook.assert_awaited_once_with("set_island_mode", {"mode": 6, "force": True}) + request_mock.assert_not_called() + + async def test_server_signing_hook_raises_until_teslemetry_signs(self) -> None: + site, request_mock = _make_site(TeslemetryEnergySite, private_key=None) + for call in ( + lambda: site.set_island_mode(EnergyIslandMode.ON_GRID), + site.go_off_grid, + site.reconnect_grid, + ): + with self.assertRaises(SignedCommandRequired): + await call() + request_mock.assert_not_called() + + async def test_teslemetry_routes_stay_unsigned_with_a_key(self) -> None: + site, request_mock = _make_site(TeslemetryEnergySite) + await site.dispatch_battery_power(5000, 60) + path, _ = _sent(request_mock) + self.assertEqual( + path, f"api/1/energy_sites/{SITE_ID}/command/dispatch_battery_power" + ) + + +class _LocalPowerwallSite: + """A local backend that cannot reach the gateway.""" + + def __init__(self) -> None: + self.calls: list[str] = [] + + async def go_off_grid(self) -> None: + self.calls.append("go_off_grid") + raise ConnectionError("local unreachable") + + +class RouterSignedTransportTests(IsolatedAsyncioTestCase): + async def test_router_falls_back_to_signed_cloud(self) -> None: + for cls in (EnergySite, TeslemetryEnergySite): + with self.subTest(cls=cls.__name__): + local = _LocalPowerwallSite() + cloud, request_mock = _make_site(cls) + router = EnergySiteRouter(local, cloud) + + await router.go_off_grid() + + self.assertEqual(local.calls, ["go_off_grid"]) + path, _ = _sent(request_mock) + self.assertEqual(path, f"api/1/energy_sites/{SITE_ID}/device_command") From 4de0484e4e6475a3f853358a1659e5db27bc213a Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Thu, 29 Oct 2026 09:49:22 +1000 Subject: [PATCH 2/2] Use Teslemetry's server-signed routes when no client key is given Per the captain's rule on #198: a TeslemetryEnergySite signs locally only when it has an RSA key. Without one, set_island_mode, get_backup_events and get_teg_config call the routes api #888 added (POST command/set_island_mode, GET command/backup_events, GET command/config), which Teslemetry signs with its own key. This replaces the hook that raised SignedCommandRequired. --- AGENTS.md | 7 +-- docs/energy_local_control.md | 8 ++- docs/fleet_api_energy_sites.md | 2 +- tesla_fleet_api/teslemetry/energysite.py | 59 +++++++++++++-------- tests/test_energysite_signed_transport.py | 62 ++++++++++++++++------- 5 files changed, 94 insertions(+), 44 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index d5f7706..c854a8a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -92,9 +92,10 @@ top-level workflow, never `workflow_call` — the trusted publisher is bound to - Failover can double-execute a non-idempotent command that failed mid-flight. `BluetoothUnconfirmedCommand` must propagate without replay. - `set_island_mode`/`go_off_grid`/`reconnect_grid` **must never send the unsigned - `grpc_command`** and keep raising `SignedCommandRequired` without a signing key: - the unsigned command is acknowledged but does not move the contactor. Only a - signed path (local, or cloud `device_command`) actuates; verify state after. + `grpc_command`**: it is acknowledged but does not move the contactor. Without a + signing key, Fleet `EnergySite` raises `SignedCommandRequired`, and + `TeslemetryEnergySite` calls Teslemetry's server-signed route. Only a signed + path actuates; verify state after. ## ObservationFunnel (`funnel.py`) diff --git a/docs/energy_local_control.md b/docs/energy_local_control.md index 4aae28c..d9d0912 100644 --- a/docs/energy_local_control.md +++ b/docs/energy_local_control.md @@ -250,16 +250,20 @@ and the `health` check). > Gateways have been observed accepting the **unsigned** cloud > `grpc_command` for `set_island_mode()` / `go_off_grid()` / > `reconnect_grid()` without physically operating the grid contactor. So a -> cloud `EnergySite` built without a signing key raises +> Fleet API `EnergySite` built without a signing key raises > `tesla_fleet_api.exceptions.SignedCommandRequired` for these three methods > instead of returning a false success. Built with `private_key` and `din` > (see [Client-signed device commands](fleet_api_energy_sites.md#client-signed-device-commands)), > it signs them and sends them through the cloud `device_command` endpoint. +> A `TeslemetryEnergySite` without a key calls Teslemetry's route, which +> signs with Teslemetry's key. > > Routing through `EnergySiteRouter` with an `aiopowerwall` local primary > sends a signed request over the LAN first. If the local primary fails and > `Router` fails over to the cloud secondary, a keyed cloud site sends the -> signed command, and a site without a key raises `SignedCommandRequired`. +> signed command. A Fleet API site without a key raises +> `SignedCommandRequired`, and a Teslemetry site without a key calls +> Teslemetry's signed route. > A command that failed mid-flight on the LAN may already have run, so the > cloud retry can run it twice; island mode is absolute, so a second run sets > the same mode. `aiopowerwall`'s own docs also diff --git a/docs/fleet_api_energy_sites.md b/docs/fleet_api_energy_sites.md index d638f6a..247d09e 100644 --- a/docs/fleet_api_energy_sites.md +++ b/docs/fleet_api_energy_sites.md @@ -409,7 +409,7 @@ The site signs only commands that need a signature and posts them to the undocum `None` means unsigned: there is no fallback to `api.rsa_private_key`. The signature expires after 30 seconds and has no counter, so the same bytes resent inside that window execute again; the library never retries a signed command. The reply is returned unchanged. A success-shaped reply does not prove the contactor moved: read `island_status` from `live_status()` after each step. -On `Teslemetry`, the same signed commands go through Teslemetry's `device_command` route. Without a client key, `TeslemetryEnergySite.set_island_mode` goes to the `_server_signed_command` hook for Teslemetry to sign on its server; until that route signs, it raises `SignedCommandRequired`. +On `Teslemetry`, a client key works the same way: the commands that need a signature are signed with that key and go through Teslemetry's `device_command` route. Without a client key, `TeslemetryEnergySite` never raises `SignedCommandRequired` for them. It calls Teslemetry's routes instead (`POST command/set_island_mode`, `GET command/backup_events`, `GET command/config`), and Teslemetry signs with its own key. Teslemetry's key must be `VERIFIED` on the gateway, or those routes answer 409 `energy_site_not_paired` and send nothing. Use `list_authorized_clients()` only as a secondary, best-effort cloud check while pairing a local key. The reliable verification is a successful signed diff --git a/tesla_fleet_api/teslemetry/energysite.py b/tesla_fleet_api/teslemetry/energysite.py index 866100a..7b7eb8d 100644 --- a/tesla_fleet_api/teslemetry/energysite.py +++ b/tesla_fleet_api/teslemetry/energysite.py @@ -24,7 +24,6 @@ AuthorizedClientPairingTimedOut, AuthorizedClientWaitExpired, InvalidResponse, - SignedCommandRequired, TeslaFleetError, ) from tesla_fleet_api.tesla.energysite import ( @@ -338,8 +337,10 @@ class TeslemetryEnergySite(EnergySite): With a signing ``private_key`` (see :class:`EnergySite`), a command that needs a signature calls ``super()`` to sign it with that key and sends it - through Teslemetry's ``device_command`` route. Without a key, such a - command goes to :meth:`_server_signed_command`, for Teslemetry to sign. + through Teslemetry's ``device_command`` route. Without a key, it uses its + Teslemetry route, where Teslemetry signs it with its own key. That key + must be verified on the gateway, or the route answers 409 + ``energy_site_not_paired`` and sends nothing. """ async def _device_command( @@ -361,20 +362,6 @@ async def _device_command( ), ) - async def _server_signed_command( - self, route: str, data: dict[str, Any] - ) -> dict[str, Any]: - """Send a command that needs a signature for Teslemetry to sign on the server. - - This is the hook for a site without a client key. The target is - ``POST api/1/energy_sites/{id}/command/{route}`` with ``data`` as the - body. Today that route sends the unsigned ``grpc_command``, which the - gateway acknowledges without acting for these commands, so this raises - ``SignedCommandRequired`` before any request instead of returning a - false success. Send ``data`` to the route once Teslemetry signs there. - """ - raise SignedCommandRequired - async def set_island_mode( self, mode: EnergyIslandMode | int, @@ -382,16 +369,46 @@ async def set_island_mode( ) -> dict[str, Any]: """Set the island mode on the energy gateway. - With a signing key this signs the command with that key; without one - it goes to :meth:`_server_signed_command`. See - :meth:`EnergySite.set_island_mode`. + With a signing key this signs the command with that key. Without one + it calls the Teslemetry route, which signs it with Teslemetry's key. + See :meth:`EnergySite.set_island_mode`. """ if self._signer is not None: return await super().set_island_mode(mode, force) data: dict[str, Any] = {"mode": int(mode)} if force is not None: data["force"] = force - return await self._server_signed_command("set_island_mode", data) + return await self._request( + Method.POST, + f"api/1/energy_sites/{self.energy_site_id}/command/set_island_mode", + json=data, + ) + + async def get_backup_events(self) -> dict[str, Any]: + """Get backup events from the energy gateway. + + With a signing key this signs the command with that key. Without one + it calls the Teslemetry route, which signs it with Teslemetry's key. + """ + if self._signer is not None: + return await super().get_backup_events() + return await self._request( + Method.GET, + f"api/1/energy_sites/{self.energy_site_id}/command/backup_events", + ) + + async def get_teg_config(self) -> dict[str, Any]: + """Get Tesla Energy Gateway configuration from the TEG service. + + With a signing key this signs the command with that key. Without one + it calls the Teslemetry route, which signs it with Teslemetry's key. + """ + if self._signer is not None: + return await super().get_teg_config() + return await self._request( + Method.GET, + f"api/1/energy_sites/{self.energy_site_id}/command/config", + ) async def add_authorized_client( self, diff --git a/tests/test_energysite_signed_transport.py b/tests/test_energysite_signed_transport.py index 19a8cb8..75e4ce3 100644 --- a/tests/test_energysite_signed_transport.py +++ b/tests/test_energysite_signed_transport.py @@ -307,25 +307,53 @@ async def test_set_island_mode_with_key_signs_through_super(self) -> None: self.assertEqual(request.mode, EnergyIslandMode.OFF_GRID) self.assertTrue(request.force) - async def test_set_island_mode_without_key_uses_server_signing_hook(self) -> None: - site, request_mock = _make_site(TeslemetryEnergySite, private_key=None) - assert isinstance(site, TeslemetryEnergySite) - hook = AsyncMock(return_value={"response": {}}) - site._server_signed_command = hook # type: ignore[method-assign] - await site.go_off_grid() - hook.assert_awaited_once_with("set_island_mode", {"mode": 6, "force": True}) - request_mock.assert_not_called() + async def test_reads_with_key_sign_through_super(self) -> None: + for name, request_field in ( + ("get_backup_events", "get_backup_events_request"), + ("get_teg_config", "get_config_request"), + ): + with self.subTest(name=name): + site, request_mock = _make_site(TeslemetryEnergySite) + await getattr(site, name)() + path, body = _sent(request_mock) + self.assertEqual(path, f"api/1/energy_sites/{SITE_ID}/device_command") + message = _decode(base64.b64decode(body["routable_message"])) + _verify(message, DIN) + self.assertEqual( + _envelope(message).teg.WhichOneof("message"), request_field + ) - async def test_server_signing_hook_raises_until_teslemetry_signs(self) -> None: - site, request_mock = _make_site(TeslemetryEnergySite, private_key=None) - for call in ( - lambda: site.set_island_mode(EnergyIslandMode.ON_GRID), - site.go_off_grid, - site.reconnect_grid, + async def test_island_mode_without_key_uses_the_server_signed_route(self) -> None: + calls: list[tuple[str, tuple[Any, ...], dict[str, Any]]] = [ + ("go_off_grid", (), {"mode": 6, "force": True}), + ("reconnect_grid", (), {"mode": 1}), + ( + "set_island_mode", + (EnergyIslandMode.ON_GRID, False), + {"mode": 1, "force": False}, + ), + ] + for name, args, body in calls: + with self.subTest(name=name): + site, request_mock = _make_site(TeslemetryEnergySite, private_key=None) + await getattr(site, name)(*args) + request_mock.assert_awaited_once_with( + Method.POST, + f"api/1/energy_sites/{SITE_ID}/command/set_island_mode", + json=body, + ) + + async def test_reads_without_key_use_the_server_signed_routes(self) -> None: + for name, route in ( + ("get_backup_events", "backup_events"), + ("get_teg_config", "config"), ): - with self.assertRaises(SignedCommandRequired): - await call() - request_mock.assert_not_called() + with self.subTest(name=name): + site, request_mock = _make_site(TeslemetryEnergySite, private_key=None) + await getattr(site, name)() + request_mock.assert_awaited_once_with( + Method.GET, f"api/1/energy_sites/{SITE_ID}/command/{route}" + ) async def test_teslemetry_routes_stay_unsigned_with_a_key(self) -> None: site, request_mock = _make_site(TeslemetryEnergySite)