HMAC signing

HMAC-SHA256( parts.join("|"), secret )  →  hex digest (64 chars)

export function canonicalJsonStringify(obj: unknown): string {
  if (obj === null || obj === undefined) return JSON.stringify(obj);
  if (typeof obj !== 'object') return JSON.stringify(obj);
  if (Array.isArray(obj)) {
    return '[' + obj.map((item) => canonicalJsonStringify(item)).join(',') + ']';
  }
  const sortedKeys = Object.keys(obj as Record<string, unknown>).sort();
  const parts = sortedKeys.map(
    (key) =>
      JSON.stringify(key) + ':' + canonicalJsonStringify((obj as Record<string, unknown>)[key]),
  );
  return '{' + parts.join(',') + '}';
}

# apps/docs/content/examples/plc/hmac_sign.py
"""Canonical HMAC-SHA256 helpers for PLC message signing.

Mirrors packages/shared/src/utils/hmac.ts exactly. If you change one side,
change both and re-check the known-value tests.

Key facts (verified against the TypeScript source):
  - Separator:  "|"  (pipe, U+007C)
  - Telemetry:  deviceId | ts | nonce | canonical_json_stringify(data)
  - Command:    deviceId | cmdId | ts | action | canonical_json_stringify(payload)
  - Ack:        deviceId | cmdId | ts | status | nonce
  - Alarm:      deviceId | ts | nonce | ev | alarmId | code | sev

  The telemetry payload is the message body dict with ts/n/sig stripped out,
  serialised with sorted keys (recursive) and no spaces — matching
  canonicalJsonStringify() in hmac.ts.

Dependencies: Python 3.11+ standard library only.
"""
from __future__ import annotations

import hashlib
import hmac
import json
from typing import Any

# ─── Canonical JSON ───────────────────────────────────────────────────────────

def canonical_json_stringify(obj: Any) -> str:  # noqa: ANN401
    """Recursively serialise *obj* with sorted keys and no spaces.

    Mirrors canonicalJsonStringify() in packages/shared/src/utils/hmac.ts.
    Arrays are preserved in order; object keys are sorted lexicographically at
    every nesting level.
    """
    if obj is None:
        return "null"
    if isinstance(obj, bool):
        # Python's json.dumps renders True/False correctly, but we must handle
        # the bool-before-int check because bool is a subclass of int.
        return "true" if obj else "false"
    if isinstance(obj, (int, float)):
        # Use json.dumps for numbers so NaN/Infinity raise the same way as JS.
        return json.dumps(obj)
    if isinstance(obj, str):
        return json.dumps(obj)
    if isinstance(obj, list):
        return "[" + ",".join(canonical_json_stringify(item) for item in obj) + "]"
    if isinstance(obj, dict):
        sorted_keys = sorted(obj.keys())
        parts = [
            json.dumps(k) + ":" + canonical_json_stringify(obj[k])
            for k in sorted_keys
        ]
        return "{" + ",".join(parts) + "}"
    # Fallback — shouldn't appear in well-formed telemetry dicts.
    return json.dumps(obj)


# ─── Core primitives ─────────────────────────────────────────────────────────

# Separator between ordered parts. Must match the Node helper.
_SEP = "|"


def compute_signature(parts: list[str], secret: str) -> str:
    """HMAC-SHA256 hex digest over ``|``-joined UTF-8 parts.

    Mirrors computeSignature() in hmac.ts:
        const message = parts.join('|');
        return createHmac('sha256', secret).update(message).digest('hex');
    """
    message = _SEP.join(parts).encode("utf-8")
    return hmac.new(secret.encode("utf-8"), message, hashlib.sha256).hexdigest()


def verify_signature(parts: list[str], secret: str, signature: str) -> bool:
    """Constant-time comparison of computed vs supplied signature.

    Mirrors verifySignature() / timingSafeEqual() in hmac.ts.
    Returns False if the hex lengths differ (avoids a padding oracle).
    """
    expected = compute_signature(parts, secret)
    if len(expected) != len(signature):
        return False
    return hmac.compare_digest(expected, signature)


# ─── Domain signers ───────────────────────────────────────────────────────────

def sign_telemetry(
    device_id: str,
    ts: int,
    nonce: str,
    data: dict[str, Any],
    secret: str,
) -> str:
    """Sign a telemetry message.

    Part order (from computeTelemetrySignature in hmac.ts):
        deviceId | ts | nonce | canonicalJsonStringify(data)

    *data* is the message body dict with the envelope fields removed
    (``ts``, ``n``, ``nonce``, ``sig`` must be stripped before passing).
    The dict is serialised with sorted keys so signature is key-order invariant.

    Example::

        body = {k: v for k, v in raw.items() if k not in ("ts", "n", "nonce", "sig")}
        sig = sign_telemetry(plant_id, raw["ts"], raw["n"], body, secret)
    """
    return compute_signature(
        [device_id, str(ts), nonce, canonical_json_stringify(data)],
        secret,
    )


def verify_telemetry(
    device_id: str,
    ts: int,
    nonce: str,
    data: dict[str, Any],
    secret: str,
    signature: str,
) -> bool:
    """Verify a telemetry message signature (mirrors verifyTelemetrySignature)."""
    return verify_signature(
        [device_id, str(ts), nonce, canonical_json_stringify(data)],
        secret,
        signature,
    )


def sign_command(
    device_id: str,
    cmd_id: str,
    ts: int,
    action: str,
    payload: dict[str, Any],
    secret: str,
) -> str:
    """Sign an outbound command envelope (server → PLC).

    Part order (from computeCommandSignature in hmac.ts):
        deviceId | cmdId | ts | action | canonicalJsonStringify(payload)

    PLCs that verify incoming commands use this function to check Voke's
    signature on the ``PlcCommand`` envelope.
    """
    return compute_signature(
        [device_id, cmd_id, str(ts), action, canonical_json_stringify(payload)],
        secret,
    )


def sign_command_ack(
    device_id: str,
    cmd_id: str,
    ts: int,
    status: str,
    nonce: str,
    secret: str,
) -> str:
    """Sign a command ACK published to ``cpi/{deviceId}/ack``.

    Part order (from computeAckSignature in hmac.ts):
        deviceId | cmdId | ts | status | nonce

    *status* must be one of: RECEIVED | IN_PROGRESS | COMPLETED | FAILED
    """
    return compute_signature([device_id, cmd_id, str(ts), status, nonce], secret)


def verify_command_ack(
    device_id: str,
    cmd_id: str,
    ts: int,
    status: str,
    nonce: str,
    secret: str,
    signature: str,
) -> bool:
    """Verify a command ACK signature (mirrors verifyAckSignature)."""
    return verify_signature([device_id, cmd_id, str(ts), status, nonce], secret, signature)


def sign_alarm(
    device_id: str,
    ts: int,
    nonce: str,
    ev: str,
    alarm_id: str,
    code: int,
    sev: int,
    secret: str,
) -> str:
    """Sign a PLC alarm published to ``cpi/{deviceId}/alarm``.

    Part order (from computeAlarmSignature in hmac.ts):
        deviceId | ts | nonce | ev | alarmId | code | sev

    *ev* is one of: RAISE | RESOLVE
    *sev* is 1, 2, or 3.
    """
    return compute_signature(
        [device_id, str(ts), nonce, ev, alarm_id, str(code), str(sev)],
        secret,
    )


def verify_alarm(
    device_id: str,
    ts: int,
    nonce: str,
    ev: str,
    alarm_id: str,
    code: int,
    sev: int,
    secret: str,
    signature: str,
) -> bool:
    """Verify a PLC alarm signature (mirrors verifyAlarmSignature)."""
    return verify_signature(
        [device_id, str(ts), nonce, ev, alarm_id, str(code), str(sev)],
        secret,
        signature,
    )

SECRET = "test-secret-32-characters-long!!"

sig = compute_signature(["device-1", "1700000000000", "abc123"], SECRET)
assert len(sig) == 64              # 64-char hex
assert verify_signature(["device-1", "1700000000000", "abc123"], SECRET, sig)
assert not verify_signature(["device-1", "1700000000001", "abc123"], SECRET, sig)

assert canonical_json_stringify({"b": 2, "a": 1}) == '{"a":1,"b":2}'
assert canonical_json_stringify({"a": 1, "b": 2}) == '{"a":1,"b":2}'

data_ab = {"temperature": 22.5, "humidity": 60}
data_ba = {"humidity": 60, "temperature": 22.5}
s1 = sign_telemetry("device-abc", 1700000000000, "nonce-xyz", data_ab, SECRET)
s2 = sign_telemetry("device-abc", 1700000000000, "nonce-xyz", data_ba, SECRET)
assert s1 == s2   # key order must not affect the signature
assert verify_telemetry("device-abc", 1700000000000, "nonce-xyz", data_ab, SECRET, s1)

ack_sig = sign_command_ack(
    "device-1", "cmd-1", 1700000000000, "COMPLETED", "ack-nonce-xyz", SECRET
)
assert verify_command_ack(
    "device-1", "cmd-1", 1700000000000, "COMPLETED", "ack-nonce-xyz", SECRET, ack_sig
)
# Tampered status must fail
assert not verify_command_ack(
    "device-1", "cmd-1", 1700000000000, "FAILED", "ack-nonce-xyz", SECRET, ack_sig
)