浅念
4dcfc66741
Live validation found two real bugs: - YunkeClient sent 'source: yunkepc' on every request; msg/chats binds source as Integer -> 500. Removed (cookie auth suffices, per Phase 0). - WeChat message talker_id used the masked 'talker' field; the timeline joins on the full peer id. Now: 1:1 -> oriTalker (full), group -> talker (member). ori_talker keeps the raw oriTalker (peer/chatroom). Co-Authored-By: Claude <noreply@anthropic.com>
264 lines
10 KiB
Python
264 lines
10 KiB
Python
"""Tests for the yunke session layer: ``app.session.login`` + ``YunkeClient``.
|
|
|
|
These tests use ``respx`` to mock the upstream httpx transport so we can
|
|
assert exactly what gets sent (headers, body) and drive the auto-login +
|
|
401-relogin behaviour without touching the real CRM.
|
|
|
|
Phase-0 caveat: the real ``pcLogin`` request-body field names and response
|
|
shape are still inferences (see ``app/session/login.py``). Password
|
|
encryption is now confirmed real RSA (PKCS#1 v1.5) against the
|
|
``getSecureKey`` modulus/exponent. The mocks here encode the confirmed
|
|
``getSecureKey`` payload shape and our inferred response shape: JSON with a
|
|
``data`` object carrying ``userid`` + ``company``. When the response shape
|
|
is captured, only the parsing inside ``pc_login`` needs to change — these
|
|
tests stay green.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
from cryptography.hazmat.primitives.asymmetric import padding, rsa
|
|
|
|
from app.session.login import LoginResult
|
|
from app.session.yunke_client import YunkeClient
|
|
|
|
BASE = "https://crm2.yunkecn.com"
|
|
USERID = "a" * 32 # 32-hex user id, as observed in real auth headers
|
|
COMPANY = "8m0atd"
|
|
|
|
# Fixed RSA keypair backing the mocked ``getSecureKey`` endpoint so the real
|
|
# RSA step in pc_login has a valid key to encrypt against. The matching
|
|
# private key is retained so the round-trip test can decrypt what
|
|
# encrypt_password produced. Generated once at import (cheap vs. network).
|
|
_TEST_PRIVATE_KEY = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
|
|
|
|
def _secure_key_from_test_public_key() -> dict[str, str]:
|
|
"""Build the Phase-0 confirmed ``getSecureKey`` ``data`` payload from the
|
|
test public key — ``public_exponent`` + ``modulus`` as hex strings."""
|
|
nums = _TEST_PRIVATE_KEY.public_key().public_numbers()
|
|
return {
|
|
"public_exponent": format(nums.e, "x"),
|
|
"modulus": format(nums.n, "x"),
|
|
}
|
|
|
|
|
|
_TEST_SECURE_KEY = _secure_key_from_test_public_key()
|
|
|
|
|
|
@pytest.fixture()
|
|
async def client():
|
|
"""A YunkeClient wired for username/password login.
|
|
|
|
The fixture yields the client and always closes it so respx mocks set up
|
|
inside individual tests don't leak cookies across tests.
|
|
"""
|
|
c = YunkeClient(base_url=BASE, username="tester", password="hunter2")
|
|
try:
|
|
yield c
|
|
finally:
|
|
await c.aclose()
|
|
|
|
|
|
def _stub_login_endpoints() -> tuple[respx.Route, respx.Route]:
|
|
"""Register the getSecureKey + pcLogin mocks used across tests."""
|
|
secure = respx.get(f"{BASE}/newuc/login/getSecureKey").mock(
|
|
return_value=httpx.Response(200, json={"data": _TEST_SECURE_KEY})
|
|
)
|
|
login = respx.post(f"{BASE}/usercenter/login/pcLogin").mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
json={"code": 0, "data": {"userid": USERID, "company": COMPANY}},
|
|
headers={"Set-Cookie": "JSESSIONID=deadbeef; Path=/"},
|
|
)
|
|
)
|
|
return secure, login
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test A: login() populates LoginResult + the httpx client cookie jar
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@respx.mock
|
|
async def test_login_stores_userid_and_cookies(client: YunkeClient):
|
|
_, login_route = _stub_login_endpoints()
|
|
|
|
result = await client.login()
|
|
|
|
assert login_route.called
|
|
assert isinstance(result, LoginResult)
|
|
assert result.userid == USERID
|
|
assert result.company == COMPANY
|
|
# LoginResult is the same object stored on the client.
|
|
assert client.login_result is result
|
|
# The session cookie from the mocked Set-Cookie must land in the httpx
|
|
# client's cookie jar so subsequent requests carry it.
|
|
assert client.client.cookies.get("JSESSIONID") == "deadbeef"
|
|
# The snapshot on LoginResult should also contain it.
|
|
assert result.cookies.get("JSESSIONID") == "deadbeef"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test B: every request carries the auth headers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@respx.mock
|
|
async def test_post_injects_auth_headers(client: YunkeClient):
|
|
# Skip the login flow entirely — prime the client with a logged-in state.
|
|
client.login_result = LoginResult(userid=USERID, company=COMPANY)
|
|
client.client.cookies.set("JSESSIONID", "deadbeef")
|
|
|
|
route = respx.post(f"{BASE}/wechatweb/wxFriends/friend_list/es").mock(
|
|
return_value=httpx.Response(200, json={"data": []})
|
|
)
|
|
|
|
resp = await client.post(
|
|
"/wechatweb/wxFriends/friend_list/es", json={"pageNum": 1, "pageSize": 20}
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
assert route.called
|
|
|
|
sent = route.calls.last.request
|
|
# The 5 known auth headers must be on every outgoing request.
|
|
assert sent.headers["userid"] == USERID
|
|
assert sent.headers["company"] == COMPANY
|
|
# NOTE: ``source`` is intentionally NOT sent — several endpoints (msg/chats)
|
|
# bind a ``source`` param as Integer, so ``source: "yunkepc"`` 500s. Cookie
|
|
# auth suffices for all data endpoints (Phase 0).
|
|
assert "source" not in sent.headers
|
|
assert sent.headers["channel"] == "4"
|
|
assert sent.headers["content-type"] == "application/json"
|
|
# And the session cookie travels along too.
|
|
assert sent.headers.get("cookie", "").find("JSESSIONID=deadbeef") != -1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test C: 401 -> re-login once -> retry succeeds
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@respx.mock
|
|
async def test_401_triggers_relogin_and_retry(client: YunkeClient):
|
|
_, login_route = _stub_login_endpoints()
|
|
data_route = respx.post(
|
|
f"{BASE}/wechatweb/wxFriends/friend_list/es"
|
|
).mock(
|
|
side_effect=[
|
|
httpx.Response(401, json={"msg": "session expired"}),
|
|
httpx.Response(200, json={"data": [{"id": 1}]}),
|
|
]
|
|
)
|
|
|
|
# Prime as logged-in so the first call doesn't itself trigger an
|
|
# auto-login; the 401 is what should drive exactly one re-login.
|
|
client.login_result = LoginResult(userid=USERID, company=COMPANY)
|
|
|
|
resp = await client.post("/wechatweb/wxFriends/friend_list/es", json={})
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json() == {"data": [{"id": 1}]}
|
|
# Exactly one re-login + exactly one retry — no infinite loop.
|
|
assert login_route.call_count == 1
|
|
assert data_route.call_count == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test D: persistent 401 -> raise after the single retry (no loop)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@respx.mock
|
|
async def test_persistent_401_raises_after_one_retry(client: YunkeClient):
|
|
_, login_route = _stub_login_endpoints()
|
|
data_route = respx.post(
|
|
f"{BASE}/wechatweb/wxFriends/friend_list/es"
|
|
).mock(return_value=httpx.Response(401, json={"msg": "still expired"}))
|
|
|
|
client.login_result = LoginResult(userid=USERID, company=COMPANY)
|
|
|
|
with pytest.raises(httpx.HTTPStatusError):
|
|
await client.post("/wechatweb/wxFriends/friend_list/es", json={})
|
|
|
|
# The single retry happened, then we gave up and raised.
|
|
assert login_route.call_count == 1
|
|
assert data_route.call_count == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test E: auto-login on first request when no session yet
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@respx.mock
|
|
async def test_first_request_auto_logs_in(client: YunkeClient):
|
|
secure_route, login_route = _stub_login_endpoints()
|
|
data_route = respx.post(f"{BASE}/wechatweb/wxFriends/friend_list/es").mock(
|
|
return_value=httpx.Response(200, json={"data": []})
|
|
)
|
|
|
|
# No priming — client should auto-login on first call.
|
|
resp = await client.post("/wechatweb/wxFriends/friend_list/es", json={})
|
|
|
|
assert resp.status_code == 200
|
|
assert secure_route.called
|
|
assert login_route.called
|
|
assert data_route.call_count == 1
|
|
assert client.login_result is not None
|
|
assert client.login_result.userid == USERID
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test F: encrypt_password performs real RSA (PKCS#1 v1.5) encryption
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_encrypt_password_rsa_round_trips():
|
|
"""Real RSA: encrypt_password must produce a hex PKCS#1 v1.5 ciphertext
|
|
that the matching private key decrypts back to the original password.
|
|
|
|
``secure_key`` is built from the test public key in the Phase-0 confirmed
|
|
shape (``public_exponent`` + ``modulus`` as hex strings, exactly what
|
|
``GET /newuc/login/getSecureKey`` returns). Proves real RSA end-to-end
|
|
without needing the live yunke key. No network.
|
|
"""
|
|
from app.session.login import encrypt_password
|
|
|
|
cipher_hex = encrypt_password("hunter2", _TEST_SECURE_KEY)
|
|
|
|
# Output contract: a hex string (yunke convention; see phase0-findings §1).
|
|
assert isinstance(cipher_hex, str)
|
|
assert len(cipher_hex) % 2 == 0
|
|
assert int(cipher_hex, 16) > 0
|
|
# For a 2048-bit key the PKCS#1 v1.5 ciphertext is 256 bytes -> 512 hex
|
|
# chars. Pin the length so a regression to a stub (e.g. base64 of the
|
|
# 7-byte password) is caught immediately.
|
|
assert len(cipher_hex) == 512
|
|
|
|
# The decisive assertion: decrypt with the matching private key and the
|
|
# original UTF-8 password must come back.
|
|
plain = _TEST_PRIVATE_KEY.decrypt(
|
|
bytes.fromhex(cipher_hex), padding.PKCS1v15()
|
|
)
|
|
assert plain.decode("utf-8") == "hunter2"
|
|
|
|
|
|
def test_encrypt_password_uses_phase0_confirmed_field_names():
|
|
"""encrypt_password must read ``public_exponent`` + ``modulus`` (the exact
|
|
field names ``getSecureKey`` returns), not the earlier placeholder names.
|
|
A key dict carrying only the old ``exponent`` field must NOT work.
|
|
"""
|
|
from app.session.login import encrypt_password
|
|
|
|
# The confirmed-shape dict encrypts without error.
|
|
out = encrypt_password("hunter2", _TEST_SECURE_KEY)
|
|
assert isinstance(out, str) and out
|
|
|
|
# An empty / wrong-shape dict must raise, not silently return junk.
|
|
with pytest.raises(ValueError, match="public_exponent"):
|
|
encrypt_password("hunter2", {"modulus": "00ab", "exponent": "010001"})
|