Reverse Engineering the Samsung EHS Cloud

import struct

def encode_option(delta, value):
    """Encode a single CoAP option."""
    vlen = len(value)
    d = delta if delta < 13 else (13 if delta < 269 else 14)
    d_ext = (b'' if delta < 13
             else struct.pack('B', delta - 13) if delta < 269
             else struct.pack('>H', delta - 269))
    l = vlen if vlen < 13 else (13 if vlen < 269 else 14)
    l_ext = (b'' if vlen < 13
             else struct.pack('B', vlen - 13) if vlen < 269
             else struct.pack('>H', vlen - 269))
    return struct.pack('B', (d << 4) | l) + d_ext + l_ext + value


def encode_coap_tcp(code, token=b'', options=None, payload=b''):
    """Encode a CoAP-over-TCP message (RFC 8323).

    Args:
        code: CoAP method code (1=GET, 2=POST, 225=CSM)
        token: Message token bytes
        options: List of (option_number, value_bytes) tuples
        payload: CBOR payload bytes
    """
    if options is None:
        options = []
    opt_bytes = b''
    prev = 0
    for num, val in options:
        opt_bytes += encode_option(num - prev, val)
        prev = num
    tkl = len(token)
    body_len = len(opt_bytes) + (1 + len(payload) if payload else 0)
    if body_len < 13:
        hdr = struct.pack('BB', (body_len << 4) | tkl, code)
    elif body_len < 269:
        hdr = struct.pack('BBB', (13 << 4) | tkl, body_len - 13, code)
    elif body_len < 65805:
        hdr = struct.pack('>BHB', (14 << 4) | tkl, body_len - 269, code)
    else:
        hdr = struct.pack('>BIB', (15 << 4) | tkl, body_len - 65805, code)
    return hdr + token + opt_bytes + (b'\xff' + payload if payload else b'')

import struct

def parse_coap_tcp(data):
    """Parse a CoAP-over-TCP message.

    Returns dict with 'code' (e.g. '2.04'), 'token', 'payload',
    and 'total_len' (for parsing multiple messages from a buffer).
    """
    if len(data) < 2:
        return None
    first = data[0]
    ln = (first >> 4) & 0xF
    tkl = first & 0xF
    off = 1
    if ln < 13:
        ml = ln
    elif ln == 13:
        ml = data[off] + 13; off += 1
    elif ln == 14:
        ml = struct.unpack('>H', data[off:off+2])[0] + 269; off += 2
    else:
        ml = struct.unpack('>I', data[off:off+4])[0] + 65805; off += 4
    code = data[off]; off += 1
    tok = data[off:off+tkl]; off += tkl
    rem = data[off:off+ml]
    pay = b''
    for i in range(len(rem)):
        if rem[i] == 0xFF:
            pay = rem[i+1:]
            break
    return {
        'code': f"{(code >> 5) & 7}.{code & 0x1F:02d}",
        'token': tok,
        'payload': pay,
        'total_len': off + ml,
    }

import asyncio
import ssl
import struct
import cbor2

CI_HOST = "ocfclientcon-shard-eu02s-euwest1.samsungiots.com"
CI_PORT = 443


async def cloud_signin(uid, device_id, access_token):
    """Sign in to Samsung CI server via CoAP-over-TLS.

    Args:
        uid: Samsung Account GUID (e.g. '7wbtx1jj6j')
        device_id: Client device cloud ID (UUID)
        access_token: Samsung Account OAuth access token
    """
    # TLS connection (server-only, no client cert needed)
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE

    reader, writer = await asyncio.open_connection(
        CI_HOST, CI_PORT, ssl=ctx
    )

    # Step 1: CSM (Capabilities and Settings Message)
    # Option 2 = Max-Message-Size, value = 1152
    csm = encode_coap_tcp(225, options=[
        (2, struct.pack('>I', 1152))
    ])
    writer.write(csm)
    await writer.drain()
    try:
        await asyncio.wait_for(reader.read(16384), timeout=3.0)
    except asyncio.TimeoutError:
        pass  # CSM response may not come; continue anyway

    # Step 2: Cloud Sign-In
    # POST /oic/account/session
    # Options: 11=Uri-Path, 12=Content-Format (60=CBOR)
    payload = cbor2.dumps({
        "uid": uid,
        "di": device_id,
        "accesstoken": access_token,
        "login": True,
    })
    signin_msg = encode_coap_tcp(2, token=b'\x01', options=[
        (11, b'oic'), (11, b'account'), (11, b'session'),
        (12, struct.pack('>H', 60)),  # application/cbor
    ], payload=payload)
    writer.write(signin_msg)
    await writer.drain()

    # Parse response
    data = await asyncio.wait_for(reader.read(16384), timeout=10.0)
    msg = parse_coap_tcp(data)
    if msg and msg['code'] == '2.04':
        result = cbor2.loads(msg['payload'])
        print(f"Sign-in OK! Expires in {result['expiresin']}s")
        return True

    print(f"Sign-in failed: {msg['code'] if msg else 'no response'}")
    writer.close()
    await writer.wait_closed()
    return False

import struct
import uuid
import cbor2


async def cloud_signup(uid, access_token, ci_host, ci_port=443):
    """Register a new client device with the CI server.

    Args:
        uid: Samsung Account GUID
        access_token: Samsung Account OAuth access token
        ci_host: CI server hostname
        ci_port: CI server port (default 443)

    Returns dict with certificate, redirecturi, sid, etc.
    """
    device_id = str(uuid.uuid4())  # generate fresh client ID

    # Connect and exchange CSM (same as sign-in)
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    reader, writer = await asyncio.open_connection(
        ci_host, ci_port, ssl=ctx
    )
    writer.write(encode_coap_tcp(225, options=[
        (2, struct.pack('>I', 1152))
    ]))
    await writer.drain()
    try:
        await asyncio.wait_for(reader.read(16384), timeout=3.0)
    except asyncio.TimeoutError:
        pass

    # POST /oic/account?apiVersion=v2
    # Option 15 = Uri-Query
    payload = cbor2.dumps({
        "di": device_id,
        "devicetype": "samsungconnect",
        "clientid": "6iado3s6jc",
        "authprovider": "https://eu-auth2.samsungosp.com",
        "accesstoken": access_token,
        "uid": uid,
    })
    writer.write(encode_coap_tcp(2, token=b'\x01', options=[
        (11, b'oic'), (11, b'account'),
        (12, struct.pack('>H', 60)),  # application/cbor
        (15, b'apiVersion=v2'),       # URI-Query
    ], payload=payload))
    await writer.drain()

    data = await asyncio.wait_for(reader.read(16384), timeout=15.0)
    msg = parse_coap_tcp(data)
    result = cbor2.loads(msg['payload'])

    # Response includes:
    #   certificate  - PEM client cert (issued by CI server)
    #   redirecturi  - shard URL for future connections
    #   sid          - server ID
    #   accesstoken  - CI server access token
    #   expiresin    - token expiry in seconds

    writer.close()
    await writer.wait_closed()
    return {**result, "di": device_id}

import requests

TOKEN_URL = "https://eu-auth2.samsungosp.com/auth/oauth2/token"
CLIENT_ID = "6iado3s6jc"


def refresh_samsung_token(refresh_token):
    """Exchange a Samsung Account refresh token for fresh tokens.

    The refresh token has a ~2 year rolling expiry (63,072,000 seconds).
    Each refresh returns a new refresh_token, so tokens can be
    maintained indefinitely.

    Returns:
        dict with access_token, refresh_token, userId (GUID),
        expires_in, etc.
    """
    resp = requests.post(TOKEN_URL, data={
        "grant_type": "refresh_token",
        "refresh_token": refresh_token,
        "client_id": CLIENT_ID,
    })
    resp.raise_for_status()
    data = resp.json()
    return {
        "access_token": data["access_token"],
        "refresh_token": data["refresh_token"],
        "uid": data["userId"],           # Samsung Account GUID
        "expires_in": data["expires_in"], # ~86400 (24 hours)
    }

import cbor2


async def read_ocf_resource(writer, reader, device_id, resource_path):
    """Read an OCF resource from a heat pump via CI server routing.

    Args:
        writer: asyncio StreamWriter (already signed in)
        reader: asyncio StreamReader
        device_id: Heat pump OCF device UUID
        resource_path: e.g. ['temperatures', 'indoor', 'vs', '0']

    Returns:
        dict of resource attributes, or None on failure.

    Example response for /temperatures/indoor/vs/0:
    {
        'x.com.samsung.da.offset': '0.0',
        'x.com.samsung.da.desiredTemp': 27.9,
        'x.com.samsung.da.currentTemp': 22.5,
        'x.com.samsung.da.minTemp': 15.0,
        'x.com.samsung.da.maxTemp': 55.0,
        'x.com.samsung.da.tempIncrement': 0.5,
        'x.com.samsung.da.tempUnit': 'C',
    }
    """
    # Build URI path: /oic/route/{device_id}/{resource_path}
    options = [
        (11, b'oic'), (11, b'route'),
        (11, device_id.encode()),
    ]
    for segment in resource_path:
        options.append((11, segment.encode()))

    # CoAP GET (code=1)
    writer.write(encode_coap_tcp(1, token=b'\x03', options=options))
    await writer.drain()

    try:
        data = await asyncio.wait_for(reader.read(16384), timeout=15.0)
    except asyncio.TimeoutError:
        return None

    msg = parse_coap_tcp(data)
    if msg and msg['payload']:
        return cbor2.loads(msg['payload'])
    return None