mirror of
https://github.com/alexgetmancom/miband-bot.git
synced 2026-07-18 11:40:12 +03:00
security: redact Xiaomi auth logs
This commit is contained in:
@@ -116,11 +116,7 @@ async def extract_credentials(
|
||||
service_token = await extract_service_token(http, location)
|
||||
token.service_token = service_token
|
||||
|
||||
logger.debug(
|
||||
"凭证提取完成, ssecurity={}, user_id={}",
|
||||
token.ssecurity[:8] + "...",
|
||||
token.user_id,
|
||||
)
|
||||
logger.debug("凭证提取完成, user_id={}", token.user_id)
|
||||
|
||||
|
||||
async def async_sleep(seconds: float) -> None:
|
||||
|
||||
@@ -67,9 +67,9 @@ async def login_qr(
|
||||
await qr_callback(qr_image_url, login_url)
|
||||
else:
|
||||
logger.info("请使用小米账号 APP 扫描二维码登录")
|
||||
logger.info("二维码图片: {}", qr_image_url)
|
||||
logger.info("二维码图片已获取")
|
||||
if login_url:
|
||||
logger.info("或在浏览器打开: {}", login_url)
|
||||
logger.info("浏览器登录链接已获取")
|
||||
|
||||
# Step 2: 长轮询等待扫码
|
||||
effective_timeout = min(float(qr_timeout), max_wait)
|
||||
|
||||
@@ -2,8 +2,8 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
import os
|
||||
import time
|
||||
|
||||
from loguru import logger
|
||||
|
||||
@@ -46,10 +46,6 @@ async def sts_exchange(http: RetryAsyncClient, token: AuthToken) -> None:
|
||||
resp = await http.get(STS_HEALTH_URL, params=params, cookies=cookies)
|
||||
if resp.text.strip() == "ok":
|
||||
logger.debug("STS 交换成功")
|
||||
# Extract STS serviceToken from cookies and save it to AuthToken
|
||||
logger.debug("Куки в клиенте во время STS:")
|
||||
for cookie in http.cookies.jar:
|
||||
logger.debug(f" Cookie: {cookie.name}, Domain: {cookie.domain}, Value: {cookie.value[:15]}...")
|
||||
sts_token = None
|
||||
for cookie in http.cookies.jar:
|
||||
if cookie.name == "serviceToken" and "hlth.io.mi.com" in (cookie.domain or ""):
|
||||
@@ -57,10 +53,10 @@ async def sts_exchange(http: RetryAsyncClient, token: AuthToken) -> None:
|
||||
break
|
||||
if not sts_token:
|
||||
sts_token = http.cookies.get("serviceToken")
|
||||
|
||||
|
||||
if sts_token:
|
||||
token.service_token = sts_token
|
||||
logger.debug("STS serviceToken успешно сохранен в AuthToken: {}", sts_token[:15] + "...")
|
||||
logger.debug("STS serviceToken успешно сохранен в AuthToken")
|
||||
else:
|
||||
logger.warning("STS 交换响应: {}", resp.text[:100])
|
||||
except Exception as e:
|
||||
|
||||
@@ -48,9 +48,9 @@ async def _qr_login() -> None:
|
||||
print("\n📱 请用小米账号 APP 扫描二维码登录\n")
|
||||
if login_url:
|
||||
_print_qr_to_terminal(login_url)
|
||||
print(f"\n 二维码图片: {qr_image_url}")
|
||||
if login_url:
|
||||
print(f" 浏览器打开: {login_url}")
|
||||
elif qr_image_url:
|
||||
_print_qr_to_terminal(qr_image_url)
|
||||
print("\n 登录二维码已显示在终端;URL 不会打印。")
|
||||
print("\n⏳ 等待扫码...\n")
|
||||
|
||||
async with XiaomiAuth() as auth:
|
||||
|
||||
@@ -6,6 +6,7 @@ from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from loguru import logger
|
||||
|
||||
from mi_fitness.auth import _helpers as auth_helpers
|
||||
from mi_fitness.auth import sts as auth_sts
|
||||
@@ -22,6 +23,13 @@ def _response(
|
||||
return httpx.Response(200, text=text, headers=headers, request=request)
|
||||
|
||||
|
||||
class _Cookie:
|
||||
def __init__(self, name: str, domain: str, value: str):
|
||||
self.name = name
|
||||
self.domain = domain
|
||||
self.value = value
|
||||
|
||||
|
||||
def test_parse_mi_response_supports_start_prefix() -> None:
|
||||
parsed = auth_helpers.parse_mi_response('&&&START&&&{"code":0,"desc":"ok"}')
|
||||
assert parsed == {"code": 0, "desc": "ok"}
|
||||
@@ -156,6 +164,26 @@ async def test_sts_exchange_uses_device_id_and_accepts_ok_response(
|
||||
assert params["p_ts"] == "1234"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_sts_exchange_does_not_log_cookie_values(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
http = MagicMock()
|
||||
http.get = AsyncMock(return_value=_response(text="ok"))
|
||||
http.cookies.jar = [_Cookie("serviceToken", "hlth.io.mi.com", "secret-service-token")]
|
||||
token = AuthToken(device_id="an_device", pass_token="secret-pass-token")
|
||||
messages: list[str] = []
|
||||
sink_id = logger.add(lambda message: messages.append(str(message)), level="DEBUG", format="{message}")
|
||||
|
||||
try:
|
||||
await auth_sts.sts_exchange(http, token)
|
||||
finally:
|
||||
logger.remove(sink_id)
|
||||
|
||||
logs = "\n".join(messages)
|
||||
assert "secret-service-token" not in logs
|
||||
assert "secret-pass-token" not in logs
|
||||
assert "serviceToken успешно сохранен" in logs
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_sts_exchange_swallows_network_failure() -> None:
|
||||
http = MagicMock()
|
||||
|
||||
@@ -45,7 +45,7 @@ class _FakeAuth:
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_qr_login_saves_token(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
async def test_qr_login_saves_token(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
_FakeAuth.instances.clear()
|
||||
tmp_dir = _workspace_tmp_dir()
|
||||
monkeypatch.setattr(cli, "XiaomiAuth", _FakeAuth)
|
||||
@@ -55,6 +55,9 @@ async def test_qr_login_saves_token(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
auth = _FakeAuth.instances[-1]
|
||||
assert auth.login_calls == [("qr",)]
|
||||
assert auth.saved_path == tmp_dir / "token.json"
|
||||
stdout = capsys.readouterr().out
|
||||
assert "https://example.com/login" not in stdout
|
||||
assert "https://example.com/qr.png" not in stdout
|
||||
finally:
|
||||
shutil.rmtree(tmp_dir, ignore_errors=True)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user