Add client-signed energy device_command transport

AGENTS.md

@@ -91,9 +91,11 @@ 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`**: 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`)
 

docs/energy_local_control.md

@@ -245,23 +245,28 @@ 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
+> 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 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. 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
 > 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

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`, 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

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)."
     )
 
 

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()