diff --git a/CLAUDE.md b/CLAUDE.md index d694e6e02..9cedbd30f 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -414,6 +414,7 @@ clone) או לקרוא ממנו ישירות. אין צורך בהרשאות מ | PyGithub / קריאות SDK חיצוני | `BY-STACK/external-sdk.md` | | קבצי `docs/**/*.rst` | `bugbot-rules/line-number-coupling.md` | | טסטים עם סטאבים ידניים | `TESTING-PATTERNS.md` + `bugbot-rules/widened-exception-scope.md` | +| הרכבת URL/מחרוזת שמכילה סוד, הודעות חריגה, ניקוי לוגים/Sentry | `CRITICAL-PATTERNS.md` K13 + `bugbot-rules/secret-in-derived-text.md` | ### תמיד, בלי קשר לטבלה diff --git a/docs/security.rst b/docs/security.rst index 3d7782d19..566b3ad2f 100644 --- a/docs/security.rst +++ b/docs/security.rst @@ -6,6 +6,40 @@ Security Guide אל תרשום סודות/PII בלוגים, השתמש ב‑ENV בלבד. +ניקוי טוקן הבוט מלוגים, חריגות ו‑Sentry +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +כתובות ה‑API של טלגרם נבנות כ‑``https://api.telegram.org/bot/method`` — +כלומר **כל טקסט שנגזר מכתובת כזו נושא את הטוקן במלואו**: הודעת חריגה, שורת לוג, +traceback או אירוע Sentry. בעקבות אירוע אבטחה (אוגוסט 2026) קיימת נקודת ניקוי +מרכזית ב‑``telegram_api``: + +- ``redact_bot_token(text)`` — מנקה טוקן ממחרוזת בודדת. +- ``redact_bot_token_deep(obj)`` — מנקה מבנה מקונן שלם (dict כולל מפתחות, list, + tuple/namedtuple, set, bytes, ואובייקטים זרים שהייצוג שלהם נושא טוקן). עמיד + למבנים מעגליים, וכשל ניקוי מחזיר placeholder — לעולם לא את הערך הגולמי. + +השכבות שנשענות על הנקודה הזו: + +- ``TelegramAPIError`` מנקה את ``url``/``description``/``payload`` **בהשמה**, כך + שכל מקומות הקריאה מכוסים בלי לגעת בהם — כולל כאלה שיתווספו בעתיד. +- ``SensitiveDataFilter`` (ב‑``utils``) מנקה את ההודעה **ואת ה‑traceback** דרך + רשימת דפוסים אחת (טלגרם, GitHub, Bearer). ``exc_info`` נשמר — ה‑Formatter + משתמש בקאש ``exc_text`` המנוקה, ו‑Sentry מנקה את החריגה המובנית בעצמו. +- ``before_send`` בבוט (``observability``) ובוובאפ הוא **fail-closed**: אם + הניקוי נכשל, האירוע נזרק ונרשמת אזהרת ``sentry_redaction_failed`` — עדיף + לאבד אירוע מאשר להדליף טוקן. + +כללים לקוד חדש: + +- אין לשרשר URL של Bot API לתוך הודעת שגיאה או לוג כמות שהוא — להעביר דרך + ``redact_bot_token`` קודם. +- סוג סוד חדש דורש עדכון בשתי נקודות: ``SensitiveDataFilter._PATTERNS`` (מכסה + הודעות לוג ו‑traceback) **וגם** נקודת הניקוי ב‑``telegram_api`` (מכסה חריגות + ו‑before_send של Sentry). עדכון של אחת בלבד משאיר את המסלול השני חשוף. +- הטסטים ב‑``tests/test_telegram_token_redaction.py`` נועלים את ההתנהגות; שינוי + במנגנון חייב לעבור אותם. + הרצת קוד (Code Execution Playground) ------------------------------------ ה‑WebApp כולל Playground בכתובת ``/tools/code`` (פתוח לכל משתמש מחובר), כאשר **הרצת קוד בפועל** זמינה רק ל‑Premium/Admin. diff --git a/observability.py b/observability.py index fa23d66ea..73578e384 100644 --- a/observability.py +++ b/observability.py @@ -855,6 +855,20 @@ def _before_send(event, hint): # type: ignore[no-redef] event["tags"] = tags except Exception: pass + # ניקוי טוקנים מכל האירוע — לא רק מ-extra לפי שם שדה. כתובות ה-API של + # טלגרם מכילות את הטוקן, והן מגיעות לתוך גוף החריגה ולתוך breadcrumbs, + # מקומות שסינון לפי שם מפתח לא מגיע אליהם. + try: + from telegram_api import redact_bot_token_deep # type: ignore + + event = redact_bot_token_deep(event) + except Exception: + # fail-closed: עדיף לאבד אירוע אחד מאשר לשלוח אירוע שלא נוקה + try: + LOGGER.warning("sentry event dropped: token redaction failed", extra={"event": "sentry_redaction_failed"}) + except Exception: + pass + return None return event # Resolve environment consistently with fallback to config if ENV/ENVIRONMENT not set diff --git a/telegram_api.py b/telegram_api.py index a66490349..ee8bd911c 100644 --- a/telegram_api.py +++ b/telegram_api.py @@ -1,7 +1,144 @@ from __future__ import annotations +import re from typing import Any, Dict, Optional +# מבנה טוקן של בוט טלגרם: מזהה מספרי, נקודתיים, ואז סוד באורך ~35 תווים. +# כתובות ה-API נבנות כ-https://api.telegram.org/bot/method — ולכן כל טקסט +# שנגזר מכתובת כזו (הודעת שגיאה, לוג, אירוע Sentry) עלול לשאת את הטוקן במלואו. +# דרישת 30+ תווים בסוד מצמצמת פגיעה בטקסטים לגיטימיים (hash/מזהה עם נקודתיים), +# ועדיין תופסת כל טוקן אמיתי. +_BOT_TOKEN_RE = re.compile(r"\d{5,16}:[A-Za-z0-9_-]{30,}") + +TOKEN_PLACEHOLDER = "" + +# מוחזר כשאי אפשר לנקות ערך (str() נכשל) — עדיף לאבד את הערך מאשר להדליף טוקן +UNREDACTABLE_PLACEHOLDER = "" + + +def redact_bot_token(value: Any) -> Any: + """מחליף כל טוקן בוט שמופיע בטקסט בסימון ````. + + מחזיר ``None`` כפי שהוא, וכל ערך אחר מומר למחרוזת מנוקה. אם ההמרה למחרוזת + נכשלת מוחזר ```` — כישלון ניקוי לעולם לא מחזיר את הערך הגולמי. + זו נקודת הניקוי היחידה בקוד — ``TelegramAPIError``, מסנני ה-Sentry ומסנן + הלוגים נשענים עליה. + """ + if value is None: + return None + try: + text = value if isinstance(value, str) else str(value) + return _BOT_TOKEN_RE.sub(TOKEN_PLACEHOLDER, text) + except Exception: + return UNREDACTABLE_PLACEHOLDER + + +def _redact_bytes(obj: Any) -> Any: + """מנקה טוקן מ-bytes/bytearray תוך שמירה על הטיפוס המקורי.""" + try: + cleaned_text = _BOT_TOKEN_RE.sub(TOKEN_PLACEHOLDER, obj.decode("utf-8", errors="replace")) + encoded = cleaned_text.encode("utf-8") + return bytearray(encoded) if isinstance(obj, bytearray) else encoded + except Exception: + return UNREDACTABLE_PLACEHOLDER + + +def _dedupe_key(ck: Any, cleaned_dict: Dict[Any, Any]) -> Any: + """ממספר מפתח שמתנגש עם מפתח קיים אחרי ניקוי (‎#2 למחרוזת, ‎(key, 2)‎ לאחרים). + + שני מפתחות שונים יכולים להתנקות לאותו ערך (שתי כתובות עם טוקנים שונים); + דריסה שקטה מאבדת שדה אבחוני — במקום זה המאוחר מקבל סיומת מספור. + """ + if ck not in cleaned_dict: + return ck + n = 2 + candidate = f"{ck}#{n}" if isinstance(ck, str) else (ck, n) + while candidate in cleaned_dict: + n += 1 + candidate = f"{ck}#{n}" if isinstance(ck, str) else (ck, n) + return candidate + + +def _redact_dict(obj: Dict[Any, Any], _memo: Dict[int, Any], _depth: int) -> Dict[Any, Any]: + cleaned_dict: Dict[Any, Any] = {} + # רישום לפני המילוי — כך הפניה מעגלית חוזרת לעותק המנוקה ולא למקור + _memo[id(obj)] = cleaned_dict + for k, v in obj.items(): + # גם מפתח יכול לשאת טוקן — כמחרוזת או בתוך tuple/frozenset (שנשארים hashable) + ck = _dedupe_key(redact_bot_token_deep(k, _memo, _depth + 1), cleaned_dict) + cleaned_dict[ck] = redact_bot_token_deep(v, _memo, _depth + 1) + return cleaned_dict + + +def _redact_list(obj: list, _memo: Dict[int, Any], _depth: int) -> list: + cleaned_list: list = [] + _memo[id(obj)] = cleaned_list + for v in obj: + cleaned_list.append(redact_bot_token_deep(v, _memo, _depth + 1)) + return cleaned_list + + +def _redact_tuple(obj: tuple, _memo: Dict[int, Any], _depth: int) -> tuple: + cleaned_items = [redact_bot_token_deep(v, _memo, _depth + 1) for v in obj] + try: + if hasattr(obj, "_fields"): # namedtuple — בנייה מאיברים בודדים + return type(obj)(*cleaned_items) + return type(obj)(cleaned_items) + except Exception: + # תת-מחלקה עם בנאי לא סטנדרטי — tuple רגיל עדיף על אובייקט לא מנוקה + return tuple(cleaned_items) + + +def redact_bot_token_deep(obj: Any, _memo: Optional[Dict[int, Any]] = None, _depth: int = 0) -> Any: + """מנקה טוקנים מכל המחרוזות בתוך מבנה נתונים מקונן. + + נועד למסנני Sentry: אירוע שגיאה פורש את הטוקן על פני כמה שדות (גוף החריגה, + הודעת הלוג, breadcrumbs), ורשימת שדות קבועה תמיד תפספס אחד. במקום זה עוברים + על כל המבנה — dict (כולל מפתחות), list, tuple (כולל namedtuple), set ו-frozenset. + + מבנים מעגליים מטופלים ב-memo לפי ‎id()‎ כך שכל צומת מנוקה בדיוק פעם אחת; + מגבלת העומק היא רשת ביטחון בלבד, וחצייה שלה מחזירה placeholder — לעולם לא + את הערך המקורי. + """ + if _memo is None: + _memo = {} + if _depth > 100: + # עומק כזה לא קיים באירועי Sentry אמיתיים; מחזירים placeholder ולא את המקור + return UNREDACTABLE_PLACEHOLDER + if isinstance(obj, str): + return redact_bot_token(obj) + if isinstance(obj, (bytes, bytearray)): + return _redact_bytes(obj) + if isinstance(obj, (dict, list)): + existing = _memo.get(id(obj)) + if existing is not None: + return existing + if isinstance(obj, dict): + return _redact_dict(obj, _memo, _depth) + return _redact_list(obj, _memo, _depth) + if isinstance(obj, tuple): + return _redact_tuple(obj, _memo, _depth) + if isinstance(obj, (set, frozenset)): + cleaned_set = {redact_bot_token_deep(v, _memo, _depth + 1) for v in obj} + return frozenset(cleaned_set) if isinstance(obj, frozenset) else cleaned_set + # סקלרים חסרי טקסט — אין מה לנקות בהם + if obj is None or isinstance(obj, (int, float, bool)): + return obj + # אובייקט זר (למשל מופע חריגה בתוך hint/extra): Sentry ימיר אותו למחרוזת + # אחרי שהניקוי כבר עבר — ב-repr() עבור אובייקטים שאינם JSON — ולכן בודקים + # את שני הייצוגים. אם אחד מהם נושא טוקן, מחזירים את הייצוג המנוקה במקום + # האובייקט. אובייקט נקי בשניהם נשמר כמות שהוא. + try: + text = str(obj) + rep = repr(obj) + except Exception: + return UNREDACTABLE_PLACEHOLDER + if _BOT_TOKEN_RE.search(rep): + return _BOT_TOKEN_RE.sub(TOKEN_PLACEHOLDER, rep) + if _BOT_TOKEN_RE.search(text): + return _BOT_TOKEN_RE.sub(TOKEN_PLACEHOLDER, text) + return obj + def _truncate(text: Any, limit: int = 800) -> str: try: @@ -27,10 +164,13 @@ def __init__( payload: Any = None, ) -> None: self.error_code = error_code - self.description = str(description or "").strip() - self.url = url + # ניקוי הטוקן כבר כאן, לפני ההשמה: כך גם ``self.url``/``self.description`` + # וגם טקסט החריגה נקיים, ולא משנה מי יקרא אותם או ירשום אותם ללוג. + self.description = redact_bot_token(str(description or "").strip()) + self.url = redact_bot_token(url) self.http_status = http_status - self.payload = payload + # ניקוי עמוק — גם dict/list (למשל payload מלא של תגובת טלגרם) חייבים לצאת נקיים + self.payload = redact_bot_token_deep(payload) if payload is not None else None msg = f"Telegram API error" if error_code is not None: msg += f" error_code={error_code}" @@ -38,8 +178,8 @@ def __init__( msg += f" description={self.description}" if http_status is not None: msg += f" http_status={http_status}" - if url: - msg += f" url={url}" + if self.url: + msg += f" url={self.url}" super().__init__(msg) diff --git a/tests/test_telegram_token_redaction.py b/tests/test_telegram_token_redaction.py new file mode 100644 index 000000000..cd15a8f76 --- /dev/null +++ b/tests/test_telegram_token_redaction.py @@ -0,0 +1,362 @@ +"""בדיקות שהטוקן של הבוט לא דולף לטקסטים שנשמרים או נשלחים החוצה. + +כתובות ה-API של טלגרם נבנות כ-``https://api.telegram.org/bot/method``, +ולכן כל טקסט שנגזר מהן — הודעת חריגה, שורת לוג או אירוע Sentry — עלול לשאת +את הטוקן במלואו. הבדיקות כאן נועלות את נקודות הניקוי. +""" + +import logging + +import pytest + +from telegram_api import ( + TelegramAPIError, + parse_telegram_json_from_response, + redact_bot_token, + redact_bot_token_deep, + require_telegram_ok, +) + +# טוקן בדוי במבנה אמיתי — משמש רק לבדיקה שהוא לא שורד בפלט +FAKE_TOKEN = "7628556044:AAHdqTcvCH1vGWJxfSeofSAs0K5PALDsaw" +API_URL = f"https://api.telegram.org/bot{FAKE_TOKEN}/sendMessage" + + +class _FakeResponse: + """תגובת HTTP מינימלית, כמו זו ש-requests/http_sync מחזירים.""" + + def __init__(self, *, payload=None, text="", status_code=200, url=API_URL): + self._payload = payload + self.text = text + self.status_code = status_code + self.url = url + + def json(self): + if self._payload is None: + raise ValueError("not json") + return self._payload + + +def test_redact_bot_token_replaces_token_in_url(): + assert FAKE_TOKEN not in redact_bot_token(API_URL) + assert "" in redact_bot_token(API_URL) + + +def test_redact_bot_token_keeps_surrounding_text(): + cleaned = redact_bot_token(f"POST {API_URL} failed") + assert cleaned.startswith("POST https://api.telegram.org/bot") + assert cleaned.endswith("/sendMessage failed") + + +def test_redact_bot_token_passes_through_none_and_clean_text(): + assert redact_bot_token(None) is None + assert redact_bot_token("nothing secret here") == "nothing secret here" + + +def test_error_message_has_no_token(): + err = TelegramAPIError( + error_code=403, + description="Forbidden: bot was blocked by the user", + url=API_URL, + http_status=403, + ) + assert FAKE_TOKEN not in str(err) + + +def test_error_attributes_have_no_token(): + """גם מי שקורא ``e.url`` ישירות ורושם אותו ללוג לא אמור לקבל את הטוקן.""" + err = TelegramAPIError(error_code=None, description="boom", url=API_URL) + assert FAKE_TOKEN not in str(err.url) + + +def test_error_description_carrying_token_is_cleaned(): + err = TelegramAPIError(error_code=None, description=f"failed calling {API_URL}", url=None) + assert FAKE_TOKEN not in str(err) + + +def test_require_telegram_ok_raises_without_token(): + payload = {"ok": False, "error_code": 400, "description": "Bad Request: chat not found"} + with pytest.raises(TelegramAPIError) as excinfo: + require_telegram_ok(payload, url=API_URL) + assert FAKE_TOKEN not in str(excinfo.value) + + +def test_parse_invalid_json_raises_without_token(): + resp = _FakeResponse(payload=None, text="502", status_code=502) + with pytest.raises(TelegramAPIError) as excinfo: + parse_telegram_json_from_response(resp, url=API_URL) + assert FAKE_TOKEN not in str(excinfo.value) + + +def test_parse_falls_back_to_response_url_without_token(): + """כש-url לא מועבר במפורש הוא נשלף מהתגובה — וגם אז חייב להיות נקי.""" + resp = _FakeResponse(payload=None, text="oops", status_code=500) + with pytest.raises(TelegramAPIError) as excinfo: + parse_telegram_json_from_response(resp) + assert FAKE_TOKEN not in str(excinfo.value) + + +def test_parse_non_dict_json_raises_without_token(): + resp = _FakeResponse(payload=["not", "a", "dict"], status_code=200) + with pytest.raises(TelegramAPIError) as excinfo: + parse_telegram_json_from_response(resp, url=API_URL) + assert FAKE_TOKEN not in str(excinfo.value) + + +def test_redact_deep_cleans_nested_sentry_shaped_event(): + event = { + "exception": {"values": [{"type": "TelegramAPIError", "value": f"error url={API_URL}"}]}, + "logentry": {"message": f"calling {API_URL}"}, + "breadcrumbs": [{"data": {"url": API_URL}}], + "extra": {"safe": 1, "nested": ("tuple", API_URL)}, + } + cleaned = redact_bot_token_deep(event) + assert FAKE_TOKEN not in repr(cleaned) + # מבנה הנתונים נשמר — רק המחרוזות נוקו + assert cleaned["extra"]["safe"] == 1 + assert isinstance(cleaned["extra"]["nested"], tuple) + assert cleaned["exception"]["values"][0]["type"] == "TelegramAPIError" + + +def test_redact_deep_survives_self_referencing_structure(): + """מבנה מעגלי מנוקה במלואו — ההפניה המעגלית מצביעה על העותק המנוקה, לא על המקור.""" + node: dict = {"url": API_URL} + node["self"] = node + cleaned = redact_bot_token_deep(node) + assert FAKE_TOKEN not in str(cleaned["url"]) + # ההפניה המעגלית נסגרת על העותק המנוקה — אין דרך להגיע מהתוצאה לטוקן הגולמי + assert cleaned["self"] is cleaned + assert FAKE_TOKEN not in str(cleaned["self"]["url"]) + + +def test_redact_deep_handles_deep_nesting_without_leaking(): + """קינון עמוק (מעבר לכל cap) לא מחזיר לעולם את הערך הגולמי.""" + node: dict = {"url": API_URL} + for _ in range(150): + node = {"child": node} + cleaned = redact_bot_token_deep(node) + assert FAKE_TOKEN not in repr(cleaned) + + +def test_redact_deep_cleans_dict_keys_sets_and_namedtuples(): + import collections + + Point = collections.namedtuple("Point", ["x", "y"]) + obj = { + API_URL: "key-carrying-token", + "set": {API_URL, "safe"}, + "frozen": frozenset({API_URL}), + "named": Point(x=API_URL, y=1), + } + cleaned = redact_bot_token_deep(obj) + assert FAKE_TOKEN not in repr(cleaned) + # namedtuple נשמר כ-namedtuple, set נשאר set + named = next(v for v in cleaned.values() if isinstance(v, tuple) and hasattr(v, "_fields")) + assert named.y == 1 + assert isinstance(cleaned["set"], set) + assert isinstance(cleaned["frozen"], frozenset) + + +def test_error_payload_dict_is_redacted_and_structure_kept(): + payload = {"ok": False, "description": f"failed {API_URL}", "error_code": 400} + err = TelegramAPIError(error_code=400, description="Bad Request", url=API_URL, payload=payload) + assert FAKE_TOKEN not in repr(err.payload) + assert err.payload["error_code"] == 400 + + +def test_error_payload_string_is_redacted(): + err = TelegramAPIError(error_code=None, description="boom", url=None, payload=f"body {API_URL}") + assert FAKE_TOKEN not in err.payload + + +def test_logging_filter_redacts_bot_token(): + from utils import SensitiveDataFilter + + record = logging.LogRecord( + name="test", + level=logging.ERROR, + pathname=__file__, + lineno=1, + msg=f"request failed: {API_URL}", + args=(), + exc_info=None, + ) + SensitiveDataFilter().filter(record) + assert FAKE_TOKEN not in record.getMessage() + + +def test_logging_formatter_does_not_leak_token_from_traceback(): + """ה-Formatter מדביק את ה-traceback אחרי ההודעה — גם הוא חייב לצאת נקי.""" + import sys + + from utils import SensitiveDataFilter + + try: + raise RuntimeError(f"request failed: {API_URL}") + except RuntimeError: + exc_info = sys.exc_info() + + record = logging.LogRecord( + name="test", + level=logging.ERROR, + pathname=__file__, + lineno=1, + msg="clean message", + args=(), + exc_info=exc_info, + ) + SensitiveDataFilter().filter(record) + formatted = logging.Formatter().format(record) + assert FAKE_TOKEN not in formatted + # ה-traceback עצמו לא נעלם — רק הטוקן הוחלף + assert "RuntimeError" in formatted + # exc_info נשמר: Formatter משתמש בקאש exc_text המנוקה, ו-Sentry עוד צריך + # את החריגה המובנית (הניקוי שלה קורה ב-before_send) + assert record.exc_info is not None + + +def test_redact_deep_cleans_bytes_and_non_string_dict_keys(): + """גם bytes (גוף תגובה גולמי) וגם מפתחות מורכבים חייבים לצאת נקיים.""" + obj = { + ("tuple-key", API_URL): "value", + frozenset({API_URL}): "value2", + b"body": API_URL.encode("utf-8"), + "ba": bytearray(API_URL.encode("utf-8")), + } + cleaned = redact_bot_token_deep(obj) + assert FAKE_TOKEN not in repr(cleaned) + assert FAKE_TOKEN.encode("utf-8") not in cleaned[b"body"] + # הטיפוסים נשמרים — bytes נשאר bytes, bytearray נשאר bytearray + assert isinstance(cleaned[b"body"], bytes) + assert isinstance(cleaned["ba"], bytearray) + assert any(isinstance(k, tuple) for k in cleaned) + assert any(isinstance(k, frozenset) for k in cleaned) + + +def test_redact_deep_preserves_entries_on_key_collision(): + """שני מפתחות עם טוקנים שונים מתנקים לאותו ערך — אף שדה לא נדרס בשקט.""" + other_token = "999999999:BBcdqTcvCH1vGWJxfSeofSAs0K5PALDzzz" + obj = { + f"https://api.telegram.org/bot{FAKE_TOKEN}/sendMessage": "first", + f"https://api.telegram.org/bot{other_token}/sendMessage": "second", + } + cleaned = redact_bot_token_deep(obj) + assert FAKE_TOKEN not in repr(cleaned) + assert other_token not in repr(cleaned) + # שני הערכים שרדו — המפתח השני קיבל סיומת מספור במקום לדרוס את הראשון + assert len(cleaned) == 2 + assert sorted(cleaned.values()) == ["first", "second"] + # מדיניות המספור מפורשת: המפתח המאוחר מסתיים ב-#2 + assert any(isinstance(k, str) and k.endswith("#2") for k in cleaned) + + +def test_redact_deep_key_collision_on_non_string_keys(): + """התנגשות בין מפתחות tuple — המאוחר נעטף ב-(key, 2) במקום לדרוס.""" + other_token = "999999999:BBcdqTcvCH1vGWJxfSeofSAs0K5PALDzzz" + obj = { + ("chat", API_URL): "first", + ("chat", f"https://api.telegram.org/bot{other_token}/sendMessage"): "second", + } + cleaned = redact_bot_token_deep(obj) + assert FAKE_TOKEN not in repr(cleaned) + assert other_token not in repr(cleaned) + assert len(cleaned) == 2 + assert sorted(cleaned.values()) == ["first", "second"] + # המפתח המתנגש נעטף עם מונה: (המפתח המנוקה, 2) + assert any(isinstance(k, tuple) and len(k) == 2 and k[1] == 2 for k in cleaned) + + +def test_logging_formatter_redacts_github_and_bearer_from_traceback(): + """ה-traceback עובר את אותם דפוסי ניקוי כמו ההודעה — לא רק טלגרם.""" + import sys + + from utils import SensitiveDataFilter + + gh_token = "ghp_" + "a" * 30 + try: + raise RuntimeError(f"auth failed: {gh_token} Bearer abc123def456ghi789") + except RuntimeError: + exc_info = sys.exc_info() + + record = logging.LogRecord( + name="test", + level=logging.ERROR, + pathname=__file__, + lineno=1, + msg="clean message", + args=(), + exc_info=exc_info, + ) + SensitiveDataFilter().filter(record) + formatted = logging.Formatter().format(record) + assert gh_token not in formatted + assert "abc123def456ghi789" not in formatted + assert "REDACTED" in formatted + assert "RuntimeError" in formatted + assert record.exc_info is not None + + +def test_redact_deep_cleans_foreign_objects_carrying_token(): + """אובייקט זר (כמו מופע חריגה) שהייצוג שלו נושא טוקן מוחלף בייצוג המנוקה.""" + + class _Foreign: + def __str__(self): + return f"request to {API_URL} failed" + + cleaned = redact_bot_token_deep({"exc": _Foreign()}) + assert FAKE_TOKEN not in str(cleaned["exc"]) + assert "" in str(cleaned["exc"]) + + +def test_redact_deep_catches_token_hiding_only_in_repr(): + """Sentry ממיר אובייקטים זרים ב-repr — טוקן שמופיע רק שם חייב להיתפס.""" + + class _Foreign: + def __str__(self): + return "looks clean" + + def __repr__(self): + return f"" + + cleaned = redact_bot_token_deep({"exc": _Foreign()}) + assert FAKE_TOKEN not in str(cleaned["exc"]) + assert FAKE_TOKEN not in repr(cleaned["exc"]) + assert "" in str(cleaned["exc"]) + + +def test_redact_deep_preserves_clean_foreign_objects(): + """אובייקט זר נקי נשמר כמות שהוא — לא הופכים כל אירוע למחרוזות.""" + + class _Foreign: + def __str__(self): + return "nothing secret" + + obj = _Foreign() + cleaned = redact_bot_token_deep({"exc": obj, "num": 42, "flag": True, "none": None}) + assert cleaned["exc"] is obj + assert cleaned["num"] == 42 and cleaned["flag"] is True and cleaned["none"] is None + + +def test_logging_formatter_keeps_clean_exceptions_untouched(): + """חריגה בלי טוקן שומרת על exc_info — אין פגיעה במבנה עבור Sentry וכו'.""" + import sys + + from utils import SensitiveDataFilter + + try: + raise ValueError("nothing secret") + except ValueError: + exc_info = sys.exc_info() + + record = logging.LogRecord( + name="test", + level=logging.ERROR, + pathname=__file__, + lineno=1, + msg="clean message", + args=(), + exc_info=exc_info, + ) + SensitiveDataFilter().filter(record) + assert record.exc_info is not None + assert "ValueError" in logging.Formatter().format(record) diff --git a/utils.py b/utils.py index b570768ab..8ca24784e 100644 --- a/utils.py +++ b/utils.py @@ -1418,29 +1418,67 @@ def get_language_emoji(language: str) -> str: return emoji_map.get(language.lower(), '📄') +# טוקן של בוט טלגרם — מגיע ללוגים דרך כתובות ה-API (‎/bot/method). +# הדפוס מיובא מנקודת הניקוי המרכזית; אם הייבוא נכשל יש עותק מקומי זהה, כדי +# שהניקוי לעולם לא ידולג בגלל בעיית import (fail-closed). +try: + from telegram_api import _BOT_TOKEN_RE as _TG_TOKEN_RE # type: ignore +except Exception: # pragma: no cover + import re as _re_fallback + + _TG_TOKEN_RE = _re_fallback.compile(r"\d{5,16}:[A-Za-z0-9_-]{30,}") + + class SensitiveDataFilter(logging.Filter): """מסנן שמטשטש טוקנים ונתונים רגישים בלוגים.""" + + # כל דפוסי הניקוי במקום אחד — ההודעה וה-traceback עוברים דרך אותה רשימה, + # כך שאי אפשר להוסיף דפוס למסלול אחד ולשכוח את השני + _PATTERNS = [ + (re.compile(r"ghp_[A-Za-z0-9]{20,}"), "ghp_***REDACTED***"), + (re.compile(r"github_pat_[A-Za-z0-9_]{20,}"), "github_pat_***REDACTED***"), + (re.compile(r"Bearer\s+[A-Za-z0-9\-_.=:/+]{10,}"), "Bearer ***REDACTED***"), + (_TG_TOKEN_RE, ""), + ] + + @classmethod + def _redact_text(cls, text: str) -> str: + for pat, repl in cls._PATTERNS: + text = pat.sub(repl, text) + return text + def filter(self, record: logging.LogRecord) -> bool: try: - msg = str(record.getMessage()) - # זיהוי בסיסי של טוקנים: ghp_..., github_pat_..., Bearer ... - patterns = [ - (r"ghp_[A-Za-z0-9]{20,}", "ghp_***REDACTED***"), - (r"github_pat_[A-Za-z0-9_]{20,}", "github_pat_***REDACTED***"), - (r"Bearer\s+[A-Za-z0-9\-_.=:/+]{10,}", "Bearer ***REDACTED***"), - ] - redacted = msg - import re as _re - for pat, repl in patterns: - redacted = _re.sub(pat, repl, redacted) - # עדכן רק את message הפורמטי - record.msg = redacted + record.msg = self._redact_text(str(record.getMessage())) # חשוב: נקה ארגומנטים כדי למנוע ניסיון פורמט חוזר (%s) שיוביל ל-TypeError record.args = () + # Formatter.format מדביק את ה-traceback אחרי ההודעה, ולכן סוד בתוך + # טקסט החריגה (למשל URL של טלגרם בחריגת רשת) ידלוף גם אם ההודעה נקייה. + # מנקים רק כשבאמת נמצא סוד, כדי לא לפגוע במבנה החריגה במקרה הרגיל. + self._redact_exception(record) except Exception: pass return True + @classmethod + def _redact_exception(cls, record: logging.LogRecord) -> None: + try: + exc_text = record.exc_text + if not exc_text and record.exc_info: + import traceback as _tb + + exc_text = "".join(_tb.format_exception(*record.exc_info)) + if not exc_text: + return + cleaned = cls._redact_text(exc_text) + if cleaned != exc_text: + # Formatter.format משתמש ב-exc_text כשהוא כבר מוגדר ולא מרנדר את + # exc_info מחדש, ולכן די בקאש המנוקה. את exc_info משאירים — Sentry + # בונה ממנו את החריגה המובנית ומנקה אותה ב-before_send. + record.exc_text = cleaned + except Exception: + pass + def install_sensitive_filter(): """התקנת המסנן על כל ה-handlers הקיימים.""" diff --git a/webapp/app.py b/webapp/app.py index e8a284bea..599148d8d 100644 --- a/webapp/app.py +++ b/webapp/app.py @@ -1534,11 +1534,30 @@ def _ensure_metric(name: str, create_fn): install_sensitive_filter() except Exception: pass + def _sentry_before_send(event, hint): + """מנקה טוקני בוט מכל אירוע לפני שהוא נשלח ל-Sentry. + + ה-filter על ה-logging handlers לא מכסה חריגות שנתפסות ישירות + על ידי FlaskIntegration, ושם בדיוק יושבות כתובות ה-API של טלגרם. + fail-closed: אם הניקוי נכשל האירוע נזרק ולא נשלח גולמי. + """ + try: + from telegram_api import redact_bot_token_deep # type: ignore + + return redact_bot_token_deep(event) + except Exception: + try: + logger.warning("sentry event dropped: token redaction failed", extra={"event": "sentry_redaction_failed"}) + except Exception: + pass + return None + sentry_sdk.init( dsn=getattr(__import__('config'), 'config').SENTRY_DSN, integrations=[FlaskIntegration()], traces_sample_rate=0.05, environment=getattr(__import__('config'), 'config').ENVIRONMENT, + before_send=_sentry_before_send, ) except Exception: pass