fingerbot-cli/tests/test_lan_protocol.py

import pytest

from tuya_re import lan_protocol as lp
from tuya_re.const import LanCommand

KEY = b"0123456789abcdef"


def test_pack_control_matches_golden():
    body = lp.encode_v33_command(
        KEY,
        LanCommand.CONTROL,
        {
            "devId": "synthdev0001",
            "uid": "synthdev0001",
            "t": "100",
            "dps": {"1": True},
        },
    )
    frame = lp.pack_frame(seqno=1, command=LanCommand.CONTROL, body=body)
    assert frame.hex() == (
        "000055aa0000000100000007000000673"
        "32e3300000000000000000000000051027acc8d66f9d991bb545def2c54f98f2a1d63"
        "ed129ea6a1e08abe6427fed33fbd18005b18ec06894ddff50046fbee433c1a9df3e72"
        "0960094b19b5c30d031a603e43b8f00ae44bada4be6f66160ddd4de28550000aa55"
    )


def test_pack_dpquery_matches_golden():
    body = lp.encode_v33_command(
        KEY,
        LanCommand.DP_QUERY,
        {
            "gwId": "synthdev0001",
            "devId": "synthdev0001",
            "uid": "synthdev0001",
            "t": "100",
        },
    )
    frame = lp.pack_frame(seqno=2, command=LanCommand.DP_QUERY, body=body)
    assert frame.hex() == (
        "000055aa000000020000000a00000058d958003254c7c411ddff3f18fc25cad7982598"
        "d6c3e06d7668511179ee665bec29ec20bf47d8217234c0b4cab174f9f1c8cca11f4740"
        "beecff7c92d5d3c308fdc23cef0c529f97348f21b887ff7ae67c3956bdca0000aa55"
    )


def test_unpack_roundtrip_and_crc():
    body = lp.encode_v33_command(
        KEY, LanCommand.CONTROL, {"devId": "d", "dps": {"1": False}}
    )
    frame = lp.pack_frame(3, LanCommand.CONTROL, body)
    msg = lp.unpack_frame(frame)
    assert msg.command == LanCommand.CONTROL and msg.seqno == 3
    assert lp.decode_v33_payload(KEY, msg.payload) == {
        "devId": "d",
        "dps": {"1": False},
    }


def test_unpack_bad_crc_raises():
    body = lp.encode_v33_command(KEY, LanCommand.CONTROL, {"a": 1})
    frame = bytearray(lp.pack_frame(1, LanCommand.CONTROL, body))
    frame[20] ^= 0xFF
    with pytest.raises(lp.LanFrameError):
        lp.unpack_frame(bytes(frame))


def test_decode_reply_with_retcode():
    body = lp.encode_v33_command(KEY, LanCommand.DP_QUERY, {"dps": {"1": True}})
    reply = lp.pack_frame(2, LanCommand.DP_QUERY, body, retcode=0)
    msg = lp.unpack_frame(reply)
    assert lp.decode_v33_payload(KEY, msg.payload) == {"dps": {"1": True}}


def test_decode_empty_payload_is_empty_dict():
    assert lp.decode_v33_payload(KEY, b"") == {}


def test_derive_session_key_v34():
    assert (
        lp.derive_session_key_v34(
            b"0123456789abcdef", b"0123456789abcdef", b"abcdef0123456789"
        ).hex()
        == "465ad66115cf0cb58482681d57f0133b"
    )


def test_pack_hmac_frame_start_golden():
    REAL = b"0123456789abcdef"
    from tuya_re import crypto

    body = crypto.aes_ecb_encrypt(REAL, REAL)  # ECB(real, local_nonce)
    frame = lp.pack_frame(1, LanCommand.SESS_KEY_NEG_START, body, hmac_key=REAL)
    assert frame.hex() == (
        "000055aa0000000100000003000000447272"
        "7e881edcfd0100a718687909b565377222e061a924c591cd9c27ea163ed4d7dd1f9e"
        "b60b8cb5b90748aa228534c62460d05b84665c9e692efc896dd77d680000aa55"
    )


def test_unpack_hmac_frame_verifies():
    REAL = b"0123456789abcdef"
    resp_hex = (
        "000055aa0000000200000004000000640f9be99491cb21d2e042af13e00ce19"
        "734a8706148aba7b9e47b4fd03c3428475216353fe75fa1cb7918233866ab9feb377222"
        "e061a924c591cd9c27ea163ed416389ab8912c499dc37aa85108337e419fb2469e93c62"
        "d807367617653a450a40000aa55"
    )
    msg = lp.unpack_frame(bytes.fromhex(resp_hex), hmac_key=REAL)
    assert msg.command == LanCommand.SESS_KEY_NEG_RESP
    # decrypt body with REAL, recover remote_nonce + hmac
    from tuya_re import crypto

    dec = crypto.aes_ecb_decrypt(REAL, msg.payload)
    assert dec[:16] == b"abcdef0123456789"


def test_unpack_hmac_bad_mac_raises():
    REAL = b"0123456789abcdef"
    from tuya_re import crypto

    frame = bytearray(
        lp.pack_frame(
            1, LanCommand.CONTROL, crypto.aes_ecb_encrypt(REAL, b"x"), hmac_key=REAL
        )
    )
    frame[-6] ^= 0xFF  # corrupt inside the hmac
    with pytest.raises(lp.LanFrameError):
        lp.unpack_frame(bytes(frame), hmac_key=REAL)


def test_encode_v34_control_frame_golden():
    SESSION = bytes.fromhex("465ad66115cf0cb58482681d57f0133b")
    body = lp.encode_command(
        SESSION,
        LanCommand.CONTROL,
        {
            "devId": "synthdev0001",
            "uid": "synthdev0001",
            "t": "100",
            "dps": {"1": True},
        },
        version="3.4",
    )
    frame = lp.pack_frame(4, LanCommand.CONTROL, body, hmac_key=SESSION)
    assert frame.hex() == (
        "000055aa00000004000000070000008464"
        "5421669e0a4da19c83e1c3bdc0368d7828489e443086f21b978782f671cb651a270cc3"
        "01b9635c773070fb8985b8006bebceb8c013a303b61c9729da66a9132643ca448149ef"
        "67b0ada24a728d4365132dee5659b3f58f65ded80ac7caeded60fe7b6d665c26d815ff"
        "f0ba45c252a69a6276f7f9dae56475cd0f3a59de324f0000aa55"
    )


def test_decode_v34_payload_strips_header():
    SESSION = bytes.fromhex("465ad66115cf0cb58482681d57f0133b")

    body = lp.encode_command(
        SESSION, LanCommand.CONTROL, {"dps": {"1": False}}, version="3.4"
    )
    assert lp.decode_payload(SESSION, body, version="3.4") == {"dps": {"1": False}}