GithubFollow/ha_xiaomi_home
1
0
Fork 0
mirror of https://github.com/XiaoMi/ha_xiaomi_home.git synced 2026-10-27 16:10:49 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat: add user-agent (#1804)

Browse Source
This commit is contained in:
Li Shuzhen
2026-09-28 13:59:57 +08:00
committed by GitHub
parent a253918896
commit 7201cf2277
6 changed files with 77 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
custom_components/xiaomi_home/config_flow.py
View File

@@ -92,6 +92,7 @@ from .miot.const import (
MIHOME_CERT_EXPIRE_MARGIN
)
from .miot.miot_cloud import MIoTHttpClient, MIoTOauthClient
from .miot.common import get_system_info_str
from .miot.miot_storage import MIoTStorage, MIoTCert
from .miot.miot_mdns import MipsService
from .miot.web_pages import oauth_redirect_page
@@ -175,6 +176,7 @@ class XiaomiMihomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self._oauth_redirect_url_full = ''
self._miot_oauth = None
self._miot_http = None
self._system_info = ''
self._cc_home_info = {}
self._cc_home_list_show = {}
@@ -438,11 +440,13 @@ class XiaomiMihomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
_LOGGER.info(
'async_step_oauth, redirect_url: %s',
self._oauth_redirect_url_full)
self._system_info = await get_system_info_str(self.hass)
miot_oauth = MIoTOauthClient(
client_id=OAUTH2_CLIENT_ID,
redirect_url=self._oauth_redirect_url_full,
cloud_server=self._cloud_server,
uuid=self._uuid,
system_info=self._system_info,
loop=self._main_loop)
self._cc_oauth_auth_url = miot_oauth.gen_auth_url(
redirect_url=self._oauth_redirect_url_full)
@@ -520,7 +524,9 @@ class XiaomiMihomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self._miot_http = MIoTHttpClient(
cloud_server=self._cloud_server,
client_id=OAUTH2_CLIENT_ID,
access_token=auth_info['access_token'])
access_token=auth_info['access_token'],
uuid=self._uuid,
system_info=self._system_info)
else:
self._miot_http.update_http_header(
cloud_server=self._cloud_server,
@@ -1275,10 +1281,13 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
'get_access_token, %s, %s', err, traceback.format_exc())
raise MIoTConfigError('get_token_error') from err
# Check uid
system_info = await get_system_info_str(self.hass)
m_http: MIoTHttpClient = MIoTHttpClient(
cloud_server=self._cloud_server,
client_id=OAUTH2_CLIENT_ID,
access_token=auth_info['access_token'],
uuid=self._entry_data['uuid'],
system_info=system_info,
loop=self._main_loop)
if await m_http.get_uid_async() != self._uid:
raise AbortFlow('inconsistent_account')

16
custom_components/xiaomi_home/miot/common.py
View File

@@ -186,3 +186,19 @@ class MIoTHttp:
ev_loop = loop or asyncio.get_running_loop()
return await ev_loop.run_in_executor(
None, MIoTHttp.post, url, data, headers)
async def get_system_info_str(hass) -> str:
"""Get system info string for User-Agent."""
# pylint: disable=import-outside-toplevel
from homeassistant.helpers.system_info import async_get_system_info
info = await async_get_system_info(hass)
version = info['version']
os_name = info['os_name']
os_version = info['os_version']
arch = info['arch']
installation_type = info['installation_type']
return (
f'hass/{version} ({os_name}/{os_version};'
f' {arch}; {installation_type})'
)

7
custom_components/xiaomi_home/miot/const.py
View File

@@ -45,7 +45,14 @@ off Xiaomi or its affiliates' products.
Constants.
"""
import json
from pathlib import Path
DOMAIN: str = 'xiaomi_home'
INTEGRATION_VERSION: str = json.loads(
(Path(__file__).parent.parent / 'manifest.json').read_text()
)['version']
DEFAULT_NAME: str = 'Xiaomi Home'
DEFAULT_NICK_NAME: str = 'Xiaomi'

9
custom_components/xiaomi_home/miot/miot_client.py
View File

@@ -59,7 +59,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.components import zeroconf
# pylint: disable=relative-beyond-top-level
from .common import MIoTMatcher, slugify_did
from .common import MIoTMatcher, get_system_info_str, slugify_did
from .const import (
DEFAULT_CTRL_MODE, DEFAULT_INTEGRATION_LANGUAGE, DEFAULT_NICK_NAME, DOMAIN,
MIHOME_CERT_EXPIRE_MARGIN, NETWORK_REFRESH_INTERVAL,
@@ -189,6 +189,7 @@ class MIoTClient:
storage: MIoTStorage,
mips_service: MipsService,
miot_lan: MIoTLan,
system_info: str = '',
loop: Optional[asyncio.AbstractEventLoop] = None) -> None:
# MUST run in a running event loop
self._main_loop = loop or asyncio.get_running_loop()
@@ -207,6 +208,7 @@ class MIoTClient:
self._entry_data = entry_data
self._uid = entry_data['uid']
self._cloud_server = entry_data['cloud_server']
self._system_info = system_info
self._ctrl_mode = CtrlMode.load(
entry_data.get('ctrl_mode', DEFAULT_CTRL_MODE))
self._network = network
@@ -285,12 +287,15 @@ class MIoTClient:
redirect_url=self._entry_data['oauth_redirect_url'],
cloud_server=self._cloud_server,
uuid=self._entry_data["uuid"],
system_info=self._system_info,
loop=self._main_loop)
# MIoT http client instance
self._http = MIoTHttpClient(
cloud_server=self._cloud_server,
client_id=OAUTH2_CLIENT_ID,
access_token=self._user_config['auth_info']['access_token'],
uuid=self._entry_data['uuid'],
system_info=self._system_info,
loop=self._main_loop)
# MIoT cert client
self._cert = MIoTCert(
@@ -2041,6 +2046,7 @@ async def get_miot_instance_async(
hass.data[DOMAIN]['miot_lan'] = miot_lan
_LOGGER.info('create miot_lan instance')
# MIoT client
system_info = await get_system_info_str(hass)
miot_client = MIoTClient(
entry_id=entry_id,
entry_data=entry_data,
@@ -2048,6 +2054,7 @@ async def get_miot_instance_async(
storage=storage,
mips_service=mips_service,
miot_lan=miot_lan,
system_info=system_info,
loop=loop
)
miot_client.persistent_notify = persistent_notify

27
custom_components/xiaomi_home/miot/miot_cloud.py
View File

@@ -61,6 +61,7 @@ from .common import calc_group_id
from .const import (
UNSUPPORTED_MODELS,
DEFAULT_OAUTH2_API_HOST,
INTEGRATION_VERSION,
MIHOME_HTTP_API_TIMEOUT,
OAUTH2_AUTH_URL)
from .miot_error import MIoTErrorCode, MIoTHttpError, MIoTOauthError
@@ -78,11 +79,13 @@ class MIoTOauthClient:
_client_id: int
_redirect_url: str
_device_id: str
_user_agent: str
_state: str
def __init__(
self, client_id: str, redirect_url: str, cloud_server: str,
uuid: str, loop: Optional[asyncio.AbstractEventLoop] = None
uuid: str, system_info: str = '',
loop: Optional[asyncio.AbstractEventLoop] = None
) -> None:
self._main_loop = loop or asyncio.get_running_loop()
if client_id is None or client_id.strip() == '':
@@ -101,6 +104,10 @@ class MIoTOauthClient:
else:
self._oauth_host = f'{cloud_server}.{DEFAULT_OAUTH2_API_HOST}'
self._device_id = f'ha.{uuid}'
self._user_agent = (
f'ha_xiaomi_home/{INTEGRATION_VERSION}'
f' {system_info}'
f' client/{self._device_id}')
self._state = hashlib.sha1(
f'd={self._device_id}'.encode('utf-8')).hexdigest()
self._session = aiohttp.ClientSession(loop=self._main_loop)
@@ -161,7 +168,10 @@ class MIoTOauthClient:
http_res = await self._session.get(
url=f'https://{self._oauth_host}/app/v2/ha/oauth/get_token',
params={'data': json.dumps(data)},
headers={'content-type': 'application/x-www-form-urlencoded'},
headers={
'Content-Type': 'application/x-www-form-urlencoded',
'User-Agent': self._user_agent,
},
timeout=MIHOME_HTTP_API_TIMEOUT
)
if http_res.status == 401:
@@ -239,12 +249,14 @@ class MIoTHttpClient:
_base_url: str
_client_id: str
_access_token: str
_user_agent: str
_get_prop_timer: Optional[asyncio.TimerHandle]
_get_prop_list: dict[str, dict]
def __init__(
self, cloud_server: str, client_id: str, access_token: str,
uuid: str, system_info: str = '',
loop: Optional[asyncio.AbstractEventLoop] = None
) -> None:
self._main_loop = loop or asyncio.get_running_loop()
@@ -260,9 +272,14 @@ class MIoTHttpClient:
not isinstance(cloud_server, str)
or not isinstance(client_id, str)
or not isinstance(access_token, str)
or not isinstance(uuid, str)
):
raise MIoTHttpError('invalid params')
self._user_agent = (
f'ha_xiaomi_home/{INTEGRATION_VERSION}'
f' {system_info}'
f' client/ha.{uuid}')
self.update_http_header(
cloud_server=cloud_server, client_id=client_id,
access_token=access_token)
@@ -303,6 +320,7 @@ class MIoTHttpClient:
'Content-Type': 'application/json',
'Authorization': f'Bearer{self._access_token}',
'X-Client-AppId': self._client_id,
'User-Agent': self._user_agent,
}
# pylint: disable=unused-private-member
@@ -367,7 +385,10 @@ class MIoTHttpClient:
url='https://open.account.xiaomi.com/user/profile',
params={
'clientId': self._client_id, 'token': self._access_token},
headers={'content-type': 'application/x-www-form-urlencoded'},
headers={
'Content-Type': 'application/x-www-form-urlencoded',
'User-Agent': self._user_agent,
},
timeout=MIHOME_HTTP_API_TIMEOUT
)

13
test/conftest.py
View File

@@ -4,7 +4,7 @@ import logging
import random
import shutil
import pytest
from os import path, makedirs
from os import path, makedirs, remove
from uuid import uuid4
TEST_ROOT_PATH: str = path.dirname(path.abspath(__file__))
@@ -57,6 +57,12 @@ def load_py_file():
file_name),
path.join(TEST_FILES_PATH, file_name))
_LOGGER.info('\nloaded test py files, %s', file_list)
# Copy manifest.json to test folder (parent of miot/)
shutil.copyfile(
path.join(
TEST_ROOT_PATH, '../custom_components/xiaomi_home/manifest.json'),
path.join(TEST_ROOT_PATH, 'manifest.json'))
_LOGGER.info('loaded manifest.json')
# Copy spec files to test folder
shutil.copytree(
src=path.join(
@@ -89,6 +95,11 @@ def load_py_file():
shutil.rmtree(TEST_FILES_PATH)
print('\nremoved test files, ', TEST_FILES_PATH)
manifest_path = path.join(TEST_ROOT_PATH, 'manifest.json')
if path.exists(manifest_path):
remove(manifest_path)
print('removed test manifest.json, ', manifest_path)
if path.exists(TEST_CACHE_PATH):
shutil.rmtree(TEST_CACHE_PATH)
print('removed test cache, ', TEST_CACHE_PATH)