dexcom_poc.py

#!/usr/bin/env python3
from __future__ import annotations
import secrets
import gzip
import json
import uuid
import requests
from dataclasses import dataclass, asdict, is_dataclass, field
from Crypto.PublicKey import RSA
from Crypto.Cipher import AES, PKCS1_v1_5
from Crypto.Hash import HMAC, SHA256
from Crypto.Util.Padding import pad
from base64 import urlsafe_b64encode
from datetime import datetime


class DexcomAPI:

    BASE_URL = "https://shareous1.dexcom.com/ShareWebServices/Services"
    RSA_PUB = """-----BEGIN PUBLIC KEY-----
                MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAhR1op97RL74ZPlwmTCz/
                gqJBmezl9XnEd9+qGtBadd9GP8yxTvhxysxS35Hs0vdRty7/uvOiG26Bmy2NAsUw
                xaaTy9Jf7Knceg4Zb5HpxcZR7Oku7RBuP9wTqDvLw/DLWIpq/n3norwwfZ5kQtB2
                Q6n/WN6DS6dkJvWozXJS1moBoN66znX3jJDMaq8KSW6xOg1tBPoA7ki3Kgb/NeO8
                xspYhWtjuC7HHxI5O+1elaGgs+Bb5qB2ctKqs909gtcrH62Vo+CdeMVdOHlluaTP
                TwudnaVu5zSu0ubcMyca0I4O8IloPJT3buExc2iP4uZtN3lfpjft7PGXAp95QMS4
                1wIDAQAB
                -----END PUBLIC KEY-----"""

    def __init__(self, accid, password, devicekey):
        self._appid = "d89443d2-327c-4a6f-89e5-496bbb0317db"
        self._accid = accid
        self._password = password
        self._devicekey = devicekey
        self._session = requests.session()
        self._session.headers.update(
            {
                "User-Agent": "Dexcom%20Follow/3311 CFNetwork/1406.0.4 Darwin/22.4.0",
                "Content-Type": "application/json",
            }
        )

    @dataclass
    class BaseDC:
        def __post_init__(self):
            try:
                self.Timestamp = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
                self.RequestId = str(uuid.uuid4())
            except AttributeError:
                pass

    @dataclass
    class EncryptionKey:
        KeyId: int
        Key: str
        IV: str

    @dataclass
    class SignedRequestHeader(BaseDC):
        AccId: str
        AppId: str
        EncKey: EncryptionKey
        IsZip: int
        Timestamp: str = field(init=False)

    @dataclass
    class SignedRequestHeaderUnencrypted(BaseDC):
        AccId: str
        AppId: str
        IsZip: int
        Timestamp: str = field(init=False)

    @dataclass
    class KeyRequest:
        KeyRequest: KeyRequestBody

    @dataclass
    class KeyRequestBody(BaseDC):
        AppId: str
        Password: str
        Key: str
        RequestId: str = field(init=False)
        Timestamp: str = field(init=False)
        KeyHmac: str

    def _b64enc(self, data, gz=False):
        if isinstance(data, str):
            data = data.encode()
        if gz:
            data = gzip.compress(data)
        return urlsafe_b64encode(data).decode("utf-8").replace("=", "")

    def _rsa_enc(self, data):
        cipher = PKCS1_v1_5.new(RSA.importKey(self.RSA_PUBKEY))
        return self._b64enc(cipher.encrypt(data))

    def _aes_enc(self, data, key, iv):
        cipher = AES.new(key, AES.MODE_CBC, iv)
        return self._b64enc(cipher.encrypt(pad(data, AES.block_size)))

    def _hmac(self, data, secret):
        h = HMAC.new(secret.encode(), digestmod=SHA256)
        h.update(data.encode())
        return self._b64enc(h.digest()[:16])

    def _uuid(self, uuid1, uuid2):
        tmp1 = self._hmac(uuid2, uuid1 + uuid2)
        tmp2 = self._hmac(uuid2, uuid1 + tmp1)
        return uuid1 + tmp2

    def _tojson(self, data):
        if is_dataclass(data):
            data = asdict(data)
        return json.dumps(data, separators=(",", ":"))

    def _jwt(self, obj, data, uuid):
        jwt = "{}.{}".format(self._b64enc(self._tojson(obj)), data)
        jwt += ".{}".format(self._hmac(jwt, self._uuid(self._accid, uuid)))
        return f'"{jwt}"'

    def gen_payload(self, data, enc=False):
        if not isinstance(data, str):
            data = self._tojson(data)
        iszip = 1 if len(data) > 500 else 0
        if enc:
            aes_key = secrets.token_bytes(16)
            aes_iv = secrets.token_bytes(16)
            return self._jwt(
                self.SignedRequestHeader(
                    self._accid,
                    self._appid,
                    self.EncryptionKey(
                        101,
                        self._rsa_enc(aes_key),
                        self._rsa_enc(aes_iv)
                    ),
                    iszip,
                ),
                self._aes_enc(self._b64enc(data, iszip).encode(), aes_key, aes_iv),
                self._devicekey,
            )
        else:
            return self._jwt(
                self.SignedRequestHeaderUnencrypted(
                    self._accid,
                    self._appid,
                    iszip
                ),
                self._b64enc(data, iszip),
                self._devicekey,
            )

    def req_post(self, endpoint, enc=False, data={}):
        res = self._session.post(
            f"{self.BASE_URL}{endpoint}", self.gen_payload(data, enc)
        )
        return res.json()

    def devicekey_request(self):
        req = self.KeyRequest(
            self.KeyRequestBody(
                self._appid,
                self._password,
                self._devicekey,
                self._hmac(self._devicekey, self._accid + self._devicekey),
            )
        )
        return self.req_post("/Subscriber/DeviceKeys", req, True)


def main():
    # accid & password from CreateSubscriberAccount3
    # devicekey should be random - but consistent once posted to DeviceKeys
    dx = DexcomAPI(
        accid="CHANGEME",
        password="CHANGEME",
        devicekey="00000000-0000-0000-0000-00000000000",
    )
    # res = dx.devicekey_request()
    # print(res)
    res = dx.req_post("/Subscriber/ReadSubscriber2")
    print(json.dumps(res, indent=4, sort_keys=True))

if __name__ == "__main__":
    main()