From 70e11dd1e7628abc260099726dcd4f94e02f4805 Mon Sep 17 00:00:00 2001 From: Codex Date: Thu, 7 May 2026 18:12:43 +0800 Subject: [PATCH] fix: sign Feishu webhook messages --- daily_report/robot_service.py | 20 ++++++++++++++++++-- daily_report/scheduled.py | 4 ++-- daily_report/web.py | 24 ++++++++++++++++++++++-- tests/test_robot_service.py | 8 +++++++- tests/test_scheduled.py | 4 +++- 5 files changed, 52 insertions(+), 8 deletions(-) diff --git a/daily_report/robot_service.py b/daily_report/robot_service.py index ad8677b..9485877 100644 --- a/daily_report/robot_service.py +++ b/daily_report/robot_service.py @@ -1,6 +1,10 @@ from __future__ import annotations import json +import base64 +import hashlib +import hmac +import time import urllib.error import urllib.request from typing import Any @@ -54,13 +58,25 @@ def create_summary_payload(manager_url: str, summary: dict[str, Any]) -> dict[st } -def send_webhook(webhook_url: str, payload: dict[str, Any]) -> dict[str, Any]: +def create_webhook_sign(timestamp: str, secret: str) -> str: + string_to_sign = f"{timestamp}\n{secret}".encode("utf-8") + sign = hmac.new(string_to_sign, digestmod=hashlib.sha256).digest() + return base64.b64encode(sign).decode("utf-8") + + +def send_webhook(webhook_url: str, payload: dict[str, Any], secret: str = "") -> dict[str, Any]: if not webhook_url: raise ValueError("FEISHU_WEBHOOK_URL is required") + request_payload = dict(payload) + if secret: + timestamp = str(int(time.time())) + request_payload["timestamp"] = timestamp + request_payload["sign"] = create_webhook_sign(timestamp, secret) + request = urllib.request.Request( webhook_url, - data=json.dumps(payload).encode("utf-8"), + data=json.dumps(request_payload).encode("utf-8"), headers={"content-type": "application/json"}, method="POST", ) diff --git a/daily_report/scheduled.py b/daily_report/scheduled.py index 783dd33..b7f1f93 100644 --- a/daily_report/scheduled.py +++ b/daily_report/scheduled.py @@ -15,7 +15,7 @@ def send_reminder() -> dict: if not is_workday(date.today(), config.workday_calendar_path): return {"skipped": True, "reason": "not a workday"} payload = create_reminder_payload(f"{config.base_url}/submit") - return send_webhook(config.feishu_webhook_url, payload) + return send_webhook(config.feishu_webhook_url, payload, config.feishu_webhook_secret) def send_summary(report_date: str | None = None) -> dict: @@ -30,7 +30,7 @@ def send_summary(report_date: str | None = None) -> dict: selected_date_text = selected_date.isoformat() summary = service.list_reports_for_date(selected_date_text) payload = create_summary_payload(f"{config.base_url}/manager?date={selected_date_text}", summary) - return send_webhook(config.feishu_webhook_url, payload) + return send_webhook(config.feishu_webhook_url, payload, config.feishu_webhook_secret) finally: database.close() diff --git a/daily_report/web.py b/daily_report/web.py index 55a910c..2743f6c 100644 --- a/daily_report/web.py +++ b/daily_report/web.py @@ -294,14 +294,34 @@ class DailyReportHandler(BaseHTTPRequestHandler): return if parsed.path == "/api/robot/send-reminder": payload = create_reminder_payload(f"{self.config.base_url}/submit") - self._json(200, {"ok": True, "result": send_webhook(self.config.feishu_webhook_url, payload)}) + self._json( + 200, + { + "ok": True, + "result": send_webhook( + self.config.feishu_webhook_url, + payload, + self.config.feishu_webhook_secret, + ), + }, + ) return if parsed.path == "/api/robot/send-summary": body = self._read_json() report_date = body.get("date") or today_string() summary = self.report_service.list_reports_for_date(report_date) payload = create_summary_payload(f"{self.config.base_url}/manager?date={report_date}", summary) - self._json(200, {"ok": True, "result": send_webhook(self.config.feishu_webhook_url, payload)}) + self._json( + 200, + { + "ok": True, + "result": send_webhook( + self.config.feishu_webhook_url, + payload, + self.config.feishu_webhook_secret, + ), + }, + ) return self._json(404, {"error": "not found"}) except (ValueError, json.JSONDecodeError) as error: diff --git a/tests/test_robot_service.py b/tests/test_robot_service.py index 379d02b..9c47c48 100644 --- a/tests/test_robot_service.py +++ b/tests/test_robot_service.py @@ -3,7 +3,7 @@ from __future__ import annotations import json import unittest -from daily_report.robot_service import create_reminder_payload, create_summary_payload +from daily_report.robot_service import create_reminder_payload, create_summary_payload, create_webhook_sign class RobotServiceTest(unittest.TestCase): @@ -33,6 +33,12 @@ class RobotServiceTest(unittest.TestCase): self.assertIn("Chen", text) self.assertIn("http://localhost:8787/manager", text) + def test_creates_webhook_signature(self) -> None: + signature = create_webhook_sign("1599360473", "test-secret") + + self.assertIsInstance(signature, str) + self.assertGreater(len(signature), 20) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_scheduled.py b/tests/test_scheduled.py index a733139..9853bab 100644 --- a/tests/test_scheduled.py +++ b/tests/test_scheduled.py @@ -36,9 +36,10 @@ class ScheduledTest(unittest.TestCase): captured = {} - def fake_send(webhook_url: str, payload: dict) -> dict: + def fake_send(webhook_url: str, payload: dict, secret: str = "") -> dict: captured["webhook_url"] = webhook_url captured["payload"] = payload + captured["secret"] = secret return {"ok": True} with patch("daily_report.config.__file__", str(fake_file)), patch.dict("os.environ", env, clear=True), patch( @@ -48,6 +49,7 @@ class ScheduledTest(unittest.TestCase): self.assertEqual(result, {"ok": True}) self.assertEqual(captured["webhook_url"], "https://example.test/webhook") + self.assertEqual(captured["secret"], "") text = json.dumps(captured["payload"], ensure_ascii=False) self.assertIn("2026-05-07 日报提交汇总", text) self.assertIn("0/2", text)