134 lines
4.1 KiB
Python
134 lines
4.1 KiB
Python
import unittest
|
|
from enum import IntEnum
|
|
from types import SimpleNamespace
|
|
|
|
from core.http_transport import (
|
|
OKHTTP4_ANDROID10_AKAMAI,
|
|
OKHTTP4_ANDROID10_JA3,
|
|
create_http_session,
|
|
)
|
|
|
|
|
|
class _FakeCookies:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def set(self, name, value, *, domain="", path="/", secure=False):
|
|
self.calls.append((name, value, domain, path, secure))
|
|
|
|
|
|
class _FakeCurlSession:
|
|
def __init__(self):
|
|
self.cookies = _FakeCookies()
|
|
self.calls = []
|
|
|
|
def request(self, method, url, **kwargs):
|
|
self.calls.append((method, url, kwargs))
|
|
return SimpleNamespace(status_code=200)
|
|
|
|
|
|
class HttpTransportTests(unittest.TestCase):
|
|
def test_cli_exposes_explicit_transport_ab_switch(self):
|
|
from tools.sms_login_cli import build_parser
|
|
|
|
default_args = build_parser().parse_args(["--mobile", "13800000000"])
|
|
okhttp_args = build_parser().parse_args(
|
|
["--mobile", "13800000000", "--transport", "okhttp4-android10"]
|
|
)
|
|
|
|
self.assertEqual(default_args.transport, "requests")
|
|
self.assertEqual(okhttp_args.transport, "okhttp4-android10")
|
|
|
|
def test_okhttp_transport_forces_http2_without_browser_headers(self):
|
|
inner = _FakeCurlSession()
|
|
|
|
session = create_http_session("okhttp4-android10", curl_session=inner)
|
|
session.post(
|
|
"https://HOST/rest/fixture",
|
|
data=b"a=1",
|
|
headers={
|
|
"User-Agent": "kwai-android",
|
|
"Accept-Encoding": "gzip",
|
|
"Connection": "keep-alive",
|
|
},
|
|
timeout=(5, 20),
|
|
)
|
|
|
|
method, url, kwargs = inner.calls[0]
|
|
self.assertEqual(method, "POST")
|
|
self.assertEqual(url, "https://HOST/rest/fixture")
|
|
self.assertEqual(kwargs["http_version"], "v2")
|
|
self.assertFalse(kwargs["default_headers"])
|
|
self.assertIsNone(kwargs["accept_encoding"])
|
|
self.assertNotIn("Connection", kwargs["headers"])
|
|
|
|
def test_okhttp_transport_cookie_adapter_accepts_browser_expiry(self):
|
|
inner = _FakeCurlSession()
|
|
session = create_http_session("okhttp4-android10", curl_session=inner)
|
|
|
|
session.cookies.set(
|
|
"did",
|
|
"ANDROID_FIXTURE",
|
|
domain=".example.test",
|
|
path="/",
|
|
secure=True,
|
|
expires=1_900_000_000,
|
|
)
|
|
|
|
self.assertEqual(
|
|
inner.cookies.calls,
|
|
[("did", "ANDROID_FIXTURE", ".example.test", "/", True)],
|
|
)
|
|
|
|
def test_okhttp_transport_builds_documented_fingerprint(self):
|
|
captured = {}
|
|
|
|
def factory(**kwargs):
|
|
captured.update(kwargs)
|
|
return _FakeCurlSession()
|
|
|
|
create_http_session("okhttp4-android10", curl_session_factory=factory)
|
|
|
|
self.assertEqual(captured["ja3"], OKHTTP4_ANDROID10_JA3)
|
|
self.assertEqual(captured["akamai"], OKHTTP4_ANDROID10_AKAMAI)
|
|
self.assertFalse(captured["default_headers"])
|
|
self.assertIn("tls_signature_algorithms", captured["extra_fp"])
|
|
|
|
def test_requests_transport_removes_requests_only_accept_header(self):
|
|
session = create_http_session("requests")
|
|
|
|
self.assertNotIn("Accept", session.headers)
|
|
|
|
def test_unknown_transport_is_rejected(self):
|
|
with self.assertRaisesRegex(ValueError, "unsupported HTTP transport"):
|
|
create_http_session("unknown")
|
|
|
|
def test_protocol_diagnosis_reads_curl_http2_response(self):
|
|
from core.sms_login import _do_post
|
|
|
|
class CurlHttpVersion(IntEnum):
|
|
V2_0 = 3
|
|
|
|
response = SimpleNamespace(
|
|
status_code=200,
|
|
text='{"result": 705}',
|
|
headers={},
|
|
cookies={},
|
|
request=None,
|
|
http_version=CurlHttpVersion.V2_0,
|
|
)
|
|
|
|
result = _do_post(
|
|
lambda *args, **kwargs: response,
|
|
"https://HOST/rest/fixture",
|
|
b"a=1",
|
|
base_url="https://HOST",
|
|
timeout=20,
|
|
)
|
|
|
|
self.assertEqual(result["request_meta"]["http_version"], "HTTP/2")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|