Shelly authentication without challenge

shelly_auth_test.py

#!/usr/bin/env python3


from __future__ import annotations
import asyncio
import hashlib
import logging
import secrets
import time
from typing import Any

import aiohttp
from aioshelly.common import ConnectionOptions
from aioshelly.wsrpc import WsRPC

_LOGGER = logging.getLogger(__name__)

IP = "192.168.x.y"
PASSWORD = "asdfasdf"


def hex_hash(message: str) -> str:
    """
    Get hex representation of sha256 hash of string
    """
    return hashlib.sha256(message.encode("utf-8")).hexdigest()


HA2 = hex_hash("dummy_method:dummy_uri")


def get_auth(realm: str, password: str, nonce: str | None = None) -> dict[str, Any]:
    """
    https://mongoose-os.com/docs/mongoose-os/userguide/rpc.md#rpc-authentication-via-code

    https://github.com/mongoose-os-libs/rpc-common/blob/master/src/mg_rpc.c#L967
    """
    ha1 = hex_hash(f"admin:{realm}:{password}")
    cnonce = time.time()
    if nonce is None:
        nonce = cnonce
    hashed = hex_hash(f"{ha1}:{nonce}:1:{cnonce}:auth:{HA2}")
    auth = {
        "realm": realm,
        "username": "admin",
        "nonce": nonce,
        "cnonce": cnonce,
        "response": hashed,
        "algorithm": "SHA-256",
    }
    return auth


async def test_authentication(aiohtp_session: aiohttp.ClientSession) -> WsRPC:
    "Demonstrate digest auth with SHA-256 without getting a challenge first"

    def on_ntf(ntf):
        _LOGGER.info("Got ntf: %s", ntf)

    ch = WsRPC(IP, on_ntf)

    await ch.connect(aiohtp_session)
    device_info = await ch.call("Shelly.GetDeviceInfo")
    if not device_info["auth_en"]:
        _LOGGER.warning("Enable auth on device for this demo")
        return ch

    auth = get_auth(realm=device_info["auth_domain"], password=PASSWORD)
    ch.set_auth(auth)
    _LOGGER.info("Auth is %s", auth)
    await ch.call("Shelly.ListMethods")
    await ch.call("Shelly.GetStatus")

    return ch


async def test():
    "i hate this linter"
    async with aiohttp.ClientSession() as sess:
        ch = await test_authentication(sess)
        await ch.disconnect()


if __name__ == "__main__":
    logging.basicConfig(level=logging.DEBUG)
    loop = asyncio.get_event_loop_policy().get_event_loop()
    loop.run_until_complete(test())