Add signed Parallax lock/unlock commands

README.md

@@ -50,3 +50,44 @@ editing one, regenerate its `*_pb2` modules (don't edit those by hand):
 uv run python -m grpc_tools.protoc --proto_path=src --python_out=src \
     --pyi_out=src src/rivian/parallax/proto/*.proto
 ```
+
+### Commands
+
+`send_parallax_command` sends a signed command, such as unlocking
+(`parallax.UNLOCK_RVM` with `parallax.encode_unlock()`; `LOCK_RVM` and
+`encode_lock()` to lock), and `subscribe_for_parallax_command_state` follows it.
+Subscribe first, then send:
+
+```python
+import uuid
+
+from rivian import parallax
+
+
+def on_state(data):
+    if state := parallax.decode_command_state_message(data):
+        print(state["progress"], state["status"], state["error"])
+
+
+command_id = str(uuid.uuid4())
+unsubscribe = await rivian.subscribe_for_parallax_command_state(
+    vehicle_id, command_id, on_state
+)
+if unsubscribe:
+    await rivian.send_parallax_command(
+        vehicle_id,
+        parallax.UNLOCK_RVM,
+        parallax.encode_unlock(),
+        identity_id=identity_id,
+        vehicle_key=vehicle_key,
+        private_key=private_key,
+        command_id=command_id,
+    )
+    ...  # wait for the progress "complete", or give up
+    await unsubscribe()
+```
+
+The keys are those of `send_vehicle_command`. The command is done when the progress
+is `complete`, the status `success` and there is no `error`. If you never see that,
+check the vehicle before sending it again. This unlocks a real vehicle, so keep the
+keys secret.

src/rivian/parallax/__init__.py

@@ -10,9 +10,15 @@
     encode_vehicle_operation(rvm, payload=b"", ...) -> str
         Builds a `send_vehicle_operation` request; an empty payload reads
         the RVM, anything else (e.g. `encode_climate_hold_setting`) writes it.
+    encode_command(vehicle_id, rvm, payload, ...) -> str
+        Builds and signs a `send_parallax_command` request, with a payload
+        such as `encode_unlock()`.
+    decode_command_state_message(data) -> dict | None
+        Reads a raw `subscribe_for_parallax_command_state` message.
 
 Layout:
-    core.py  - registry, payload parsing and shared helpers
+    core.py    - registry, payload parsing and shared helpers
+    command.py - signed remote commands and their state messages
     proto/   - .proto schemas and their generated *_pb2 modules
     <topic prefix>.py - one module per RVM topic prefix (body.*, etc.)
 
@@ -44,11 +50,13 @@
     vehicle,
     vehicle_access,
 )
+from .body import LOCK_RVM, UNLOCK_RVM, encode_lock, encode_unlock
 from .comfort import (
     CLIMATE_HOLD_SETTING_RVM,
     CLIMATE_HOLD_STATUS_RVM,
     encode_climate_hold_setting,
 )
+from .command import decode_command_state_message, encode_command
 from .core import (
     RVMDecoder,
     decode_parallax_message,
@@ -75,9 +83,15 @@
     "CHARGING_RVMS",
     "CLIMATE_HOLD_SETTING_RVM",
     "CLIMATE_HOLD_STATUS_RVM",
+    "LOCK_RVM",
     "PARALLAX_RVMS",
+    "UNLOCK_RVM",
+    "decode_command_state_message",
     "decode_parallax_message",
     "decode_parallax_subscription_message",
     "encode_climate_hold_setting",
+    "encode_command",
+    "encode_lock",
+    "encode_unlock",
     "encode_vehicle_operation",
 ]

src/rivian/parallax/body.py

@@ -7,6 +7,9 @@
 from .core import _LOGGER, RVMDecoder, _enum, _present
 from .proto import body_pb2
 
+LOCK_RVM: Final = "body.locks.lock"
+UNLOCK_RVM: Final = "body.locks.unlock"
+
 _CLOSURE_MAP: Final[dict[int, str]] = {
     body_pb2.DOOR_FRONT_LEFT: "doorFrontLeftClosed",
     body_pb2.DOOR_FRONT_RIGHT: "doorFrontRightClosed",
@@ -173,6 +176,16 @@ def decode_closures(m: body_pb2.ClosuresState) -> dict[str, Any]:
     return result
 
 
+def encode_lock() -> bytes:
+    """Encode a body.locks.lock payload, which has no fields."""
+    return body_pb2.LockCommand().SerializeToString()
+
+
+def encode_unlock() -> bytes:
+    """Encode a body.locks.unlock payload that unlocks everything."""
+    return body_pb2.UnlockCommand(all=True).SerializeToString()
+
+
 @RVMDecoder.register("body.locks.states", body_pb2.LocksState)
 def decode_locks(m: body_pb2.LocksState) -> dict[str, Any]:
     """body.locks.states — lock state of every lock.

src/rivian/parallax/command.py

@@ -0,0 +1,168 @@
+"""Encoder and decoder for signed Parallax remote commands.
+
+A command such as `body.locks.unlock` is sent with the `sendParallaxPayload`
+mutation (`Rivian.send_parallax_command`). Its payload is an `Operation` in an
+`Envelope`, signed with the secret that `Rivian.send_vehicle_command` derives from
+the phone's key and the vehicle's, but over the operation's bytes. In testing,
+Rivian woke a sleeping vehicle for it.
+
+The `vehicleCommandStatePx` subscription
+(`Rivian.subscribe_for_parallax_command_state`) reports on the command as it
+goes: the vehicle waking up, validating and executing it, then the result.
+"""
+
+from __future__ import annotations
+
+import base64
+import time
+import uuid
+from typing import Any, Final
+
+from google.protobuf.message import DecodeError
+
+from ..utils import generate_ble_command_hmac
+from .core import _LOGGER, _debug_message, _enum, _present
+from .proto import command_pb2
+
+_PROGRESS_MAP: Final[dict[int, str]] = {
+    command_pb2.VEHICLE_MODEL_OP_PROGRESS_UNSPECIFIED: "unspecified",
+    command_pb2.VEHICLE_MODEL_OP_PROGRESS_VEHICLE_WAKING_UP: "waking_up",
+    command_pb2.VEHICLE_MODEL_OP_PROGRESS_VEHICLE_VALIDATING: "validating",
+    command_pb2.VEHICLE_MODEL_OP_PROGRESS_VEHICLE_EXECUTING: "executing",
+    command_pb2.VEHICLE_MODEL_OP_PROGRESS_VEHICLE_COMPLETE: "complete",
+}
+
+_STATUS_MAP: Final[dict[int, str]] = {
+    command_pb2.RESPONSE_STATUS_UNDEFINED: "undefined",
+    command_pb2.RESPONSE_STATUS_SUCCESS: "success",
+    command_pb2.RESPONSE_STATUS_FAIL: "failed",
+    command_pb2.RESPONSE_STATUS_INVALID_REQUEST: "invalid_request",
+    command_pb2.RESPONSE_STATUS_NOT_SUPPORTED: "not_supported",
+    command_pb2.RESPONSE_STATUS_TIMEOUT: "timeout",
+}
+
+
+def encode_command(
+    vehicle_id: str,
+    rvm: str,
+    payload: bytes,
+    *,
+    identity_id: str,
+    vehicle_key: str,
+    private_key: str,
+    command_id: str | None = None,
+    timestamp: float | None = None,
+) -> str:
+    """Encode and sign a command, as the payload of `sendParallaxPayload`.
+
+    `payload` is the RVM's message, e.g. `encode_unlock()` for `UNLOCK_RVM`;
+    nothing checks that they go together. `identity_id`, `vehicle_key` and
+    `private_key` are the ones `send_vehicle_command` takes. The command id, a
+    UUID, is a new one unless `command_id` is given (`ValueError` if it isn't
+    one); `timestamp` is epoch milliseconds, now unless given.
+
+    The result is signed and can be replayed until the vehicle rejects its
+    timestamp, so treat it as a secret.
+    """
+    command_id_bytes = (
+        uuid.uuid4().bytes if command_id is None else uuid.UUID(command_id).bytes
+    )
+    timestamp_ms = int(time.time() * 1000 if timestamp is None else timestamp)
+    stamp = command_pb2.Timestamp(
+        seconds=timestamp_ms // 1000, nanos=timestamp_ms % 1000 * 1_000_000
+    )
+    operation = command_pb2.Operation(
+        rvm=rvm,
+        type=command_pb2.PARALLAX_OPERATION_REQUEST,
+        command_id=command_id_bytes,
+        payload=payload,
+        created_at=stamp,
+        vehicle_id=vehicle_id,
+    ).SerializeToString()
+    envelope = command_pb2.Envelope(
+        composite=command_pb2.Composite(bytes_val=operation),
+        last_update=stamp,
+        correlation_id=command_id_bytes,
+        security_info=command_pb2.SecurityInfo(
+            signature=command_pb2.Signature(
+                type=command_pb2.DATA_SIGNATURE_TYPE_HMAC_SHA256,
+                # The raw HMAC-SHA256 of the data, as BLE commands use too
+                value=generate_ble_command_hmac(operation, vehicle_key, private_key),
+            ),
+            mapped_identity=identity_id,
+        ),
+    )
+    return base64.b64encode(envelope.SerializeToString()).decode("utf-8")
+
+
+def decode_command_state(payload: str) -> dict[str, Any] | None:
+    """Decode the payload of a `vehicleCommandStatePx` message.
+
+    Returns None if the payload is empty or can't be read.
+
+    Fields:
+        commandId: str | None — the command's UUID
+        rvm: str
+        progress: str | int — "unspecified" | "waking_up" | "validating" |
+            "executing" | "complete". Any of the ones before "complete" can be
+            skipped, and "waking_up" has come more than once for one command.
+        status: str | int | None — how the vehicle answered ("undefined" |
+            "success" | "failed" | "invalid_request" | "not_supported" |
+            "timeout"); None if the message has no result
+        error: dict | None — {"code": int, "message": str}, from the result or,
+            if the whole operation failed, from the operation error (which has
+            no status)
+
+    A value this library doesn't know comes back as the number it was sent as.
+
+    The vehicle has done the command when the progress is "complete", the status
+    "success" and there is no error. "failed", "invalid_request" and
+    "not_supported" mean it has not. With "timeout" or "undefined", or if no
+    "complete" message comes at all, it isn't known, and the vehicle may do it
+    later.
+    """
+    try:
+        data = base64.b64decode(payload)
+        message = command_pb2.CommandState.FromString(data)
+    except (TypeError, ValueError, DecodeError):
+        _LOGGER.debug("Failed to decode command state", exc_info=True)
+        return None
+    if not data:
+        return None
+    _debug_message("command state", message)
+
+    status: str | int | None = None
+    error: dict[str, Any] | None = None
+    if (result := _present(message, "data")) is not None:
+        status = _enum(_STATUS_MAP, result.response_status, what="command status")
+        if (result_error := _present(result, "error")) is not None:
+            error = {"code": result_error.code, "message": result_error.message}
+    elif (operation_error := _present(message, "operation_error")) is not None:
+        error = {"code": operation_error.error, "message": operation_error.message}
+    return {
+        "commandId": (
+            str(uuid.UUID(bytes=message.command_id))
+            if len(message.command_id) == 16
+            else None
+        ),
+        "rvm": message.rvm,
+        "progress": _enum(_PROGRESS_MAP, message.progress, what="command progress"),
+        "status": status,
+        "error": error,
+    }
+
+
+def decode_command_state_message(data: dict[str, Any]) -> dict[str, Any] | None:
+    """Decode a raw `subscribe_for_parallax_command_state` message.
+
+    Unwraps `payload.data.vehicleCommandStatePx.payload` for
+    `decode_command_state`; returns None if it's missing or can't be read. That
+    includes a message that carries GraphQL errors (`data["payload"]["errors"]`)
+    instead of a state.
+    """
+    payload: Any = data
+    for key in ("payload", "data", "vehicleCommandStatePx", "payload"):
+        if not isinstance(payload, dict):
+            return None
+        payload = payload.get(key)
+    return decode_command_state(payload)

src/rivian/parallax/core.py

@@ -76,6 +76,16 @@ def _describe_unknowns(message: Message, path: str = "") -> list[str]:
     return found
 
 
+def _debug_message(label: str, message: Message) -> None:
+    """Debug-log a parsed `message` and anything its schema doesn't define."""
+    if _LOGGER.isEnabledFor(logging.DEBUG):
+        _LOGGER.debug(
+            "%s: %s", label, text_format.MessageToString(message, as_one_line=True)
+        )
+        for unknown in _describe_unknowns(message):
+            _LOGGER.debug("%s: %s", label, unknown)
+
+
 class RVMDecoder:
     """Registry of RVM topic -> (message class, decoder).
 
@@ -124,12 +134,7 @@ def _decode_rvm_payload(payload: str | None, rvm: str) -> dict[str, Any]:
     except Exception:
         _LOGGER.debug("Failed to decode %s payload", rvm, exc_info=True)
         return {}
-    if _LOGGER.isEnabledFor(logging.DEBUG):
-        _LOGGER.debug(
-            "%s: %s", rvm, text_format.MessageToString(message, as_one_line=True)
-        )
-        for unknown in _describe_unknowns(message):
-            _LOGGER.debug("%s: %s", rvm, unknown)
+    _debug_message(rvm, message)
     return result
 
 

src/rivian/parallax/proto/body.proto

@@ -266,3 +266,14 @@ enum WiperFluidState {
   WIPER_FLUID_STATE_LOW = 2;
   WIPER_FLUID_STATE_EMPTY = 3;
 }
+
+// body.locks.unlock, the payload of a signed command (see ../command.py)
+message UnlockCommand {
+  oneof target {
+    bool all = 1; // the app always sets this
+    bool user_preference = 2; // in the app's schema; its button never sends it
+  }
+}
+
+// body.locks.lock, the payload of a signed command. It has no fields.
+message LockCommand {}