From 5546c0313a815de3041f1f1448153d8c0118ee59 Mon Sep 17 00:00:00 2001 From: argrento Date: Sun, 4 Jun 2023 14:19:19 +0000 Subject: [PATCH] Enable CI checks (#72) Reviewed-on: https://codeberg.org/argrento/huami-token/pulls/72 --- .gitignore | 1 + .travis.yml | 15 ----- .woodpecker.yml | 32 ++++++++++ README.md | 2 +- huami_token.py | 139 +++++++++++++++++++++++------------------- requirements.txt | 1 + tests/__init__.py | 0 tests/test_amazfit.py | 23 +++++++ urls.py | 38 ++++++------ 9 files changed, 153 insertions(+), 98 deletions(-) delete mode 100644 .travis.yml create mode 100644 .woodpecker.yml create mode 100644 tests/__init__.py create mode 100644 tests/test_amazfit.py diff --git a/.gitignore b/.gitignore index ef7ebb7..e9b78c9 100644 --- a/.gitignore +++ b/.gitignore @@ -7,3 +7,4 @@ __pycache__ *.fw *.bin venv +.mypy_cache \ No newline at end of file diff --git a/.travis.yml b/.travis.yml deleted file mode 100644 index 5301cb9..0000000 --- a/.travis.yml +++ /dev/null @@ -1,15 +0,0 @@ -language: python -python: - - "3.6" - -# Install dependencies. -install: - - pip install -r requirements.txt - -# Run linting and tests. -script: - - pytest --pylint - -# Turn email notifications off. -notifications: - email: false diff --git a/.woodpecker.yml b/.woodpecker.yml new file mode 100644 index 0000000..bcdea79 --- /dev/null +++ b/.woodpecker.yml @@ -0,0 +1,32 @@ +pipeline: + style_check: + image: python:3.9-buster +# when: +# event: pull_request + commands: + - python -m pip install --upgrade pip + - python -m pip install -r requirements.txt + - python -m pip install pylint flake8 mypy>=0.971 + - python -m flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics + - mypy --strict ./ + - python -m pylint -f parseable ./*.py + + unit_tests: + image: python:${TAG}-buster +# when: +# event: pull_request + commands: + - ls + - python -m venv venv + - /bin/bash -c "source venv/bin/activate" + - python -m pip install --upgrade pip + - python -m pip install -r requirements.txt + - pytest tests/ + secrets: [ amazfit_email, amazfit_password ] + +matrix: + TAG: + - 3.8 + - 3.9 + - 3.10 + - 3.11 diff --git a/README.md b/README.md index b62d6f9..b7474fd 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,7 @@ Huami-token is now hosted on [codeberg.org](https://codeberg.org/argrento/huami- # Huami-token -[![Build Status](https://travis-ci.org/argrento/huami-token.svg?branch=master)](https://travis-ci.org/argrento/huami-token) +[![status-badge](https://ci.codeberg.org/api/badges/argrento/huami-token/status.svg)](https://ci.codeberg.org/argrento/huami-token) Script to obtain watch or band bluetooth access token from Huami servers. It will also download AGPS data packs `cep_alm_pak.zip` and `cep_7days.zip`. diff --git a/huami_token.py b/huami_token.py index a1aaca0..7999d59 100644 --- a/huami_token.py +++ b/huami_token.py @@ -23,9 +23,9 @@ import getpass import json import random import shutil -import urllib +import urllib.parse import uuid -from typing import Iterator, Tuple +from typing import Tuple, Dict, Union, Any, List import zipfile import zlib @@ -34,31 +34,35 @@ import requests import errors import urls -def encode_uint32(value) -> bytes: - return bytes([value & 0xff]) + bytes([(value >> 8) & 0xff]) + bytes([(value >> 16) & 0xff]) + bytes([(value >> 24) & 0xff]); +def encode_uint32(value: int) -> bytes: + """Convert 4-bytes value into a list with 4 bytes""" + return bytes([value & 0xff]) + bytes([(value >> 8) & 0xff]) + \ + bytes([(value >> 16) & 0xff]) + bytes([(value >> 24) & 0xff]) class HuamiAmazfit: """Base class for logging in and receiving auth keys and GPS packs""" - def __init__(self, method="amazfit", email=None, password=None): + def __init__(self, method: str = "amazfit", email: str = "", password: str = "") -> None: if method == 'amazfit' and (not email or not password): raise ValueError("For Amazfit method E-Mail and Password can not be null.") - self.method = method - self.email = email - self.password = password - self.access_token = None - self.country_code = None + self.method: str = method + self.email: str = email + self.password: str = password + self.access_token: str = "" + self.country_code: str = "" - self.app_token = None - self.login_token = None - self.user_id = None + self.app_token: str = "" + self.login_token: str = "" + self.user_id: str = "" self.r = str(uuid.uuid4()) # IMEI or something unique - self.device_id = "02:00:00:%02x:%02x:%02x" % (random.randint(0, 255), - random.randint(0, 255), - random.randint(0, 255)) + self.device_id = ( + f"02:00:00:{random.randint(0, 255):02x}:{random.randint(0, 255):02x}:" + f"{random.randint(0, 255):02x}" + ) + def get_access_token(self) -> str: """Get access token for log in""" @@ -76,7 +80,7 @@ class HuamiAmazfit: if 'code' not in token_url_parameters: raise ValueError("No 'code' parameter in login url.") - self.access_token = token_url_parameters['code'] + self.access_token = token_url_parameters['code'][0] self.country_code = 'US' elif self.method == 'amazfit': @@ -86,12 +90,12 @@ class HuamiAmazfit: data = urls.PAYLOADS['tokens_amazfit'] data['password'] = self.password - response = requests.post(auth_url, data=data, allow_redirects=False) + response = requests.post(auth_url, data=data, allow_redirects=False, timeout=10) response.raise_for_status() # 'Location' parameter contains url with login status redirect_url = urllib.parse.urlparse(response.headers.get('Location')) - redirect_url_parameters = urllib.parse.parse_qs(redirect_url.query) + redirect_url_parameters = urllib.parse.parse_qs(str(redirect_url.query)) if 'error' in redirect_url_parameters: raise ValueError(f"Wrong E-mail or Password." \ @@ -108,26 +112,26 @@ class HuamiAmazfit: self.country_code = region[0:2].upper() else: - self.country_code = redirect_url_parameters['country_code'] + self.country_code = redirect_url_parameters['country_code'][0] - self.access_token = redirect_url_parameters['access'] + self.access_token = redirect_url_parameters['access'][0] return self.access_token - def login(self, external_token=None) -> None: + def login(self, external_token: str = "") -> str: """Perform login and get app and login tokens""" if external_token: self.access_token = external_token login_url = urls.URLS['login_amazfit'] - data = urls.PAYLOADS['login_amazfit'] + data: Dict[str, str] = urls.PAYLOADS['login_amazfit'] data['country_code'] = self.country_code data['device_id'] = self.device_id data['third_name'] = 'huami' if self.method == 'amazfit' else 'mi-watch' data['code'] = self.access_token data['grant_type'] = 'access_token' if self.method == 'amazfit' else 'request_token' - response = requests.post(login_url, data=data, allow_redirects=False) + response = requests.post(login_url, data=data, allow_redirects=False, timeout=10) response.raise_for_status() login_result = response.json() @@ -154,7 +158,7 @@ class HuamiAmazfit: self.user_id = token_info['user_id'] return self.user_id - def get_wearables(self) -> dict: + def get_wearables(self) -> List[Dict[str, Any]]: """Request a list of linked devices""" devices_url = urls.URLS['devices'].format(user_id=urllib.parse.quote(self.user_id)) @@ -162,7 +166,7 @@ class HuamiAmazfit: headers['apptoken'] = self.app_token params = {'enableMultiDevice': 'true'} - response = requests.get(devices_url, params=params, headers=headers) + response = requests.get(devices_url, params=params, headers=headers, timeout=10) response.raise_for_status() device_request = response.json() if 'items' not in device_request: @@ -198,10 +202,10 @@ class HuamiAmazfit: return _wearables @staticmethod - def get_firmware(_wearable: dict) -> Tuple[str, str]: + def get_firmware(_wearable: Dict[str, str]) -> Tuple[List[str], List[str]]: """Check and download updates for the furmware and fonts""" fw_url = urls.URLS["fw_updates"] - params = urls.PAYLOADS["fw_updates"] + params: Dict[str, Union[str, Any]] = urls.PAYLOADS["fw_updates"] params['deviceSource'] = _wearable['device_source'] params['firmwareVersion'] = _wearable['firmware_version'] params['hardwareVersion'] = _wearable['hardware_version'] @@ -211,20 +215,20 @@ class HuamiAmazfit: 'appname': 'com.huami.midong', 'lang': 'en_US' } - response = requests.get(fw_url, params=params, headers=headers) + response = requests.get(fw_url, params=params, headers=headers, timeout=10) response.raise_for_status() fw_response = response.json() - links = [] - hashes = [] + fw_links = [] + fw_hashes = [] if 'firmwareUrl' in fw_response: - links.append(fw_response['firmwareUrl']) - hashes.append(fw_response['firmwareMd5']) + fw_links.append(fw_response['firmwareUrl']) + fw_hashes.append(fw_response['firmwareMd5']) if 'fontUrl' in fw_response: - links.append(fw_response['fontUrl']) - hashes.append(fw_response['fontMd5']) + fw_links.append(fw_response['fontUrl']) + fw_hashes.append(fw_response['fontMd5']) - return (links, hashes) + return (fw_links, fw_hashes) def get_gps_data(self) -> None: """Download GPS packs: almanac and AGPS""" @@ -237,51 +241,58 @@ class HuamiAmazfit: for pack_idx, agps_pack_name in enumerate(agps_packs): print(f"Downloading {agps_pack_name}...") - response = requests.get(agps_link.format(pack_name=agps_pack_name), headers=headers) + response = requests.get(agps_link.format(pack_name=agps_pack_name), + headers=headers, timeout=10) response.raise_for_status() agps_result = response.json()[0] if 'fileUrl' not in agps_result: raise ValueError("No 'fileUrl' parameter in files request.") - with requests.get(agps_result['fileUrl'], stream=True) as request: + with requests.get(agps_result['fileUrl'], stream=True, timeout=10) as request: with open(agps_file_names[pack_idx], 'wb') as gps_file: shutil.copyfileobj(request.raw, gps_file) def build_gps_uihh(self) -> None: + """ Prepare uihh gps file """ print("Building gps_uihh.bin") - d = {'gps_alm.bin':0x05, 'gln_alm.bin':0x0f, 'lle_bds.lle':0x86, 'lle_gps.lle':0x87, 'lle_glo.lle':0x88, 'lle_gal.lle':0x89, 'lle_qzss.lle':0x8a} - cep_archive = zipfile.ZipFile('cep_7days.zip', 'r') - lle_archive = zipfile.ZipFile('lle_1week.zip', 'r') - f = open('gps_uihh.bin', 'wb') - content = bytes() - filecontent = bytes() - fileheader = bytes() + d = {'gps_alm.bin':0x05, 'gln_alm.bin':0x0f, 'lle_bds.lle':0x86, 'lle_gps.lle':0x87, + 'lle_glo.lle':0x88, 'lle_gal.lle':0x89, 'lle_qzss.lle':0x8a} + with zipfile.ZipFile('cep_7days.zip', 'r') as cep_archive, \ + zipfile.ZipFile('lle_1week.zip', 'r') as lle_archive, \ + open('gps_uihh.bin', 'wb') as uihh_file: + content = bytes() + filecontent = bytes() + fileheader = bytes() - for key, value in d.items(): - if value >= 0x86: - filecontent = lle_archive.read(key) - else: - filecontent = cep_archive.read(key) + for key, value in d.items(): + if value >= 0x86: + filecontent = lle_archive.read(key) + else: + filecontent = cep_archive.read(key) - fileheader = bytes([1]) + bytes([value]) + encode_uint32(len(filecontent)) + encode_uint32(zlib.crc32(filecontent) & 0xffffffff) - content += fileheader + filecontent + fileheader = bytes([1]) + bytes([value]) + encode_uint32(len(filecontent)) + \ + encode_uint32(zlib.crc32(filecontent) & 0xffffffff) + content += fileheader + filecontent - header = b'UIHH' + bytes([0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01]) + encode_uint32(zlib.crc32(content) & 0xffffffff) + \ - bytes([0x00, 0x00, 0x00, 0x00, 0x00, 0x00]) + encode_uint32(len(content)) + bytes([0x00, 0x00, 0x00, 0x00, 0x00, 0x00]) + header = b'UIHH' + \ + bytes([0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01]) + \ + encode_uint32(zlib.crc32(content) & 0xffffffff) + \ + bytes([0x00, 0x00, 0x00, 0x00, 0x00, 0x00]) + \ + encode_uint32(len(content)) + \ + bytes([0x00, 0x00, 0x00, 0x00, 0x00, 0x00]) - content = header + content - f.write(content) - f.close() + content = header + content + uihh_file.write(content) - def logout(self) -> None: + def logout(self) -> str: """Log out from the current account""" logout_url = urls.URLS['logout'] data = urls.PAYLOADS['logout'] data['login_token'] = self.login_token - response = requests.post(logout_url, data=data) - logout_result = response.json()['result'] - return logout_result + response = requests.post(logout_url, data=data, timeout=10) + result = str(response.json()['result']) + return result if __name__ == "__main__": @@ -382,7 +393,7 @@ if __name__ == "__main__": print("Be extremely careful with downloaded files!") for idx, wearable in enumerate(wearables): - if idx == wearable_id or wearable_id == -1: + if wearable_id in (idx, -1): print(f"\n\u2553\u2500\u2500\u2500Device {idx}") links, hashes = device.get_firmware(wearables[int(idx)]) if links: @@ -390,11 +401,11 @@ if __name__ == "__main__": file_name = link.split('/')[-1] print(f"\u2551 File: {file_name}") print(f"\u2551 Hash: {hash_sum}") - with requests.get(link, stream=True) as r: + with requests.get(link, stream=True, timeout=10) as r: with open(file_name, 'wb') as f: shutil.copyfileobj(r.raw, f) else: - print(f"\u2551 No updates found") + print("\u2551 No updates found") print(footer) if args.no_logout: diff --git a/requirements.txt b/requirements.txt index ba7e185..e87114c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,4 @@ requests pytest-pylint==0.18.0 pytest-flake8==1.0.6 +types-requests \ No newline at end of file diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_amazfit.py b/tests/test_amazfit.py new file mode 100644 index 0000000..9f76e09 --- /dev/null +++ b/tests/test_amazfit.py @@ -0,0 +1,23 @@ +import os +import unittest +from huami_token import HuamiAmazfit + +class TestAmazfit(unittest.TestCase): + def test_login(self) -> None: + email: str = os.environ.get('AMAZFIT_EMAIL', '') + password: str = os.environ.get('AMAZFIT_PASSWORD', '') + + device = HuamiAmazfit(method="amazfit", + email=email, + password=password) + access_token = device.get_access_token() + user_id = device.login(external_token=access_token) + logout_result = device.logout() + print(user_id) + + self.assertEqual(user_id, + "1132356262", + "Unexpected user id") + +if __name__ == '__main__': + unittest.main() diff --git a/urls.py b/urls.py index 649f609..bb2f5a8 100644 --- a/urls.py +++ b/urls.py @@ -20,6 +20,8 @@ """Module for storin urls and payloads fro different requests""" +from typing import Dict + URLS = { 'login_xiaomi': 'https://account.xiaomi.com/oauth2/authorize?skip_confirm=false&' 'client_id=2882303761517383915&pt=0&scope=1+6000+16001+20000&' @@ -34,12 +36,12 @@ URLS = { 'fw_updates': 'https://api-mifit-us2.huami.com/devices/ALL/hasNewVersion' } -PAYLOADS = { - 'login_xiaomi': None, +PAYLOADS: Dict[str, Dict[str, str]] = { + 'login_xiaomi': {}, 'tokens_amazfit': { 'state': 'REDIRECTION', 'client_id': 'HuaMi', - 'password': None, + 'password': "", 'redirect_uri': 'https://s3-us-west-2.amazonws.com/hm-registration/successsignin.html', 'region': 'us-west-2', 'token': 'access', @@ -51,39 +53,39 @@ PAYLOADS = { 'api-analytics.huami.com,api-mifit.huami.com', 'app_version': '5.9.2-play_100355', 'source': 'com.huami.watch.hmwatchmanager', - 'country_code': None, - 'device_id': None, - 'third_name': None, + 'country_code': "", + 'device_id': "", + 'third_name': "", 'lang': 'en', 'device_model': 'android_phone', 'allow_registration': 'false', 'app_name': 'com.huami.midong', - 'code': None, - 'grant_type': None + 'code': "", + 'grant_type': "" }, 'devices': { - 'apptoken': None, + 'apptoken': "", # 'enableMultiDevice': 'true' }, 'agps': { - 'apptoken': None + 'apptoken': "" }, 'data_short': { - 'apptoken': None, - 'startDay': None, - 'endDay': None + 'apptoken': "", + 'startDay': "", + 'endDay': "" }, 'logout': { - 'login_token': None + 'login_token': "" }, 'fw_updates': { - 'productionSource': None, - 'deviceSource': None, + 'productionSource': "", + 'deviceSource': "", 'fontVersion': '0', 'fontFlag': '0', 'appVersion': '5.9.2-play_100355', - 'firmwareVersion': None, - 'hardwareVersion': None, + 'firmwareVersion': "", + 'hardwareVersion': "", 'support8Bytes': 'true' } }