fix: 优化清理失效账号执行响应的 success 标志,对 test_account/delete_account 引入外部 API 日志审计,并对账号 ID 进行安全转义

This commit is contained in:
liumangmang
2026-07-02 00:25:24 +08:00
parent 536c0aa3d4
commit 3b23f21ef3
3 changed files with 134 additions and 63 deletions
+16 -2
View File
@@ -1085,9 +1085,23 @@ def execute_cleanup_invalid_accounts(wid: int, db: Session = Depends(get_db), _=
db.commit() db.commit()
deleted_count = sum(1 for item in items if item.action_status in ("deleted", "already_deleted")) deleted_count = sum(1 for item in items if item.action_status in ("deleted", "already_deleted"))
failed_count = sum(1 for item in items if item.action_status == "failed")
skipped_count = sum(1 for item in items if item.action_status == "skipped")
success = (failed_count == 0)
msg_parts = []
if deleted_count:
msg_parts.append(f"成功清理 {deleted_count}")
if failed_count:
msg_parts.append(f"失败 {failed_count}")
if skipped_count:
msg_parts.append(f"跳过 {skipped_count}")
message = "清理执行完毕:" + "".join(msg_parts) if msg_parts else "没有需要清理的失效账号"
return CleanupInvalidAccountsExecuteResponse( return CleanupInvalidAccountsExecuteResponse(
success=True, success=success,
message=f"清理执行完毕,成功处理了 {deleted_count} 个失效账号", message=message,
items=items items=items
) )
+84 -53
View File
@@ -363,75 +363,106 @@ class Sub2ApiWebsiteClient:
若 timeout / 无法建立连接,返回 {"status": "timeout"/"error", "cleanup_allowed": False} 若 timeout / 无法建立连接,返回 {"status": "timeout"/"error", "cleanup_allowed": False}
""" """
import json import json
url = self._url(f"{endpoint}/{account_id}/test") quoted_id = quote(account_id, safe="")
path = f"{endpoint}/{quoted_id}/test"
url = self._url(path)
headers = self._headers() headers = self._headers()
# Accept text/event-stream since it's an SSE stream # Accept text/event-stream since it's an SSE stream
headers["Accept"] = "text/event-stream" headers["Accept"] = "text/event-stream"
started = time.monotonic()
status_code: int | None = None
error_type: str | None = None
error_msg: str | None = None
try: try:
with self._client.stream("POST", url, headers=headers) as response: try:
if response.status_code == 404: with self._client.stream("POST", url, headers=headers) as response:
return {"status": "404", "cleanup_allowed": True, "message": "账号在远端不存在 (404)"} status_code = response.status_code
if response.status_code >= 400: if response.status_code == 404:
return {"status": "http_error", "cleanup_allowed": False, "message": f"测试接口返回 HTTP {response.status_code}"} return {"status": "404", "cleanup_allowed": True, "message": "账号在远端不存在 (404)"}
if response.status_code >= 400:
response.raise_for_status()
# Read SSE stream line by line # Read SSE stream line by line
current_event = None current_event = None
success_value = None success_value = None
has_error_event = False has_error_event = False
for line in response.iter_lines(): for line in response.iter_lines():
if not line: if not line:
continue continue
if line.startswith("event:"): if line.startswith("event:"):
current_event = line[6:].strip() current_event = line[6:].strip()
elif line.startswith("data:"): elif line.startswith("data:"):
data_str = line[5:].strip() data_str = line[5:].strip()
if current_event == "error": if current_event == "error":
has_error_event = True has_error_event = True
if current_event == "test_complete" or not current_event: if current_event == "test_complete" or not current_event:
try: try:
parsed = json.loads(data_str) parsed = json.loads(data_str)
if isinstance(parsed, dict): if isinstance(parsed, dict):
if "success" in parsed: if "success" in parsed:
success_value = parsed["success"] success_value = parsed["success"]
elif parsed.get("event") == "test_complete" and "success" in parsed: elif parsed.get("event") == "test_complete" and "success" in parsed:
success_value = parsed["success"] success_value = parsed["success"]
except Exception: except Exception:
pass pass
if success_value is True: if success_value is True:
return {"status": "success", "cleanup_allowed": False, "message": "测试成功,账号有效"} return {"status": "success", "cleanup_allowed": False, "message": "测试成功,账号有效"}
elif success_value is False: elif success_value is False:
return {"status": "failed", "cleanup_allowed": True, "message": "测试失败,账号失效"} return {"status": "failed", "cleanup_allowed": True, "message": "测试失败,账号失效"}
elif has_error_event: elif has_error_event:
return {"status": "error", "cleanup_allowed": False, "message": "测试接口返回了错误事件"} error_type = "WebsiteError"
else: error_msg = "测试接口返回了错误事件"
return {"status": "invalid_sse", "cleanup_allowed": False, "message": "未收到有效的测试完成事件"} return {"status": "error", "cleanup_allowed": False, "message": "测试接口返回了错误事件"}
else:
except httpx.TimeoutException as exc: error_type = "WebsiteError"
return {"status": "timeout", "cleanup_allowed": False, "message": f"连接超时: {exc}"} error_msg = "未收到有效的测试完成事件"
except Exception as exc: return {"status": "invalid_sse", "cleanup_allowed": False, "message": "未收到有效的测试完成事件"}
return {"status": "error", "cleanup_allowed": False, "message": f"请求异常: {exc}"} except httpx.TimeoutException as exc:
error_type, error_msg = type(exc).__name__, str(exc)[:500]
return {"status": "timeout", "cleanup_allowed": False, "message": f"连接超时: {exc}"}
except httpx.HTTPStatusError as exc:
error_type, error_msg = type(exc).__name__, str(exc)[:500]
status_code = exc.response.status_code
return {"status": "http_error", "cleanup_allowed": False, "message": f"测试接口返回 HTTP {status_code}"}
except Exception as exc:
error_type, error_msg = type(exc).__name__, str(exc)[:500]
return {"status": "error", "cleanup_allowed": False, "message": f"请求异常: {exc}"}
finally:
elapsed = int((time.monotonic() - started) * 1000)
parsed = urlparse(url)
log_external_api_call(
direction="website",
target_type="website",
method="POST",
path=parsed.path or path,
url_host=parsed.hostname or "",
duration_ms=elapsed,
status_code=status_code,
success=error_type is None,
target_id=self.target_id,
target_name=self.target_name,
error_type=error_type,
error_message=error_msg,
)
def delete_account(self, account_id: str, endpoint: str = "/accounts") -> dict[str, Any]: def delete_account(self, account_id: str, endpoint: str = "/accounts") -> dict[str, Any]:
"""删除目标账号。 """删除目标账号。
若 404 视为 already_deleted。 若 404 视为 already_deleted。
""" """
url = self._url(f"{endpoint}/{account_id}") quoted_id = quote(account_id, safe="")
headers = self._headers() path = f"{endpoint}/{quoted_id}"
try: try:
resp = self._client.request("DELETE", url, headers=headers) self._request("DELETE", path)
if resp.status_code == 404:
return {"status": "already_deleted", "message": "账号不存在或已在远端被删除"}
resp.raise_for_status()
return {"status": "deleted", "message": "账号已删除"} return {"status": "deleted", "message": "账号已删除"}
except httpx.HTTPStatusError as exc: except WebsiteError as exc:
if exc.response.status_code == 404: cause = exc.__cause__
if isinstance(cause, httpx.HTTPStatusError) and cause.response.status_code == 404:
return {"status": "already_deleted", "message": "账号不存在或已在远端被删除"} return {"status": "already_deleted", "message": "账号不存在或已在远端被删除"}
raise WebsiteError(_friendly_http_error(exc)) from exc raise
except Exception as exc:
raise WebsiteError(f"删除账号失败: {exc}") from exc
@staticmethod @staticmethod
def extract_id(value: Any) -> str: def extract_id(value: Any) -> str:
+34 -8
View File
@@ -56,7 +56,7 @@ def mock_stream(status_code, lines, raise_exc=None):
def test_website_client_test_account_sse_scenarios(monkeypatch): def test_website_client_test_account_sse_scenarios(monkeypatch):
"""测试 Sub2ApiWebsiteClient.test_account 在各种 SSE 场景下的解析处理。""" """测试 Sub2ApiWebsiteClient.test_account 在各种 SSE 场景下的解析处理、ID 转义和 API 日志记录"""
c = Sub2ApiWebsiteClient( c = Sub2ApiWebsiteClient(
base_url="http://w1", base_url="http://w1",
api_prefix="api/v1", api_prefix="api/v1",
@@ -64,41 +64,65 @@ def test_website_client_test_account_sse_scenarios(monkeypatch):
auth_config={"token": "t1"}, auth_config={"token": "t1"},
) )
# 1. success=true log_calls = []
def mock_log(**kwargs):
log_calls.append(kwargs)
monkeypatch.setattr("app.services.website_client.log_external_api_call", mock_log)
# 1. success=true,包含带有斜杠的 ID 校验以测试 URL quote 转义
monkeypatch.setattr(c._client, "stream", mock_stream(200, ["event: test_complete", "data: {\"success\": true}"])) monkeypatch.setattr(c._client, "stream", mock_stream(200, ["event: test_complete", "data: {\"success\": true}"]))
res = c.test_account("acc1") res = c.test_account("acc/1")
assert res["status"] == "success" assert res["status"] == "success"
assert res["cleanup_allowed"] is False assert res["cleanup_allowed"] is False
# 验证日志记录和 URL 转义:acc/1 应被转义为 acc%2F1
assert len(log_calls) == 1
assert log_calls[-1]["method"] == "POST"
assert log_calls[-1]["path"] == "/api/v1/accounts/acc%2F1/test"
assert log_calls[-1]["success"] is True
# 2. success=false # 2. success=false
monkeypatch.setattr(c._client, "stream", mock_stream(200, ["event: test_complete", "data: {\"success\": false}"])) monkeypatch.setattr(c._client, "stream", mock_stream(200, ["event: test_complete", "data: {\"success\": false}"]))
res = c.test_account("acc1") res = c.test_account("acc/1")
assert res["status"] == "failed" assert res["status"] == "failed"
assert res["cleanup_allowed"] is True assert res["cleanup_allowed"] is True
assert len(log_calls) == 2
assert log_calls[-1]["success"] is True
# 3. HTTP 404 # 3. HTTP 404
monkeypatch.setattr(c._client, "stream", mock_stream(404, [])) monkeypatch.setattr(c._client, "stream", mock_stream(404, []))
res = c.test_account("acc1") res = c.test_account("acc/1")
assert res["status"] == "404" assert res["status"] == "404"
assert res["cleanup_allowed"] is True assert res["cleanup_allowed"] is True
assert len(log_calls) == 3
assert log_calls[-1]["status_code"] == 404
assert log_calls[-1]["success"] is True
# 4. error event # 4. error event
monkeypatch.setattr(c._client, "stream", mock_stream(200, ["event: error", "data: \"internal error\""])) monkeypatch.setattr(c._client, "stream", mock_stream(200, ["event: error", "data: \"internal error\""]))
res = c.test_account("acc1") res = c.test_account("acc/1")
assert res["status"] == "error" assert res["status"] == "error"
assert res["cleanup_allowed"] is False assert res["cleanup_allowed"] is False
assert len(log_calls) == 4
assert log_calls[-1]["success"] is False
assert log_calls[-1]["error_type"] == "WebsiteError"
# 5. 超时 # 5. 超时
monkeypatch.setattr(c._client, "stream", mock_stream(200, [], raise_exc=httpx.TimeoutException("Connection timed out"))) monkeypatch.setattr(c._client, "stream", mock_stream(200, [], raise_exc=httpx.TimeoutException("Connection timed out")))
res = c.test_account("acc1") res = c.test_account("acc/1")
assert res["status"] == "timeout" assert res["status"] == "timeout"
assert res["cleanup_allowed"] is False assert res["cleanup_allowed"] is False
assert len(log_calls) == 5
assert log_calls[-1]["success"] is False
assert log_calls[-1]["error_type"] == "TimeoutException"
# 6. 非法 SSE (没有 test_complete) # 6. 非法 SSE (没有 test_complete)
monkeypatch.setattr(c._client, "stream", mock_stream(200, ["data: hello world"])) monkeypatch.setattr(c._client, "stream", mock_stream(200, ["data: hello world"]))
res = c.test_account("acc1") res = c.test_account("acc/1")
assert res["status"] == "invalid_sse" assert res["status"] == "invalid_sse"
assert res["cleanup_allowed"] is False assert res["cleanup_allowed"] is False
assert len(log_calls) == 6
assert log_calls[-1]["success"] is False
assert log_calls[-1]["error_type"] == "WebsiteError"
def test_cleanup_invalid_preview_and_execute(db_session, monkeypatch): def test_cleanup_invalid_preview_and_execute(db_session, monkeypatch):
@@ -229,6 +253,8 @@ def test_cleanup_invalid_preview_and_execute(db_session, monkeypatch):
test_calls.clear() test_calls.clear()
execute_res = execute_cleanup_invalid_accounts(wid=w.id, db=db_session) execute_res = execute_cleanup_invalid_accounts(wid=w.id, db=db_session)
assert execute_res.success is True assert execute_res.success is True
assert "成功清理 2 个" in execute_res.message
assert "跳过 1 个" in execute_res.message
# 验证是否重新跑了测试 # 验证是否重新跑了测试
assert "acc-k1" in test_calls assert "acc-k1" in test_calls