"""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"})