feat(signature): 引入 SessionManager 以优化风控参数 #87
Merged
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
- 引入 SessionManager 来管理和维护一个模拟会话的状态，包括页面加载时间戳和各种计数器。 - 重构签名方法以接受可选的 SessionManager 实例，从而生成更逼真的、包含单调递增计数器的签名。 - 添加了相应的单元测试来验证新会话管理器的功能和向后兼容性。
审查者指南通过引入 SessionManager / SignState 抽象，并在客户端签名 API 中贯穿一个可选的 session 参数直到加密负载构建器，实现了“会话感知”的签名流程。这样一来，签名既可以像之前一样以无状态方式生成，也可以基于每个会话的确定性状态生成，并添加了测试覆盖这两种行为。 带可选 SessionManager 的会话感知签名流程时序图sequenceDiagram
actor UserCode
participant Xhshow
participant SessionManager
participant CryptoProcessor
participant BitOps
participant B64Encoder
UserCode->>Xhshow: sign_headers(method, uri, cookies, xsec_appid, params/payload, timestamp, session)
Xhshow->>Xhshow: _parse_cookies(cookies)
Xhshow->>Xhshow: extract a1 from cookie_dict
alt session provided
Xhshow->>SessionManager: get_current_state(uri)
SessionManager-->>Xhshow: SignState(page_load_timestamp, sequence_value, window_props_length, uri_length)
Xhshow->>CryptoProcessor: build_payload_array(d_value, a1_value, xsec_appid, content_string, timestamp, sign_state)
else no session
Xhshow->>CryptoProcessor: build_payload_array(d_value, a1_value, xsec_appid, content_string, timestamp, sign_state=None)
end
CryptoProcessor->>CryptoProcessor: env_fingerprint_a(timestamp)
alt sign_state provided
CryptoProcessor->>CryptoProcessor: env_fingerprint_b(sign_state.page_load_timestamp)
CryptoProcessor->>CryptoProcessor: use sign_state.sequence_value
CryptoProcessor->>CryptoProcessor: use sign_state.window_props_length
CryptoProcessor->>CryptoProcessor: use sign_state.uri_length
else no sign_state
CryptoProcessor->>CryptoProcessor: generate_random_byte_in_range for time_offset
CryptoProcessor->>CryptoProcessor: env_fingerprint_b(timestamp - time_offset)
CryptoProcessor->>CryptoProcessor: generate_random_byte_in_range for sequence_value
CryptoProcessor->>CryptoProcessor: generate_random_byte_in_range for window_props_length
CryptoProcessor->>CryptoProcessor: len(string_param) for uri_length
end
CryptoProcessor-->>Xhshow: payload_array
Xhshow->>BitOps: xor_transform_array(payload_array)
BitOps-->>Xhshow: xor_result
Xhshow->>B64Encoder: encode_x3(xor_result[0..123])
B64Encoder-->>Xhshow: x3_signature
Xhshow->>Xhshow: build signature_data using SIGNATURE_DATA_TEMPLATE and x3_signature
Xhshow->>B64Encoder: encode(json.dumps(signature_data))
B64Encoder-->>Xhshow: encoded_signature
Xhshow-->>UserCode: headers with x_s, x_s_common, x_t, trace ids
SessionManager、SignState、Xhshow 和 CryptoProcessor 的类图classDiagram
class SignState {
<<NamedTuple>>
+int page_load_timestamp
+int sequence_value
+int window_props_length
+int uri_length
}
class SessionManager {
-int page_load_timestamp
-int sequence_value
-int window_props_length
+SessionManager()
+update_state() void
+get_current_state(uri str) SignState
}
class CryptoProcessor {
+config
+bit_ops
+b64encoder
+random_gen
+build_payload_array(d_value str, a1_value str, app_identifier str, string_param str, timestamp float, sign_state SignState) list~int~
+env_fingerprint_a(timestamp_ms int) list~int~
+env_fingerprint_b(timestamp_ms int) list~int~
+_int_to_le_bytes(value int, length int) list~int~
}
class Xhshow {
+crypto_processor CryptoProcessor
+sign_xs(method str, uri str, a1_value str, xsec_appid str, payload dict~str,Any~, timestamp float, session SessionManager) str
+sign_xs_get(uri str, a1_value str, xsec_appid str, params dict~str,Any~, timestamp float, session SessionManager) str
+sign_xs_post(uri str, a1_value str, xsec_appid str, payload dict~str,Any~, timestamp float, session SessionManager) str
+sign_headers(method str, uri str, cookies dict~str,str~, xsec_appid str, params dict~str,Any~, payload dict~str,Any~, timestamp float, session SessionManager) dict~str,str~
+sign_headers_get(uri str, cookies dict~str,str~, xsec_appid str, params dict~str,Any~, timestamp float, session SessionManager) dict~str,str~
+sign_headers_post(uri str, cookies dict~str,str~, xsec_appid str, payload dict~str,Any~, timestamp float, session SessionManager) dict~str,str~
+_build_content_string(method str, uri str, payload dict~str,Any~) str
+_generate_d_value(content_string str) str
+_parse_cookies(cookies dict~str,str~) dict~str,str~
}
class RequestSignatureValidator {
+__call__(...) bool
+validate_request_signature_params(...)
+validate_get_signature_params(...)
+validate_post_signature_params(...)
}
class validate_xs_common_params {
<<decorator>>
}
class validate_get_signature_params {
<<decorator>>
}
class validate_post_signature_params {
<<decorator>>
}
SessionManager --> SignState : creates
Xhshow --> CryptoProcessor : uses
Xhshow --> SessionManager : optional session
Xhshow --> SignState : passes to build_payload_array
CryptoProcessor ..> SignState : TYPE_CHECKING reference
validate_get_signature_params ..> RequestSignatureValidator : uses
validate_post_signature_params ..> RequestSignatureValidator : uses
validate_xs_common_params ..> RequestSignatureValidator : uses
文件级变更
提示与命令与 Sourcery 交互
自定义你的体验访问你的 控制面板 以：
获得帮助Original review guide in EnglishReviewer's GuideIntroduces a session-aware signing flow by adding a SessionManager/SignState abstraction and threading an optional session parameter through the client signing APIs into the crypto payload builder so that signatures can be generated either in the original stateless way or using deterministic per-session state, with tests added to cover both behaviors. Sequence diagram for session-aware signing flow with optional SessionManagersequenceDiagram
actor UserCode
participant Xhshow
participant SessionManager
participant CryptoProcessor
participant BitOps
participant B64Encoder
UserCode->>Xhshow: sign_headers(method, uri, cookies, xsec_appid, params/payload, timestamp, session)
Xhshow->>Xhshow: _parse_cookies(cookies)
Xhshow->>Xhshow: extract a1 from cookie_dict
alt session provided
Xhshow->>SessionManager: get_current_state(uri)
SessionManager-->>Xhshow: SignState(page_load_timestamp, sequence_value, window_props_length, uri_length)
Xhshow->>CryptoProcessor: build_payload_array(d_value, a1_value, xsec_appid, content_string, timestamp, sign_state)
else no session
Xhshow->>CryptoProcessor: build_payload_array(d_value, a1_value, xsec_appid, content_string, timestamp, sign_state=None)
end
CryptoProcessor->>CryptoProcessor: env_fingerprint_a(timestamp)
alt sign_state provided
CryptoProcessor->>CryptoProcessor: env_fingerprint_b(sign_state.page_load_timestamp)
CryptoProcessor->>CryptoProcessor: use sign_state.sequence_value
CryptoProcessor->>CryptoProcessor: use sign_state.window_props_length
CryptoProcessor->>CryptoProcessor: use sign_state.uri_length
else no sign_state
CryptoProcessor->>CryptoProcessor: generate_random_byte_in_range for time_offset
CryptoProcessor->>CryptoProcessor: env_fingerprint_b(timestamp - time_offset)
CryptoProcessor->>CryptoProcessor: generate_random_byte_in_range for sequence_value
CryptoProcessor->>CryptoProcessor: generate_random_byte_in_range for window_props_length
CryptoProcessor->>CryptoProcessor: len(string_param) for uri_length
end
CryptoProcessor-->>Xhshow: payload_array
Xhshow->>BitOps: xor_transform_array(payload_array)
BitOps-->>Xhshow: xor_result
Xhshow->>B64Encoder: encode_x3(xor_result[0..123])
B64Encoder-->>Xhshow: x3_signature
Xhshow->>Xhshow: build signature_data using SIGNATURE_DATA_TEMPLATE and x3_signature
Xhshow->>B64Encoder: encode(json.dumps(signature_data))
B64Encoder-->>Xhshow: encoded_signature
Xhshow-->>UserCode: headers with x_s, x_s_common, x_t, trace ids
Class diagram for SessionManager, SignState, Xhshow, and CryptoProcessorclassDiagram
class SignState {
<<NamedTuple>>
+int page_load_timestamp
+int sequence_value
+int window_props_length
+int uri_length
}
class SessionManager {
-int page_load_timestamp
-int sequence_value
-int window_props_length
+SessionManager()
+update_state() void
+get_current_state(uri str) SignState
}
class CryptoProcessor {
+config
+bit_ops
+b64encoder
+random_gen
+build_payload_array(d_value str, a1_value str, app_identifier str, string_param str, timestamp float, sign_state SignState) list~int~
+env_fingerprint_a(timestamp_ms int) list~int~
+env_fingerprint_b(timestamp_ms int) list~int~
+_int_to_le_bytes(value int, length int) list~int~
}
class Xhshow {
+crypto_processor CryptoProcessor
+sign_xs(method str, uri str, a1_value str, xsec_appid str, payload dict~str,Any~, timestamp float, session SessionManager) str
+sign_xs_get(uri str, a1_value str, xsec_appid str, params dict~str,Any~, timestamp float, session SessionManager) str
+sign_xs_post(uri str, a1_value str, xsec_appid str, payload dict~str,Any~, timestamp float, session SessionManager) str
+sign_headers(method str, uri str, cookies dict~str,str~, xsec_appid str, params dict~str,Any~, payload dict~str,Any~, timestamp float, session SessionManager) dict~str,str~
+sign_headers_get(uri str, cookies dict~str,str~, xsec_appid str, params dict~str,Any~, timestamp float, session SessionManager) dict~str,str~
+sign_headers_post(uri str, cookies dict~str,str~, xsec_appid str, payload dict~str,Any~, timestamp float, session SessionManager) dict~str,str~
+_build_content_string(method str, uri str, payload dict~str,Any~) str
+_generate_d_value(content_string str) str
+_parse_cookies(cookies dict~str,str~) dict~str,str~
}
class RequestSignatureValidator {
+__call__(...) bool
+validate_request_signature_params(...)
+validate_get_signature_params(...)
+validate_post_signature_params(...)
}
class validate_xs_common_params {
<<decorator>>
}
class validate_get_signature_params {
<<decorator>>
}
class validate_post_signature_params {
<<decorator>>
}
SessionManager --> SignState : creates
Xhshow --> CryptoProcessor : uses
Xhshow --> SessionManager : optional session
Xhshow --> SignState : passes to build_payload_array
CryptoProcessor ..> SignState : TYPE_CHECKING reference
validate_get_signature_params ..> RequestSignatureValidator : uses
validate_post_signature_params ..> RequestSignatureValidator : uses
validate_xs_common_params ..> RequestSignatureValidator : uses
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey，我发现了 8 个问题，并且给出了一些总体反馈：
- 在
src/xhshow/client.py中，__all__包含"SignState"，但该符号在此模块中既没有被导入也没有被定义，因此from xhshow.client import SignState会失败；要么从.session中导入它，要么从__all__中移除。 SessionManager的 API 有点令人困惑：文档里说调用方应该在签名前调用update_state()，但get_current_state()内部也会调用update_state()，如果两者都被使用就会导致状态被更新两次；可以考虑只在一个地方显式推进状态（例如仅在update_state()中，而不是在get_current_state()内部），或者调整 docstring 来澄清预期用法。- 在
tests/test_url_utils.py中的test_client_sign_with_session里，print(signature)会在测试运行期间产生噪声输出，并且没有用于断言；可以安全地移除。
面向 AI Agent 的提示
请根据本次代码评审中的评论进行修改：
## 总体评论
- 在 `src/xhshow/client.py` 中，`__all__` 包含 `"SignState"`，但该符号在此模块中既没有被导入也没有被定义，因此 `from xhshow.client import SignState` 会失败；要么从 `.session` 中导入它，要么从 `__all__` 中移除。
- `SessionManager` 的 API 有点令人困惑：文档里说调用方应该在签名前调用 `update_state()`，但 `get_current_state()` 内部也会调用 `update_state()`，如果两者都被使用就会导致状态被更新两次；可以考虑只在一个地方显式推进状态（例如仅在 `update_state()` 中，而不是在 `get_current_state()` 内部），或者调整 docstring 来澄清预期用法。
- 在 `tests/test_url_utils.py` 中的 `test_client_sign_with_session` 里，`print(signature)` 会在测试运行期间产生噪声输出，并且没有用于断言；可以安全地移除。
## 逐条评论
### Comment 1
<location> `src/xhshow/client.py:23` </location>
<code_context>
+if TYPE_CHECKING:
+ from .session import SessionManager
+
+__all__ = ["Xhshow", "SessionManager", "SignState"]
</code_context>
<issue_to_address>
**issue (bug_risk):** SessionManager 和 SignState 被列在 __all__ 中，但在这个模块里实际上从未绑定。
由于这些名字只在 `TYPE_CHECKING` 分支下被导入，它们在运行时并不存在于 `xhshow.client` 上，所以尽管它们在 `__all__` 中，`from xhshow.client import SessionManager` 仍然会失败。请在运行时导入并重新导出它们，或者把它们从 `__all__` 中移除，以保证公共 API 与实际可用的内容保持一致。
</issue_to_address>
### Comment 2
<location> `src/xhshow/client.py:148` </location>
<code_context>
- signature_data["x3"] = self.crypto_processor.config.X3_PREFIX + self._build_signature(
- d_value, a1_value, xsec_appid, content_string, timestamp
+
+ sign_state = session.get_current_state(content_string) if session else None
+
+ payload_array = self.crypto_processor.build_payload_array(
</code_context>
<issue_to_address>
**issue (bug_risk):** 对 get_current_state 使用 content_string 而不是裸 URI，可能会破坏预期的 URI 长度建模。
`SessionManager.get_current_state` 期望的是原始 URI，并使用 `len(uri)` 作为 `SignState.uri_length`。传入 `content_string`（它把 method/URI/payload 组合在一起）意味着 `uri_length` 将不再匹配真实 URI。为了让 URI 长度建模保持准确并与 docstring 一致，这里应该传入真实的 `uri`，而 `_build_content_string` 可以继续单独使用组合后的值。
</issue_to_address>
### Comment 3
<location> `src/xhshow/session.py:35-38` </location>
<code_context>
+
+ This method should be called before each signing operation.
+ """
+ # Simulate realistic counter increments
+ # self.sequence_value += random.randint(0, 1)
+ self.window_props_length += random.randint(1, 10)
+ self.uri_length += random.randint(0, 2)
</code_context>
<issue_to_address>
**suggestion (bug_risk):** sequence_value 从未被更新，这削弱了将其作为会话状态演化部分进行追踪的意义。
`SignState` 和 docstring 表明 `sequence_value` 应该在每个请求中发生变化，但 `update_state` 只更新了 `window_props_length` 和 `uri_length`。由于递增操作被注释掉，同一会话中的所有签名操作都会使用相同的 `sequence_value`。如果这不是有意为之，请重新启用递增（如有需要可添加适当的边界）以便状态在每次调用之间真正演化。
```suggestion
# Simulate realistic counter increments
self.sequence_value += random.randint(0, 1)
self.window_props_length += random.randint(1, 10)
self.uri_length += random.randint(0, 2)
```
</issue_to_address>
### Comment 4
<location> `src/xhshow/session.py:27` </location>
<code_context>
+ self.page_load_timestamp: int = int(time.time() * 1000)
+ self.sequence_value: int = random.randint(15, 17)
+ self.window_props_length: int = random.randint(1000, 2000)
+ self.uri_length: int = random.randint(200, 400)
+
+ def update_state(self):
</code_context>
<issue_to_address>
**suggestion:** 存储的 uri_length 计数实际上没有被有效使用，因为 get_current_state 总是用 len(uri)。
`uri_length` 在初始化时被设置，并在 `update_state` 中更新，但 `get_current_state` 总是设置 `uri_length=len(uri)` 而不是使用存储的值，使得该字段实际上是多余的。请要么移除 `uri_length` 及其更新逻辑，要么修改 `get_current_state` 从 `self.uri_length` 读取（可以选择与 `len(uri)` 一起使用），如果它本来就应该代表演化中的 URI 状态。
建议实现：
```python
self.page_load_timestamp: int = int(time.time() * 1000)
self.sequence_value: int = random.randint(15, 17)
self.window_props_length: int = random.randint(1000, 2000)
```
```python
# Simulate realistic counter increments
# self.sequence_value += random.randint(0, 1)
self.window_props_length += random.randint(1, 10)
```
在 `src/xhshow/session.py` 的其余部分（以及所有调用方）中搜索对 `uri_length` 的剩余引用（例如作为属性、在 `get_current_state` 中或在类型注解中）并将其移除。之后，确保 `get_current_state`（或等价方法）只使用 `len(uri)` 或其他可用字段，而不会再依赖 `self.uri_length` 的存在。
</issue_to_address>
### Comment 5
<location> `src/xhshow/client.py:120-121` </location>
<code_context>
timestamp: Unix timestamp in seconds (defaults to current time)
+ session: Optional session manager for stateful signing.
Returns:
str: Complete signature string
</code_context>
<issue_to_address>
**nitpick (typo):** sign_headers_post 的 docstring 在 x-b3-traceid 命名上与其他方法不一致。
请更新 docstring，使其使用 `x-b3-traceid`（如果这仍然是实际的 header 名称），这样该 header 名称在所有辅助方法中保持一致，从而让 API 文档更清晰。
```suggestion
"""
Generate request signature (supports GET and POST) for use in the
`x-b3-traceid` header.
- GET request: params value
- POST request: payload value
timestamp: Unix timestamp in seconds (defaults to current time)
session: Optional session manager for stateful signing.
```
</issue_to_address>
### Comment 6
<location> `tests/test_url_utils.py:154-163` </location>
<code_context>
params="invalid", # type: ignore
)
+
+ def test_client_sign_with_session(self):
+ client = Xhshow()
+ session = SessionManager()
+ for i in range(10):
+ signature = client.sign_xs_get(
+ uri="/api/sns/web/v1/user_posted",
+ a1_value="test_a1_value",
+ params={"num": "30"},
+ session=session,
+ )
+ session.update_state()
+ print(signature)
+
+ assert signature.startswith("XYS_")
</code_context>
<issue_to_address>
**suggestion (testing):** 加强基于 Session 的签名测试并避免副作用。
这个测试只断言签名以 `"XYS_"` 开头，而这一点已经由实现保证了，并不能验证基于 Session 的签名与无状态情况之间的差异。它同时在循环中调用 `session.update_state()`（尽管 `SessionManager.get_current_state` 已经会推进状态），并打印签名。
建议：
- 去掉循环中的显式 `session.update_state()` 调用（或者清楚地说明为何需要两套机制同时存在）。
- 移除 `print(signature)`，以避免测试中产生噪声副作用。
- 加强断言，例如收集所有签名并断言它们并非全部相同，或者检查某些依赖状态的属性。也可以对比有无 `SessionManager` 时的行为，以确认新流程被真正覆盖。
</issue_to_address>
### Comment 7
<location> `tests/test_session.py:48-57` </location>
<code_context>
+ cookies = {"a1": "test_a1", "web_session": "test_session"}
+ content_string = uri # Simplified for this test
+
+ # Perform signing
+ client.sign_headers_get(uri=uri, cookies=cookies, session=session)
+
+ # Get the expected state from the session
+ expected_state = session.get_current_state(content_string)
+
+ # Assert that build_payload_array was called
+ mock_crypto_processor.build_payload_array.assert_called_once()
+
+ # Get the actual arguments passed to the mock
+ _, kwargs = mock_crypto_processor.build_payload_array.call_args
+ actual_state = kwargs.get("sign_state")
+
+ # Verify the state matches
+ assert actual_state is not None
+ assert actual_state.page_load_timestamp == expected_state.page_load_timestamp
+ assert actual_state.sequence_value == expected_state.sequence_value
+ assert actual_state.window_props_length == expected_state.window_props_length
+ assert actual_state.uri_length == len(content_string)
+
+
</code_context>
<issue_to_address>
**issue (testing):** 为了避免测试不稳定，不要在计算 expected_state 时再次调用 `get_current_state`。
`client.sign_headers_get(..., session=session)` 已经调用过一次 `session.get_current_state(content_string)`，并通过 `update_state` 改变了状态。为了得到 `expected_state` 再调用一次 `session.get_current_state(content_string)` 会第二次推进状态，因此 `expected_state` 可能与实际传给 `build_payload_array` 的值不一致，一旦 `update_state` 的逻辑发生演化，就可能导致测试不稳定。
与其重新计算 `expected_state`，不如直接对 `actual_state` 进行断言，例如：
- `assert isinstance(actual_state, SignState)`
- `assert actual_state.page_load_timestamp == session.page_load_timestamp`
- `assert actual_state.uri_length == len(content_string)`
- 还可以可选地检查单调属性，例如 `window_props_length` / `uri_length` 相对于初始值是否 `>=`。
这样可以在不依赖额外有副作用调用的前提下，验证状态传递是否正确。
</issue_to_address>
### Comment 8
<location> `tests/test_session.py:69-78` </location>
<code_context>
+ assert actual_state.uri_length == len(content_string)
+
+
+def test_signing_without_session(mock_crypto_processor):
+ """
+ Verify that signing falls back to the old method when no session is provided.
+ """
+ client = Xhshow()
+ uri = "/api/sns/web/v1/user/posted"
+ cookies = {"a1": "test_a1", "web_session": "test_session"}
+
+ # Perform signing without a session
+ client.sign_headers_get(uri=uri, cookies=cookies)
+
+ # Assert that build_payload_array was called
+ mock_crypto_processor.build_payload_array.assert_called_once()
+
+ # Get the actual arguments passed to the mock
+ _, kwargs = mock_crypto_processor.build_payload_array.call_args
+ actual_state = kwargs.get("sign_state")
+
+ # Verify that no state was passed (it should be None)
+ assert actual_state is None
</code_context>
<issue_to_address>
**suggestion (testing):** 建议为使用 `SessionManager` 的 POST 变体增加测试。
当前测试只覆盖了带 / 不带 `SessionManager` 的 `sign_headers_get`，但这个 PR 也为 `sign_headers_post` / `sign_xs_post` 添加了 `session` 支持。请增加对应的 POST 测试：
- `test_signing_with_session_post`：在调用 `sign_headers_post(..., session=session)` 时，断言 `sign_state` 被传递给 `build_payload_array`。
- `test_signing_without_session_post`：在调用 `sign_headers_post(..., session=None)` 时，断言 `sign_state` 为 `None`。
这将同时验证 POST 请求中新引入的 Session 行为。
建议实现：
```python
# Get the actual arguments passed to the mock
_, kwargs = mock_crypto_processor.build_payload_array.call_args
actual_state = kwargs.get("sign_state")
# Verify that no state was passed (it should be None)
assert actual_state is None
def test_signing_with_session_post(mock_crypto_processor, mock_session_manager):
"""
Verify that signing uses the session state when provided for POST requests.
"""
client = Xhshow()
uri = "/api/sns/web/v1/user/posted"
cookies = {"a1": "test_a1", "web_session": "test_session"}
data = {"foo": "bar"}
# Obtain a session (using the same pattern as the GET tests)
session = mock_session_manager.get_or_create_session.return_value
# Perform signing with a session for POST
client.sign_headers_post(uri=uri, data=data, cookies=cookies, session=session)
# Assert that build_payload_array was called
mock_crypto_processor.build_payload_array.assert_called_once()
# Get the actual arguments passed to the mock
_, kwargs = mock_crypto_processor.build_payload_array.call_args
actual_state = kwargs.get("sign_state")
# Verify that session state was passed
assert actual_state is not None
def test_signing_without_session_post(mock_crypto_processor):
"""
Verify that signing falls back to the old method when no session is provided for POST requests.
"""
client = Xhshow()
uri = "/api/sns/web/v1/user/posted"
cookies = {"a1": "test_a1", "web_session": "test_session"}
data = {"foo": "bar"}
# Perform signing without a session for POST
client.sign_headers_post(uri=uri, data=data, cookies=cookies, session=None)
# Assert that build_payload_array was called
mock_crypto_processor.build_payload_array.assert_called_once()
# Get the actual arguments passed to the mock
_, kwargs = mock_crypto_processor.build_payload_array.call_args
actual_state = kwargs.get("sign_state")
# Verify that no state was passed (it should be None)
assert actual_state is None
```
如果现有的 GET 测试使用了不同的 fixture 名称或获取 Session 的方式（例如 `session = mock_session_manager.return_value` 或单独的 `session` fixture），请相应地调整 `test_signing_with_session_post` 的参数和 Session 初始化方式，以与文件中现有约定保持一致，使这些测试能够无缝集成。
</issue_to_address>帮我变得更有用！请对每条评论点 👍 或 👎，我会根据反馈改进后续的评审质量。
Original comment in English
Hey - I've found 8 issues, and left some high level feedback:
- In
src/xhshow/client.py,__all__includes"SignState"but the symbol is never imported or defined in this module, sofrom xhshow.client import SignStatewill fail; either import it from.sessionor remove it from__all__. - The
SessionManagerAPI is a bit confusing:update_state()is documented as something the caller should invoke before signing, butget_current_state()also callsupdate_state()internally, which can lead to double updates if both are used; consider making state advancement explicit in one place (e.g., only inupdate_state()and not insideget_current_state()) or adjusting the docstrings to clarify the intended usage. - In
test_client_sign_with_session(intests/test_url_utils.py), theprint(signature)call introduces noisy output during test runs and isn’t used in assertions; it can be safely removed.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `src/xhshow/client.py`, `__all__` includes `"SignState"` but the symbol is never imported or defined in this module, so `from xhshow.client import SignState` will fail; either import it from `.session` or remove it from `__all__`.
- The `SessionManager` API is a bit confusing: `update_state()` is documented as something the caller should invoke before signing, but `get_current_state()` also calls `update_state()` internally, which can lead to double updates if both are used; consider making state advancement explicit in one place (e.g., only in `update_state()` and not inside `get_current_state()`) or adjusting the docstrings to clarify the intended usage.
- In `test_client_sign_with_session` (in `tests/test_url_utils.py`), the `print(signature)` call introduces noisy output during test runs and isn’t used in assertions; it can be safely removed.
## Individual Comments
### Comment 1
<location> `src/xhshow/client.py:23` </location>
<code_context>
+if TYPE_CHECKING:
+ from .session import SessionManager
+
+__all__ = ["Xhshow", "SessionManager", "SignState"]
</code_context>
<issue_to_address>
**issue (bug_risk):** SessionManager and SignState are listed in __all__ but never actually bound in this module.
Because these names are only imported under `TYPE_CHECKING`, they don’t exist at runtime on `xhshow.client`, so `from xhshow.client import SessionManager` will fail despite being in `__all__`. Either import and re-export them at runtime, or remove them from `__all__` to keep the public API consistent with what’s actually available.
</issue_to_address>
### Comment 2
<location> `src/xhshow/client.py:148` </location>
<code_context>
- signature_data["x3"] = self.crypto_processor.config.X3_PREFIX + self._build_signature(
- d_value, a1_value, xsec_appid, content_string, timestamp
+
+ sign_state = session.get_current_state(content_string) if session else None
+
+ payload_array = self.crypto_processor.build_payload_array(
</code_context>
<issue_to_address>
**issue (bug_risk):** Using content_string instead of the bare URI for get_current_state likely breaks the intended URI length modeling.
`SessionManager.get_current_state` expects the raw URI and uses `len(uri)` for `SignState.uri_length`. Passing `content_string` (which combines method/URI/payload) means `uri_length` no longer matches the actual URI. To keep the URI length modeling accurate and consistent with the docstring, this should receive the real `uri`, while `_build_content_string` can continue to use the composed value separately.
</issue_to_address>
### Comment 3
<location> `src/xhshow/session.py:35-38` </location>
<code_context>
+
+ This method should be called before each signing operation.
+ """
+ # Simulate realistic counter increments
+ # self.sequence_value += random.randint(0, 1)
+ self.window_props_length += random.randint(1, 10)
+ self.uri_length += random.randint(0, 2)
</code_context>
<issue_to_address>
**suggestion (bug_risk):** sequence_value is never updated, which undermines the purpose of tracking it as evolving session state.
`SignState` and the docstring imply `sequence_value` should change per request, but `update_state` only updates `window_props_length` and `uri_length`. With the increment commented out, all sign operations in a session use the same `sequence_value`. If this isn’t intentional, re-enable the increment (with appropriate bounds if needed) so the state actually evolves between calls.
```suggestion
# Simulate realistic counter increments
self.sequence_value += random.randint(0, 1)
self.window_props_length += random.randint(1, 10)
self.uri_length += random.randint(0, 2)
```
</issue_to_address>
### Comment 4
<location> `src/xhshow/session.py:27` </location>
<code_context>
+ self.page_load_timestamp: int = int(time.time() * 1000)
+ self.sequence_value: int = random.randint(15, 17)
+ self.window_props_length: int = random.randint(1000, 2000)
+ self.uri_length: int = random.randint(200, 400)
+
+ def update_state(self):
</code_context>
<issue_to_address>
**suggestion:** The stored uri_length counter is effectively unused because get_current_state always uses len(uri).
`uri_length` is initialized and updated in `update_state`, but `get_current_state` always sets `uri_length=len(uri)` instead of using the stored value, making the field effectively redundant. Please either remove `uri_length` and its updates, or change `get_current_state` to read from `self.uri_length` (optionally alongside `len(uri)`) if it’s meant to represent evolving URI state.
Suggested implementation:
```python
self.page_load_timestamp: int = int(time.time() * 1000)
self.sequence_value: int = random.randint(15, 17)
self.window_props_length: int = random.randint(1000, 2000)
```
```python
# Simulate realistic counter increments
# self.sequence_value += random.randint(0, 1)
self.window_props_length += random.randint(1, 10)
```
Search the rest of `src/xhshow/session.py` (and any callers) for any remaining references to `uri_length` (e.g., as an attribute, in `get_current_state`, or in type hints) and remove them. After that, ensure that `get_current_state` (or any equivalent method) only uses `len(uri)` or other available fields, without expecting `self.uri_length` to exist.
</issue_to_address>
### Comment 5
<location> `src/xhshow/client.py:120-121` </location>
<code_context>
timestamp: Unix timestamp in seconds (defaults to current time)
+ session: Optional session manager for stateful signing.
Returns:
str: Complete signature string
</code_context>
<issue_to_address>
**nitpick (typo):** Docstring for sign_headers_post is inconsistent with other methods regarding x-b3-traceid naming.
Please update the docstring to use `x-b3-traceid` (if that’s still the actual header name) so the header name is consistent across all helper methods and the API docs remain clear.
```suggestion
"""
Generate request signature (supports GET and POST) for use in the
`x-b3-traceid` header.
- GET request: params value
- POST request: payload value
timestamp: Unix timestamp in seconds (defaults to current time)
session: Optional session manager for stateful signing.
```
</issue_to_address>
### Comment 6
<location> `tests/test_url_utils.py:154-163` </location>
<code_context>
params="invalid", # type: ignore
)
+
+ def test_client_sign_with_session(self):
+ client = Xhshow()
+ session = SessionManager()
+ for i in range(10):
+ signature = client.sign_xs_get(
+ uri="/api/sns/web/v1/user_posted",
+ a1_value="test_a1_value",
+ params={"num": "30"},
+ session=session,
+ )
+ session.update_state()
+ print(signature)
+
+ assert signature.startswith("XYS_")
</code_context>
<issue_to_address>
**suggestion (testing):** Strengthen the session-based signing test and avoid side effects
This test only asserts that the signature starts with `"XYS_"`, which is already guaranteed by the implementation and doesn’t verify that session-based signing differs from the stateless case. It also both calls `session.update_state()` (even though `SessionManager.get_current_state` already advances state) and prints the signature.
I’d recommend:
- Dropping the explicit `session.update_state()` in the loop (or clearly documenting why both mechanisms are needed).
- Removing `print(signature)` to avoid noisy side effects in tests.
- Strengthening assertions, e.g. collecting all signatures and asserting they’re not all identical, or otherwise checking a state-dependent property. Optionally, compare behavior with and without `SessionManager` to confirm the new flow is exercised.
</issue_to_address>
### Comment 7
<location> `tests/test_session.py:48-57` </location>
<code_context>
+ cookies = {"a1": "test_a1", "web_session": "test_session"}
+ content_string = uri # Simplified for this test
+
+ # Perform signing
+ client.sign_headers_get(uri=uri, cookies=cookies, session=session)
+
+ # Get the expected state from the session
+ expected_state = session.get_current_state(content_string)
+
+ # Assert that build_payload_array was called
+ mock_crypto_processor.build_payload_array.assert_called_once()
+
+ # Get the actual arguments passed to the mock
+ _, kwargs = mock_crypto_processor.build_payload_array.call_args
+ actual_state = kwargs.get("sign_state")
+
+ # Verify the state matches
+ assert actual_state is not None
+ assert actual_state.page_load_timestamp == expected_state.page_load_timestamp
+ assert actual_state.sequence_value == expected_state.sequence_value
+ assert actual_state.window_props_length == expected_state.window_props_length
+ assert actual_state.uri_length == len(content_string)
+
+
</code_context>
<issue_to_address>
**issue (testing):** Avoid flakiness by not calling `get_current_state` a second time when computing `expected_state`
`client.sign_headers_get(..., session=session)` already calls `session.get_current_state(content_string)`, which mutates state via `update_state`. Calling `session.get_current_state(content_string)` again for `expected_state` advances the state a second time, so `expected_state` may not match what was actually passed to `build_payload_array`, making the test flaky if `update_state` evolves.
Instead of recomputing `expected_state`, assert directly on `actual_state`, e.g.:
- `assert isinstance(actual_state, SignState)`
- `assert actual_state.page_load_timestamp == session.page_load_timestamp`
- `assert actual_state.uri_length == len(content_string)`
- Optionally check monotonic properties like `window_props_length` / `uri_length` being `>=` initial values.
That way the test verifies the state wiring without relying on an extra mutating call.
</issue_to_address>
### Comment 8
<location> `tests/test_session.py:69-78` </location>
<code_context>
+ assert actual_state.uri_length == len(content_string)
+
+
+def test_signing_without_session(mock_crypto_processor):
+ """
+ Verify that signing falls back to the old method when no session is provided.
+ """
+ client = Xhshow()
+ uri = "/api/sns/web/v1/user/posted"
+ cookies = {"a1": "test_a1", "web_session": "test_session"}
+
+ # Perform signing without a session
+ client.sign_headers_get(uri=uri, cookies=cookies)
+
+ # Assert that build_payload_array was called
+ mock_crypto_processor.build_payload_array.assert_called_once()
+
+ # Get the actual arguments passed to the mock
+ _, kwargs = mock_crypto_processor.build_payload_array.call_args
+ actual_state = kwargs.get("sign_state")
+
+ # Verify that no state was passed (it should be None)
+ assert actual_state is None
</code_context>
<issue_to_address>
**suggestion (testing):** Consider adding tests for POST variants using `SessionManager`
The current tests only cover `sign_headers_get` with/without `SessionManager`, but this PR also adds `session` support to `sign_headers_post` / `sign_xs_post`. Please add analogous POST tests:
- `test_signing_with_session_post`: assert `sign_state` is passed to `build_payload_array` when calling `sign_headers_post(..., session=session)`.
- `test_signing_without_session_post`: assert `sign_state` is `None` when calling `sign_headers_post(..., session=None)`.
This will validate the new session behavior for POST requests as well.
Suggested implementation:
```python
# Get the actual arguments passed to the mock
_, kwargs = mock_crypto_processor.build_payload_array.call_args
actual_state = kwargs.get("sign_state")
# Verify that no state was passed (it should be None)
assert actual_state is None
def test_signing_with_session_post(mock_crypto_processor, mock_session_manager):
"""
Verify that signing uses the session state when provided for POST requests.
"""
client = Xhshow()
uri = "/api/sns/web/v1/user/posted"
cookies = {"a1": "test_a1", "web_session": "test_session"}
data = {"foo": "bar"}
# Obtain a session (using the same pattern as the GET tests)
session = mock_session_manager.get_or_create_session.return_value
# Perform signing with a session for POST
client.sign_headers_post(uri=uri, data=data, cookies=cookies, session=session)
# Assert that build_payload_array was called
mock_crypto_processor.build_payload_array.assert_called_once()
# Get the actual arguments passed to the mock
_, kwargs = mock_crypto_processor.build_payload_array.call_args
actual_state = kwargs.get("sign_state")
# Verify that session state was passed
assert actual_state is not None
def test_signing_without_session_post(mock_crypto_processor):
"""
Verify that signing falls back to the old method when no session is provided for POST requests.
"""
client = Xhshow()
uri = "/api/sns/web/v1/user/posted"
cookies = {"a1": "test_a1", "web_session": "test_session"}
data = {"foo": "bar"}
# Perform signing without a session for POST
client.sign_headers_post(uri=uri, data=data, cookies=cookies, session=None)
# Assert that build_payload_array was called
mock_crypto_processor.build_payload_array.assert_called_once()
# Get the actual arguments passed to the mock
_, kwargs = mock_crypto_processor.build_payload_array.call_args
actual_state = kwargs.get("sign_state")
# Verify that no state was passed (it should be None)
assert actual_state is None
```
If the existing GET tests use a different fixture name or pattern for obtaining a session (e.g. `session = mock_session_manager.return_value` or a custom `session` fixture), adjust the `test_signing_with_session_post` signature and session initialization to match that existing convention so the tests integrate cleanly with the rest of the file.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Owner
|
感谢你的贡献! 我会在空闲时间处理这个pr |
Owner
|
我对这个 PR 做了些修改: 主要改动:
需要讨论的问题: 我改了
这个不一致不知道是不是故意的?如果不是,我可以改成统一的逻辑。担心改签名算法会影响现有功能。 如果有问题我可以先回退这部分修改。 |
Owner
|
@sourcery-ai review |
There was a problem hiding this comment.
Hey - 我发现了 3 个问题，并给出了一些总体性的反馈：
- client 模块中的
__all__ = ["Xhshow", "SessionManager", "SignState"]有些误导，因为SessionManager/SignState并没有在client.py中定义或重新导出；你可以选择在client.py中导入并重新导出它们，或者把它们从__all__中移除，仅在xhshow.session中保留。 SessionManager.get_current_state()已经会调用update_state()，但update_state()的文档字符串却建议在每次签名前都调用它；这很容易导致状态被重复递增（就像你测试中那样）——可以考虑把update_state()设为私有方法，或者更清晰地说明预期用法，避免调用方在调用get_current_state()之前再手动推进状态。
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The client module’s `__all__ = ["Xhshow", "SessionManager", "SignState"]` is misleading as `SessionManager`/`SignState` are not defined or re-exported in `client.py`; either import and re-export them there or drop them from `__all__` and keep them only in `xhshow.session`.
- `SessionManager.get_current_state()` already calls `update_state()`, but the docstring of `update_state()` suggests it should be called before each signing; this can easily lead to double increments (as in your tests)—consider making `update_state()` private or clarifying the expected usage so callers don’t manually advance state before calling `get_current_state()`.
## Individual Comments
### Comment 1
<location> `src/xhshow/session.py:26-35` </location>
<code_context>
+ def __init__(self):
+ self.page_load_timestamp: int = int(time.time() * 1000)
+ self.sequence_value: int = random.randint(15, 17)
+ self.window_props_length: int = random.randint(1000, 2000)
+
+ def update_state(self):
+ """
+ Updates the session state to simulate user activity between requests.
+
+ This method should be called before each signing operation.
+ """
+ self.sequence_value += random.randint(0, 1)
+ self.window_props_length += random.randint(1, 10)
+
+ def get_current_state(self, uri: str) -> SignState:
</code_context>
<issue_to_address>
**suggestion (bug_risk):** `window_props_length` 在长生命周期的会话中会无限增长，可能会严重偏离真实的范围。
由于 `window_props_length` 初始值在 `[1000, 2000]` 之间，并且每次调用 `update_state` 时都会加上 `random.randint(1, 10)`，在长时间或高并发的会话中，这个值会无界增长。如果这个字段是为了模拟真实客户端窗口相关的指标，可以考虑强制一个最大值（例如使用一个可配置的 `MAX_WINDOW_PROPS_LENGTH`），或者定期重新随机到一个合理范围，从而避免该值漂移到明显不真实的区间。
建议实现方式：
```python
"""
Manages the state for a simulated user session to generate more realistic signatures.
This class maintains counters that should persist and evolve across multiple requests
within the same logical session.
"""
WINDOW_PROPS_MIN_LENGTH: int = 1000
WINDOW_PROPS_MAX_LENGTH: int = 2000
def __init__(self):
self.page_load_timestamp: int = int(time.time() * 1000)
self.sequence_value: int = random.randint(15, 17)
self.window_props_length: int = random.randint(
self.WINDOW_PROPS_MIN_LENGTH,
self.WINDOW_PROPS_MAX_LENGTH,
)
```
```python
def update_state(self):
"""
Updates the session state to simulate user activity between requests.
This method should be called before each signing operation.
"""
self.sequence_value += random.randint(0, 1)
self.window_props_length += random.randint(1, 10)
# Prevent window_props_length from drifting unbounded by re-randomizing
# into a realistic range once it exceeds the configured maximum.
if self.window_props_length > self.WINDOW_PROPS_MAX_LENGTH:
self.window_props_length = random.randint(
self.WINDOW_PROPS_MIN_LENGTH,
self.WINDOW_PROPS_MAX_LENGTH,
)
```
</issue_to_address>
### Comment 2
<location> `tests/test_url_utils.py:154-163` </location>
<code_context>
params="invalid", # type: ignore
)
+
+ def test_client_sign_with_session(self):
+ client = Xhshow()
+ session = SessionManager()
+ for _ in range(10):
+ signature = client.sign_xs_get(
+ uri="/api/sns/web/v1/user_posted",
+ a1_value="test_a1_value",
+ params={"num": "30"},
+ session=session,
+ )
+ session.update_state()
+
+ assert signature.startswith("XYS_")
</code_context>
<issue_to_address>
**suggestion (testing):** 加强 `test_client_sign_with_session`，以更好地测试与会话相关的行为。
当前这个测试在每次迭代中会对会话状态进行两次推进（`sign_xs_get` → `get_current_state` → `update_state`，再加上显式的 `session.update_state()`），如果你只是想验证“带会话的签名可以工作”，这种做法会让测试意图变得不清晰。同时，它只检查最终一次签名的前缀，因此并没有真正验证与会话相关的行为。
可以考虑：
- 移除循环和额外的 `session.update_state()`，只保留一个简单的冒烟测试；或者
- 把它做成一个更强的集成测试：收集多次签名并断言在相同会话下多次调用的签名会发生变化，和/或像 `tests/test_session.py` 那样 mock `CryptoProcessor`，以断言在提供 `session` 时确实使用了 `sign_state`。
</issue_to_address>
### Comment 3
<location> `src/xhshow/client.py:147` </location>
<code_context>
+
+ sign_state = session.get_current_state(uri) if session else None
+
+ payload_array = self.crypto_processor.build_payload_array(
+ d_value, a1_value, xsec_appid, content_string, timestamp, sign_state=sign_state
)
</code_context>
<issue_to_address>
**issue (complexity):** 建议把构建底层 x3 签名的逻辑移回 `CryptoProcessor` 中的一个高层封装方法里，让 `Xhshow.sign_xs` 只负责组织请求/会话相关的数据。
你可以保留现在的会话感知行为，同时通过重新引入 `CryptoProcessor` 中更高层的抽象来避免在 `Xhshow.sign_xs` 中暴露底层加密细节。
### 1. 在 `CryptoProcessor` 中添加一个高层的 x3 构建方法
在 `core/crypto.py`（或者 `CryptoProcessor` 所在的位置）中新增一个方法，封装 payload 数组 + xor + base64 的流水线，并接受可选的会话状态：
```python
class CryptoProcessor:
# existing code...
def build_x3_signature(
self,
d_value: str,
a1_value: str,
xsec_appid: str,
content_string: str,
timestamp: float | None = None,
sign_state: SignState | None = None,
) -> str:
payload_array = self.build_payload_array(
d_value,
a1_value,
xsec_appid,
content_string,
timestamp,
sign_state=sign_state,
)
xor_result = self.bit_ops.xor_transform_array(payload_array)
return self.b64encoder.encode_x3(xor_result[:124])
```
这样既能保留新的会话感知逻辑，又能把底层细节从 `client.py` 中隐藏起来。
### 2. 将 `Xhshow.sign_xs` 简化回高层的编排逻辑
然后在 `Xhshow.sign_xs` 中，改为委托给这个新 helper，而不是在其中直接编排 `build_payload_array`/`xor`/`encode_x3`：
```python
@validate_signature_params
def sign_xs(
self,
method: Literal["GET", "POST"],
uri: str,
a1_value: str,
xsec_appid: str = "xhs-pc-web",
payload: dict[str, Any] | None = None,
timestamp: float | None = None,
session: SessionManager | None = None,
) -> str:
uri = extract_uri(uri)
content_string = self._build_content_string(method, uri, payload)
d_value = self._generate_d_value(content_string)
sign_state = session.get_current_state(uri) if session else None
x3_signature = self.crypto_processor.build_x3_signature(
d_value,
a1_value,
xsec_appid,
content_string,
timestamp,
sign_state=sign_state,
)
signature_data = self.crypto_processor.config.SIGNATURE_DATA_TEMPLATE.copy()
signature_data["x3"] = self.crypto_processor.config.X3_PREFIX + x3_signature
return self.crypto_processor.config.XYS_PREFIX + self.crypto_processor.b64encoder.encode(
json.dumps(signature_data, separators=(",", ":"), ensure_ascii=False)
)
```
这可以恢复之前的分层结构：
- `Xhshow` 负责请求层面的逻辑（URI、method、payload、session 查找）。
- `CryptoProcessor` 负责加密流水线，包括新的会话感知行为。
功能保持不变（包括 `sign_xs_get`、`sign_xs_post` 和 `sign_headers*` 中的会话传递），但复杂度被重新聚合回加密层中。
</issue_to_address>帮我变得更有用！请在每条评论上点 👍 或 👎，我会根据你的反馈改进后续的审查。
Original comment in English
Hey - I've found 3 issues, and left some high level feedback:
- The client module’s
__all__ = ["Xhshow", "SessionManager", "SignState"]is misleading asSessionManager/SignStateare not defined or re-exported inclient.py; either import and re-export them there or drop them from__all__and keep them only inxhshow.session. SessionManager.get_current_state()already callsupdate_state(), but the docstring ofupdate_state()suggests it should be called before each signing; this can easily lead to double increments (as in your tests)—consider makingupdate_state()private or clarifying the expected usage so callers don’t manually advance state before callingget_current_state().
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The client module’s `__all__ = ["Xhshow", "SessionManager", "SignState"]` is misleading as `SessionManager`/`SignState` are not defined or re-exported in `client.py`; either import and re-export them there or drop them from `__all__` and keep them only in `xhshow.session`.
- `SessionManager.get_current_state()` already calls `update_state()`, but the docstring of `update_state()` suggests it should be called before each signing; this can easily lead to double increments (as in your tests)—consider making `update_state()` private or clarifying the expected usage so callers don’t manually advance state before calling `get_current_state()`.
## Individual Comments
### Comment 1
<location> `src/xhshow/session.py:26-35` </location>
<code_context>
+ def __init__(self):
+ self.page_load_timestamp: int = int(time.time() * 1000)
+ self.sequence_value: int = random.randint(15, 17)
+ self.window_props_length: int = random.randint(1000, 2000)
+
+ def update_state(self):
+ """
+ Updates the session state to simulate user activity between requests.
+
+ This method should be called before each signing operation.
+ """
+ self.sequence_value += random.randint(0, 1)
+ self.window_props_length += random.randint(1, 10)
+
+ def get_current_state(self, uri: str) -> SignState:
</code_context>
<issue_to_address>
**suggestion (bug_risk):** `window_props_length` grows unbounded over a long-lived session, which may drift far from realistic ranges.
Since `window_props_length` starts in `[1000, 2000]` and is incremented by `random.randint(1, 10)` on every `update_state` call, it will grow without bound in long-lived or high-traffic sessions. If this field is meant to mimic real client window metrics, consider enforcing a maximum (e.g., using a configurable `MAX_WINDOW_PROPS_LENGTH`) or periodically re-randomizing it into a realistic range so the value doesn’t drift into clearly synthetic territory.
Suggested implementation:
```python
"""
Manages the state for a simulated user session to generate more realistic signatures.
This class maintains counters that should persist and evolve across multiple requests
within the same logical session.
"""
WINDOW_PROPS_MIN_LENGTH: int = 1000
WINDOW_PROPS_MAX_LENGTH: int = 2000
def __init__(self):
self.page_load_timestamp: int = int(time.time() * 1000)
self.sequence_value: int = random.randint(15, 17)
self.window_props_length: int = random.randint(
self.WINDOW_PROPS_MIN_LENGTH,
self.WINDOW_PROPS_MAX_LENGTH,
)
```
```python
def update_state(self):
"""
Updates the session state to simulate user activity between requests.
This method should be called before each signing operation.
"""
self.sequence_value += random.randint(0, 1)
self.window_props_length += random.randint(1, 10)
# Prevent window_props_length from drifting unbounded by re-randomizing
# into a realistic range once it exceeds the configured maximum.
if self.window_props_length > self.WINDOW_PROPS_MAX_LENGTH:
self.window_props_length = random.randint(
self.WINDOW_PROPS_MIN_LENGTH,
self.WINDOW_PROPS_MAX_LENGTH,
)
```
</issue_to_address>
### Comment 2
<location> `tests/test_url_utils.py:154-163` </location>
<code_context>
params="invalid", # type: ignore
)
+
+ def test_client_sign_with_session(self):
+ client = Xhshow()
+ session = SessionManager()
+ for _ in range(10):
+ signature = client.sign_xs_get(
+ uri="/api/sns/web/v1/user_posted",
+ a1_value="test_a1_value",
+ params={"num": "30"},
+ session=session,
+ )
+ session.update_state()
+
+ assert signature.startswith("XYS_")
</code_context>
<issue_to_address>
**suggestion (testing):** Strengthen `test_client_sign_with_session` to better exercise the session-aware behavior.
This test advances the session state twice per iteration (`sign_xs_get` → `get_current_state` → `update_state`, plus the explicit `session.update_state()`), which obscures the intent if you only want to verify “signing works with a session.” It also only checks the prefix of the final signature, so it doesn’t actually validate session-aware behavior.
Consider either:
- Removing the loop and extra `session.update_state()` for a simple smoke test, or
- Making it a stronger integration check by capturing multiple signatures and asserting they change across calls with the same session and/or mocking `CryptoProcessor` (like in `tests/test_session.py`) to assert that `sign_state` is used when `session` is provided.
</issue_to_address>
### Comment 3
<location> `src/xhshow/client.py:147` </location>
<code_context>
+
+ sign_state = session.get_current_state(uri) if session else None
+
+ payload_array = self.crypto_processor.build_payload_array(
+ d_value, a1_value, xsec_appid, content_string, timestamp, sign_state=sign_state
)
</code_context>
<issue_to_address>
**issue (complexity):** Consider moving the low-level x3 signature construction back into a high-level helper on CryptoProcessor so Xhshow.sign_xs only orchestrates request/session data.
You can keep the new session-aware behavior but still avoid leaking low‑level crypto details into `Xhshow.sign_xs` by re‑introducing a higher‑level abstraction in `CryptoProcessor`.
### 1. Add a high-level x3 builder to `CryptoProcessor`
In `core/crypto.py` (or wherever `CryptoProcessor` lives), add a method that encapsulates the payload array + xor + base64 pipeline and accepts the optional session state:
```python
class CryptoProcessor:
# existing code...
def build_x3_signature(
self,
d_value: str,
a1_value: str,
xsec_appid: str,
content_string: str,
timestamp: float | None = None,
sign_state: SignState | None = None,
) -> str:
payload_array = self.build_payload_array(
d_value,
a1_value,
xsec_appid,
content_string,
timestamp,
sign_state=sign_state,
)
xor_result = self.bit_ops.xor_transform_array(payload_array)
return self.b64encoder.encode_x3(xor_result[:124])
```
This keeps all the new session-aware logic intact, but hides the low‑level details from `client.py`.
### 2. Simplify `Xhshow.sign_xs` back to high-level orchestration
Then, in `Xhshow.sign_xs`, delegate to the new helper instead of orchestrating `build_payload_array`/`xor`/`encode_x3` directly:
```python
@validate_signature_params
def sign_xs(
self,
method: Literal["GET", "POST"],
uri: str,
a1_value: str,
xsec_appid: str = "xhs-pc-web",
payload: dict[str, Any] | None = None,
timestamp: float | None = None,
session: SessionManager | None = None,
) -> str:
uri = extract_uri(uri)
content_string = self._build_content_string(method, uri, payload)
d_value = self._generate_d_value(content_string)
sign_state = session.get_current_state(uri) if session else None
x3_signature = self.crypto_processor.build_x3_signature(
d_value,
a1_value,
xsec_appid,
content_string,
timestamp,
sign_state=sign_state,
)
signature_data = self.crypto_processor.config.SIGNATURE_DATA_TEMPLATE.copy()
signature_data["x3"] = self.crypto_processor.config.X3_PREFIX + x3_signature
return self.crypto_processor.config.XYS_PREFIX + self.crypto_processor.b64encoder.encode(
json.dumps(signature_data, separators=(",", ":"), ensure_ascii=False)
)
```
This restores the previous layering:
- `Xhshow` handles request-level concerns (URI, method, payload, session lookup).
- `CryptoProcessor` owns the crypto pipeline, including the new session-aware behavior.
Functionality (including session plumbing in `sign_xs_get`, `sign_xs_post`, and `sign_headers*`) remains unchanged, but the complexity is localized back into the crypto layer.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
功能建议：风控参数模拟 #86
由 Sourcery 提供的总结
引入可感知会话的请求签名机制，在保留现有无状态行为的同时，生成更加真实、有状态的签名。
新功能：
SessionManager和SignState抽象，用于维护每个会话的签名状态，例如时间戳和计数器。增强：
CryptoProcessor载荷构造逻辑，以接受可选的签名状态，从而生成基于会话的确定性载荷字段。测试：
SessionManager时客户端签名工作正常。Original summary in English
Summary by Sourcery
Introduce session-aware request signing to generate more realistic, stateful signatures while preserving existing stateless behavior.
New Features:
Enhancements:
Tests:
Original summary in English
由 Sourcery 提供的总结
引入可感知会话的请求签名机制，在保留现有无状态行为的同时，生成更加真实、有状态的签名。
新功能：
SessionManager和SignState抽象，用于维护每个会话的签名状态，例如时间戳和计数器。增强：
CryptoProcessor载荷构造逻辑，以接受可选的签名状态，从而生成基于会话的确定性载荷字段。测试：
SessionManager时客户端签名工作正常。Original summary in English
Summary by Sourcery
Introduce session-aware request signing to generate more realistic, stateful signatures while preserving existing stateless behavior.
New Features:
Enhancements:
Tests: