feat: RSA login encryption + confirmed wechat type codes
Replace the Phase-0 placeholder password encryption with real RSA
(PKCS#1 v1.5). encrypt_password now builds an RSA public key from the
getSecureKey modulus/exponent (RSAPublicNumbers, e=int(hex,16),
n=int(hex,16)), encrypts the UTF-8 password, and returns hex (documented
assumption -- base64 is the one-line fallback noted in the docstring).
get_secure_key now returns the inner data dict to match the new contract;
pc_login keeps {account, password} with a comment on the username/password
and account/pwd fallbacks if pcLogin rejects.
Extend normalize_wechat_type with the Phase-0 confirmed codes:
2 -> image, 3 -> image_legacy, 4 -> video, 10 -> system (1 -> text and
-2 -> divider unchanged; 6/9/13/15/21/22 stay unknown until confirmed).
Tests:
- encrypt_password round-trip: generate an RSA keypair, build secure_key
from its public numbers, encrypt, then decrypt with the private key and
assert the original password comes back. Pins 512-hex output for a
2048-bit key. No network.
- Parser parametrize table extended for the new codes (and a downstream
message-fixture assertion updated from unknown -> image_legacy).
Suite: 87 -> 97 passing. ruff check clean.
Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
浅念
@@ -18,17 +18,27 @@ from app.scraper.parsers import normalize_wechat_type
|
||||
@pytest.mark.parametrize(
|
||||
"raw,expected",
|
||||
[
|
||||
# Documented codes.
|
||||
# Phase-0 confirmed codes (see docs/superpowers/specs/phase0-findings.md §2).
|
||||
(1, "text"),
|
||||
(-2, "divider"),
|
||||
# Phase-0 unknown codes — pass through as "unknown", never dropped.
|
||||
(3, "unknown"), # image
|
||||
(34, "unknown"), # voice
|
||||
(-2, "divider"), # date separator — NOT a real message (scraper drops it)
|
||||
(2, "image"), # image / sticker (webp) — file/fileTh are direct URLs
|
||||
(3, "image_legacy"), # old image format — file is an opaque ID (needs another endpoint)
|
||||
(4, "video"), # mp4 — file is a direct URL
|
||||
(10, "system"), # system / link / card / invite — non-media, skip download
|
||||
# Still-unconfirmed codes — pass through as "unknown", never dropped.
|
||||
# (6, 9, 13, 15, 21, 22 — voice/file/transfer/redpacket/quote/etc; raw_json retains the code.)
|
||||
(6, "unknown"),
|
||||
(9, "unknown"),
|
||||
(34, "unknown"), # voice (legacy WeChat code, not yet seen in yunke)
|
||||
(49, "unknown"), # file / link / transfer
|
||||
(10000, "unknown"),
|
||||
# Stringified ints (the upstream API may send either form).
|
||||
("1", "text"),
|
||||
("-2", "divider"),
|
||||
("2", "image"),
|
||||
("3", "image_legacy"),
|
||||
("4", "video"),
|
||||
("10", "system"),
|
||||
("34", "unknown"),
|
||||
],
|
||||
)
|
||||
|
||||
@@ -114,12 +114,12 @@ async def test_scrape_wechat_messages_drops_divider_and_paginates_backward():
|
||||
assert r1["sender_head_url"] == "https://cdn.example.com/alice.png"
|
||||
assert r1["raw_json"] == MESSAGE_RECORDS_PAGE1[0]
|
||||
|
||||
# m2 — image / chatroom / outgoing.
|
||||
# m2 — image (legacy) / chatroom / outgoing.
|
||||
assert r2["msg_id"] == "1000002"
|
||||
assert r2["room_id"] == "room12345@chatroom"
|
||||
assert r2["ori_talker"] == "room12345@chatroom"
|
||||
assert r2["is_send"] == 1
|
||||
assert r2["type"] == "unknown" # unmapped code passes through
|
||||
assert r2["type"] == "image_legacy" # code 3 -> legacy image (phase0 §2)
|
||||
assert r2["type_raw"] == 3 # raw image-type code preserved
|
||||
assert r2["file_url"] == "https://cdn.example.com/img2.jpg"
|
||||
assert r2["file_th_url"] == "https://cdn.example.com/img2_th.jpg"
|
||||
|
||||
@@ -4,11 +4,13 @@ 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 / password encryption /
|
||||
response shape were not captured. The mocks here encode our documented
|
||||
assumptions (see ``app/session/login.py``): response JSON has a ``data``
|
||||
object with ``userid`` + ``company`` fields. When the real shapes are
|
||||
confirmed, only the parsing inside ``pc_login`` needs to change — these
|
||||
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.
|
||||
"""
|
||||
|
||||
@@ -17,6 +19,7 @@ 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
|
||||
@@ -25,6 +28,25 @@ 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():
|
||||
@@ -43,9 +65,7 @@ async def client():
|
||||
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": {"modulus": "00ab", "exponent": "010001"}}
|
||||
)
|
||||
return_value=httpx.Response(200, json={"data": _TEST_SECURE_KEY})
|
||||
)
|
||||
login = respx.post(f"{BASE}/usercenter/login/pcLogin").mock(
|
||||
return_value=httpx.Response(
|
||||
@@ -190,19 +210,51 @@ async def test_first_request_auto_logs_in(client: YunkeClient):
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test F: encrypt_password is a documented Phase-0 placeholder
|
||||
# Test F: encrypt_password performs real RSA (PKCS#1 v1.5) encryption
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_encrypt_password_is_swappable_phase0_placeholder():
|
||||
"""Phase 0: encrypt_password must be deterministic and non-trivial but is
|
||||
NOT the real RSA scheme. It must accept any secure-key shape and return a
|
||||
str so the login flow is end-to-end runnable in tests.
|
||||
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
|
||||
|
||||
out = encrypt_password("hunter2", {"modulus": "00ab", "exponent": "010001"})
|
||||
assert isinstance(out, str)
|
||||
assert out # non-empty
|
||||
# Deterministic for the same inputs (so tests / debugging are stable).
|
||||
assert encrypt_password("hunter2", {"modulus": "00ab"}) == out
|
||||
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"})
|
||||
|
||||
Reference in New Issue
Block a user