Sign v1r requests through EnergyDeviceSigner

tesla_fleet_api/tesla/powerwall/transport.py

@@ -16,19 +16,16 @@
 import asyncio
 import gzip
 import logging
-import math
 import re
-import struct
-import time
-import uuid
 from contextlib import suppress
 from typing import Any, Final, cast
 
 import aiohttp
-from cryptography.hazmat.primitives import hashes, serialization
-from cryptography.hazmat.primitives.asymmetric import padding, rsa
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
 from tesla_protocol.energy_device import signed_message_pb2
 
+from ..energy_signing import EnergyDeviceSigner
 from .exceptions import (
     PowerwallAuthenticationError,
     PowerwallConnectionError,
@@ -39,16 +36,6 @@
 
 _LOGGER = logging.getLogger(__name__)
 
-# v1r signature TLV tag identifiers — signed_message_pb2, enum Tag.
-_TAG_SIGNATURE_TYPE: Final = signed_message_pb2.TAG_SIGNATURE_TYPE
-_TAG_DOMAIN: Final = signed_message_pb2.TAG_DOMAIN
-_TAG_PERSONALIZATION: Final = signed_message_pb2.TAG_PERSONALIZATION
-_TAG_EXPIRES_AT: Final = signed_message_pb2.TAG_EXPIRES_AT
-_TAG_END: Final = signed_message_pb2.TAG_END
-
-_SIGNATURE_TYPE_RSA: Final = signed_message_pb2.SIGNATURE_TYPE_RSA
-_DOMAIN_ENERGY_DEVICE: Final = signed_message_pb2.DOMAIN_ENERGY_DEVICE
-
 # Signed messages must be consumed by the gateway within this window.
 _SIGNATURE_TTL_SECONDS: Final = 12
 
@@ -71,11 +58,6 @@ def _decompress(content: bytes) -> bytes:
     return content
 
 
-def _tlv(tag: int, value: bytes) -> bytes:
-    """Encode a single tag-length-value entry."""
-    return bytes([tag, len(value)]) + value
-
-
 def _customer_password(gateway_password: str) -> str:
     """Derive the `/api/login/Basic` "customer" password from the gateway password.
 
@@ -113,12 +95,8 @@ def __init__(
         self._timeout = aiohttp.ClientTimeout(total=timeout)
         self._token: str | None = None
         self._token_lock = asyncio.Lock()
-        self._private_key = self._load_private_key(rsa_private_key_pem)
-        # PKCS1 DER of the public key — sent as the v1r KeyIdentity so the
-        # gateway can look up the registered key.
-        self._public_key_der: bytes = self._private_key.public_key().public_bytes(
-            serialization.Encoding.DER,
-            serialization.PublicFormat.PKCS1,
+        self._signer = EnergyDeviceSigner(
+            self._load_private_key(rsa_private_key_pem), ttl=_SIGNATURE_TTL_SECONDS
         )
 
     # ── Public API ──────────────────────────────────────────────────────────
@@ -349,47 +327,10 @@ async def _post_octets(
         except TimeoutError as err:
             raise PowerwallConnectionError("v1r POST timed out") from err
 
-    async def _build_signed_request(
-        self, envelope_bytes: bytes, din: str
-    ) -> bytes:
+    async def _build_signed_request(self, envelope_bytes: bytes, din: str) -> bytes:
         """Wrap `envelope_bytes` in a signed RoutableMessage and serialize it.
 
         Signing is offloaded to a thread because RSA-4096 with SHA-512 is
         ~10 ms of CPU work and would otherwise block the event loop.
         """
-        routable = signed_message_pb2.RoutableMessage()
-        routable.to_destination.domain = _DOMAIN_ENERGY_DEVICE
-        routable.protobuf_message_as_bytes = envelope_bytes
-        routable.uuid = uuid.uuid4().bytes
-
-        expires_at = math.ceil(time.time()) + _SIGNATURE_TTL_SECONDS
-        tlv_payload = self._build_tlv(din, expires_at, envelope_bytes)
-        signature = await asyncio.to_thread(self._sign, tlv_payload)
-
-        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
-
-        serialized: bytes = routable.SerializeToString()
-        return serialized
-
-    @staticmethod
-    def _build_tlv(din: str, expires_at: int, inner: bytes) -> bytes:
-        return b"".join(
-            (
-                _tlv(_TAG_SIGNATURE_TYPE, bytes([_SIGNATURE_TYPE_RSA])),
-                _tlv(_TAG_DOMAIN, bytes([_DOMAIN_ENERGY_DEVICE])),
-                _tlv(_TAG_PERSONALIZATION, din.encode("ascii")),
-                _tlv(_TAG_EXPIRES_AT, struct.pack(">I", expires_at)),
-                bytes([_TAG_END]),
-                inner,
-            )
-        )
-
-    def _sign(self, tlv_payload: bytes) -> bytes:
-        signature: bytes = self._private_key.sign(
-            data=tlv_payload,
-            padding=padding.PKCS1v15(),
-            algorithm=hashes.SHA512(),
-        )
-        return signature
+        return await asyncio.to_thread(self._signer.sign, envelope_bytes, din)

tests/powerwall/test_transport_signing.py

@@ -0,0 +1,144 @@
+"""Fixed byte vector for the `/tedapi/v1r` signed request.
+
+RSA PKCS#1 v1.5 signing is deterministic, so with a fixed key, envelope, DIN,
+clock and message UUID the whole signed `RoutableMessage` is fixed. Any change
+to the TLV header, the 12 second LAN TTL or the message layout breaks this test.
+"""
+
+from __future__ import annotations
+
+import uuid
+from typing import Any
+from unittest.mock import patch
+
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.asymmetric import padding, rsa
+from cryptography.hazmat.primitives.serialization import load_pem_private_key
+from tesla_protocol.energy_device import signed_message_pb2
+
+from tesla_fleet_api.tesla.powerwall.transport import V1rTransport
+
+# A test key only; it is not registered on any gateway.
+TEST_KEY_PEM = b"""-----BEGIN PRIVATE KEY-----
+MIICdQIBADANBgkqhkiG9w0BAQEFAASCAl8wggJbAgEAAoGBAJmlOoA/z98lbrJt
+SCpsXVPsgRDc2KkclGXGDR+4iMP/Fttoy0Hx+t8q01WIDHCoOpoQv1NwOWg1ycEU
+DLvIryyXH4eOrqY+ZWaAqiqyZ+HjMeNhW4HpD43SkquIn9+hosXJaJxfrNvI1h9g
+PIxUnqdKQM+zYmhl3p2NxKkpbnQ7AgMBAAECgYB/TSM3d6en37/kHh+Y5sg87MvT
+SOhK1d8uo/XjvmWy3vv1NtKSBQgzN6x6RYrpHLnRROcBHw6xhT9wtkgGnrRkGPqu
+cKK7w9Dr6jr02IlUL2vUVb2OY1exef9CP40BMfMM5QxqSE2IrG+u6YR+I3dMPPEP
+PrJqucOf/Y78MJuUoQJBAMfGCBJKy19ByzP2xv5d//w7o0E8vGZlYdOy8LtYMNeX
+4cM74LzZgA9MmDruT+v1tLTwPKAt2HgnDuEA9AYTdU8CQQDE45lrTItafMAnC9a4
+x9Zu0PgPl2z3ibtd0k0vTvVhYN7k6+ulRy5EaEPfdSkxSRfAayA/pXDPy/3IIe8I
+4S9VAkA6ohnOXJtyGtpsJvriv24ZdtuEkzHsj8b1nmhfZBP5wv8lERb19pRKaT2o
+1jco1xWxoWz5ylrlcPlB+sikW+OrAkAu0jUsyBInnZ0QbAdO/+7lgXpMMuVt5GoT
+QKysQGmoE6JN7DCoqEm/aNAjNrhqCQjF8HPFzvEJ6WfbjP5MLUjVAkAa4eA/MCuB
+ip8rpmlk2gKy5kT48ehO4GSwb168TI+Bkw0hRAwV+4r1STkCJR+z7fWEi+uBn7VI
+IZvqjXsKph1j
+-----END PRIVATE KEY-----
+"""
+
+DIN = "1232100-00-E--TG123456789ABC"
+NOW = 1_790_000_000.25
+MESSAGE_UUID = bytes(range(16))
+# A MessageEnvelope from the customer mobile app client, addressed to DIN.
+ENVELOPE = bytes.fromhex(
+    "120220011a1e0a1c313233323130302d30302d452d2d5447313233343536373839414243"
+)
+
+EXPECTED_SIGNED_REQUEST = bytes.fromhex(
+    "320208075224120220011a1e0a1c313233323130302d30302d452d2d54473132"
+    "333435363738394142436a9d020a8f010a8c013081890281810099a53a803fcf"
+    "df256eb26d482a6c5d53ec8110dcd8a91c9465c60d1fb888c3ff16db68cb41f1"
+    "fadf2ad355880c70a83a9a10bf5370396835c9c1140cbbc8af2c971f878eaea6"
+    "3e656680aa2ab267e1e331e3615b81e90f8dd292ab889fdfa1a2c5c9689c5fac"
+    "dbc8d61f603c8c549ea74a40cfb3626865de9d8dc4a9296e743b02030100013a"
+    "88010d8d3bb16a1280016f5fd6a464eeb8e293b9efe76d1036c4fb09ac801dcd"
+    "d6aebc3fd06ed15629626f307a1f7cd16cd5fbec82adb9b0b82faaaf9327452f"
+    "3d93ebe0b8538b40090900ce9bd0f4f83b3783bd9a33ba97d72ce921cc0fa40c"
+    "3b7eb65dea88cd26ef013085b2cf3315caa23a13429ca71b3fd97ee41411d97a"
+    "2948117e0e07d5e7c55b9a0310000102030405060708090a0b0c0d0e0f"
+)
+
+REPLY_ENVELOPE = b"\x2a\x03\xb2\x02\x00"
+
+
+class _FakeResponse:
+    def __init__(self, body: bytes) -> None:
+        self.status = 200
+        self._body = body
+
+    async def __aenter__(self) -> _FakeResponse:
+        return self
+
+    async def __aexit__(self, *exc_info: object) -> None:
+        return None
+
+    async def read(self) -> bytes:
+        return self._body
+
+
+class _FakeSession:
+    def __init__(self) -> None:
+        self.posted: list[bytes] = []
+
+    def post(self, url: str, *, data: bytes, **kwargs: Any) -> _FakeResponse:
+        self.posted.append(data)
+        reply = signed_message_pb2.RoutableMessage()
+        reply.protobuf_message_as_bytes = REPLY_ENVELOPE
+        return _FakeResponse(reply.SerializeToString())
+
+
+async def _post_fixed_request() -> tuple[bytes, bytes]:
+    session = _FakeSession()
+    transport = V1rTransport(
+        host="192.168.91.1",
+        password="MyWiFiPassword123",
+        rsa_private_key_pem=TEST_KEY_PEM,
+        session=session,  # type: ignore[arg-type]
+    )
+    with (
+        patch("time.time", return_value=NOW),
+        patch("uuid.uuid4", return_value=uuid.UUID(bytes=MESSAGE_UUID)),
+    ):
+        inner = await transport.post_v1r(ENVELOPE, DIN)
+    assert len(session.posted) == 1
+    return session.posted[0], inner
+
+
+async def test_signed_request_matches_fixed_byte_vector() -> None:
+    posted, inner = await _post_fixed_request()
+
+    assert posted == EXPECTED_SIGNED_REQUEST
+    assert inner == REPLY_ENVELOPE
+
+
+async def test_fixed_vector_signature_verifies_with_12_second_ttl() -> None:
+    posted, _ = await _post_fixed_request()
+    message = signed_message_pb2.RoutableMessage()
+    message.ParseFromString(posted)
+
+    expires_at = 1_790_000_001 + 12
+    assert message.signature_data.rsa_data.expires_at == expires_at
+    assert message.uuid == MESSAGE_UUID
+    assert message.protobuf_message_as_bytes == ENVELOPE
+
+    key = load_pem_private_key(TEST_KEY_PEM, password=None)
+    assert isinstance(key, rsa.RSAPrivateKey)
+    personalization = DIN.encode("ascii")
+    signed_data = (
+        bytes([0, 1, 7])  # TAG_SIGNATURE_TYPE = SIGNATURE_TYPE_RSA
+        + bytes([1, 1, 7])  # TAG_DOMAIN = DOMAIN_ENERGY_DEVICE
+        + bytes([2, len(personalization)])
+        + personalization
+        + bytes([4, 4])
+        + expires_at.to_bytes(4, "big")
+        + bytes([255])  # TAG_END
+        + ENVELOPE
+    )
+    # Raises InvalidSignature on a mismatch.
+    key.public_key().verify(
+        message.signature_data.rsa_data.signature,
+        signed_data,
+        padding.PKCS1v15(),
+        hashes.SHA512(),
+    )