Coverage for utils / sentry_init.py: 96.69%

352 statements  

« prev     ^ index     » next       coverage.py v7.13.4, created at 2026-06-17 01:43 +0000

1"""Sentry SDK初期化モジュール 

2 

3エラー監視サービスSentryとの統合を提供。 

4structlogと連携し、ERROR以上のログをSentryに送信。 

5 

6依存関係: 

7 - config/settings.py: SentryConfig(DSN、有効化フラグ等) 

8 - sentry-sdk[httpx] >= 2.61.0: before_send / new_scope APIを使用 

9 

10初期化タイミング: 

11 アプリケーション起動時、ログ設定後に一度だけ呼び出し。 

12 structlogのconfigure()後、最初のログ出力前が推奨。 

13 

14使用例: 

15 from utils.sentry_init import init_sentry 

16 

17 # アプリケーション起動時 

18 if init_sentry(): 

19 logger.info("Sentry monitoring enabled") 

20 

21デバッグ: 

22 初期化失敗時は warning ログを常時出力する(本番監視対応)。 

23 

24セキュリティ: 

25 - before_sendフックで機密データを自動除外(44種類のキーパターン) 

26 - DSNはSecretStrで管理(config/settings.py) 

27 - enabled=Falseで完全無効化可能 

28""" 

29 

30from __future__ import annotations 

31 

32import re 

33import secrets 

34import sys 

35import threading 

36from functools import lru_cache 

37from typing import TYPE_CHECKING, Any, cast 

38from urllib.parse import parse_qsl, unquote, urlencode, urlparse, urlunparse 

39 

40from config.settings import get_settings 

41from utils.logger import get_logger 

42 

43# 再帰防止用の内部識別子。scrub 失敗を Sentry に通知する際にこの tag を付与し、 

44# _before_send 冒頭で検出して scrub をスキップ通過させることで無限ループ 

45# (capture_message → _before_send → 例外 → capture_message ...) を遮断する。 

46# value はプロセス起動ごとにランダム生成し、OSS で公開された固定値を悪用した 

47# scrub バイパス (任意 event tag への注入で _before_send を素通りさせる攻撃) を防ぐ。 

48# _has_internal_tag / _emit_scrub_failure_to_sentry は 

49# 同一モジュール global を参照するため一貫して機能する。 

50_INTERNAL_TAG_KEY: str = "sentry_internal_event" 

51_INTERNAL_TAG_VALUE: str = secrets.token_hex(16) 

52 

53 

54def _has_internal_tag(tags: Any) -> bool: 

55 """内部識別 tag の有無を dict / list[tuple] 両形式で検出する。 

56 

57 Sentry SDK の現行版 (sentry-sdk >= 2.x) では ``scope.set_tag`` 経由で event に 

58 到達する ``tags`` は常に dict 形式 (``Scope._apply_tags_to_event`` 参照)。 

59 ただし ``_before_send`` 自体は SDK 仕様に従い list[tuple[str, str]] 形式も 

60 受け入れる契約 (``test_before_send_list_tags_redacts_sensitive_key``) のため、 

61 防御の対称性として recursion guard も両形式に対応する (defense-in-depth)。 

62 

63 dict が常態の現行 SDK では list 分岐は一見 YAGNI に見えるが、 

64 上記の ``_before_send`` list[tuple] 受け入れ契約をテストが独立に検証しているため 

65 意図的に維持する。当該契約テスト廃止時にのみ本分岐の削除を検討すること。 

66 

67 判定仕様: 

68 - dict 形式: ``tags[_INTERNAL_TAG_KEY] == _INTERNAL_TAG_VALUE`` で判定。 

69 - list 形式: ``(_INTERNAL_TAG_KEY, _INTERNAL_TAG_VALUE)`` タプルの完全一致 

70 で判定 (key のみ一致 / value が異なるペアは通常 event として扱い、 

71 recursion guard は発火しない)。 

72 - 上記以外 (None, str, int 等): 常に False。 

73 

74 Args: 

75 tags: event["tags"] 相当の値 (dict / list / その他)。 

76 

77 Returns: 

78 内部識別 tag が検出された場合 True。 

79 """ 

80 if isinstance(tags, dict): 

81 return tags.get(_INTERNAL_TAG_KEY) == _INTERNAL_TAG_VALUE 

82 if isinstance(tags, list): 

83 return (_INTERNAL_TAG_KEY, _INTERNAL_TAG_VALUE) in tags 

84 return False 

85 

86 

87_logger = get_logger(__name__) 

88 

89 

90def _safe_log_warning(event: str, **fields: Any) -> None: 

91 """PII scrub フロー内で fail-open 用に warning ログを送出する(例外抑止)。 

92 

93 ``_scrub_sensitive_data`` / ``_scrub_exception_field`` の 6 箇所に重複していた 

94 ``try: _logger.warning(...) / except Exception: pass` パターンをDRY 化したヘルパー。 

95 関数内で ``# noqa: BLE001, S110`` を 1 箇所に集約し、 

96 ``try/except Exception: pass`` でロガー例外を抑止する。 

97 

98 fail-open ロジック内部でのみ使用する想定。logger 失敗時は完全無音を避け、 

99 最低限 stderr へ通知する(障害隠蔽防止)。``except Exception`` は抑止するが 

100 MemoryError / RecursionError は fail-fast で再 raise する。 

101 

102 **PII 漏洩防止**: ``event`` 引数は静的な識別子文字列 (例: "sentry_field_type_unexpected") 

103 のみを渡すこと。動的なユーザーデータや変数値を直接渡すと、ログ経由で PII が 

104 漏洩する。動的な値は ``**fields`` の keyword 引数として渡すこと。 

105 """ 

106 try: 

107 _logger.warning(event, **fields) 

108 except (MemoryError, RecursionError): # fmt: skip 

109 # MemoryError / RecursionError は Exception 派生のため、再raise しないと 

110 # 下流の except Exception に捕捉されサイレント隠蔽される。 

111 # 致命的エラーとして必ず再raise(fail-fast)。 

112 # `# fmt: skip`: ruff format はタプル括弧を除去するが、Python 3.14 (PEP 758) 

113 # では括弧なし `except A, B:` も有効な構文(旧 Py2 binding ではない)。 

114 # 可読性のため括弧付きタプルを保持する(utils/ 全体で統一の規約)。 

115 raise 

116 except Exception as exc: # noqa: BLE001 

117 # ロガー失敗 → fail-open(イベント drop 防止)だが、 

118 # 完全無音は障害を隠蔽するため最低限 stderr に通知する。 

119 try: 

120 print( 

121 f"[sentry_init] _safe_log_warning failed: " 

122 f"event={event!r} error_type={type(exc).__name__} " 

123 f"error_module={type(exc).__module__} " 

124 # fields のキー名は呼び出し元によっては機密語を含みうるため件数のみ出力する 

125 # (event 名で呼び出し箇所は特定可能)。 

126 f"fields_count={len(fields)}", 

127 file=sys.stderr, 

128 flush=True, 

129 ) 

130 except Exception: # noqa: BLE001, S110 

131 # stderr 自体が壊れている場合は本当に何もできない 

132 pass 

133 

134 

135def _scrub_exception_frame(frame: Any) -> Any: 

136 """Sentry exception frame を fail-open でスクラブする。""" 

137 if not isinstance(frame, dict): 

138 _safe_log_warning( 

139 "sentry_exception_frame_unexpected_type", 

140 actual_type=type(frame).__name__, 

141 action="skip_frame_scrub", 

142 pii_leak_risk="HIGH", 

143 ) 

144 return frame 

145 

146 scrubbed_frame = dict(frame) 

147 # ソースコンテキスト (pre_context / context_line / post_context) を除去する。 

148 # Sentry はデフォルトでエラー行周辺のソース行を収集するが、ソース中にハードコードされた 

149 # 機密値や PII がそのまま送信されるリスクがある (CWE-312)。_scrub_sensitive_data は 

150 # vars (変数値) のみを対象としソース行テキストは非カバーのため、防御の深さとして drop する。 

151 for _source_context_key in ("pre_context", "context_line", "post_context"): 

152 scrubbed_frame.pop(_source_context_key, None) 

153 frame_vars = scrubbed_frame.get("vars") 

154 if isinstance(frame_vars, dict): 

155 scrubbed_frame["vars"] = _scrub_sensitive_data(frame_vars) 

156 elif frame_vars is not None: 

157 _safe_log_warning( 

158 "sentry_exception_frame_vars_unexpected_type", 

159 actual_type=type(frame_vars).__name__, 

160 action="skip_vars_scrub", 

161 ) 

162 return scrubbed_frame 

163 

164 

165def _scrub_exception_stacktrace(stacktrace: dict[str, Any]) -> dict[str, Any]: 

166 """Sentry exception stacktrace を fail-open でスクラブする。""" 

167 frames = stacktrace.get("frames") 

168 if isinstance(frames, list): 

169 scrubbed_stacktrace = dict(stacktrace) 

170 scrubbed_stacktrace["frames"] = [_scrub_exception_frame(frame) for frame in frames] 

171 return scrubbed_stacktrace 

172 if frames is not None: 

173 _safe_log_warning( 

174 "sentry_exception_frames_unexpected_type", 

175 actual_type=type(frames).__name__, 

176 action="skip_frames_scrub", 

177 ) 

178 return stacktrace 

179 

180 

181def _scrub_exception_value_item_extra_keys(scrubbed_value: dict[str, Any]) -> None: 

182 """exception value item の value/stacktrace 以外のトップレベルキーを in-place で 

183 機密スクラブする。 

184 

185 標準 Sentry exception value item は type/module/mechanism 等で PII を含まないが、 

186 カスタム SDK 統合が token/password 等を value item に直接付与した場合の漏洩を 

187 防ぐ defense-in-depth。``type`` は ``_is_sensitive_key`` 非該当のため redact されず 

188 観測性を保つ(S-1 の type 非 redact 方針と両立)。dict 値の反復中に値のみを 

189 更新する(キーの追加/削除はしないため反復は安全)。 

190 """ 

191 for key in scrubbed_value: 

192 if key in ("value", "stacktrace"): 

193 continue 

194 item = scrubbed_value[key] 

195 if _is_sensitive_key(key): 

196 scrubbed_value[key] = "[REDACTED]" 

197 elif isinstance(item, dict): 197 ↛ 198line 197 didn't jump to line 198 because the condition on line 197 was never true

198 scrubbed_value[key] = _scrub_sensitive_data(item) 

199 elif isinstance(item, (list, tuple)): 199 ↛ 200line 199 didn't jump to line 200 because the condition on line 199 was never true

200 scrubbed_value[key] = type(item)(_scrub_list_item(elem, _depth=0) for elem in item) 

201 

202 

203def _scrub_exception_value_item(value_item: Any) -> Any: 

204 """Sentry exception values[*] を fail-open でスクラブする。 

205 

206 value フィールドは例外メッセージ文字列のため PII 保護優先で無条件 [REDACTED] に 

207 置換する(観測性とのトレードオフ。詳細は _scrub_exception_field docstring 参照)。 

208 """ 

209 if not isinstance(value_item, dict): 

210 if isinstance(value_item, (list, tuple)): 

211 return type(value_item)(_scrub_list_item(item, _depth=0) for item in value_item) 

212 # str/bytes は PII を含む可能性があるため内容確認せず一律 REDACT。 

213 # int/float/bool は PII 非含のため素通し。 

214 if isinstance(value_item, (str, bytes)): 

215 return "[REDACTED]" 

216 _safe_log_warning( 

217 "sentry_exception_value_item_unexpected_type", 

218 actual_type=type(value_item).__name__, 

219 action="skip_item", 

220 ) 

221 return value_item 

222 

223 scrubbed_value = dict(value_item) 

224 if isinstance(scrubbed_value.get("value"), (str, bytes)): 

225 scrubbed_value["value"] = "[REDACTED]" 

226 

227 stacktrace = scrubbed_value.get("stacktrace") 

228 if isinstance(stacktrace, dict): 

229 scrubbed_value["stacktrace"] = _scrub_exception_stacktrace(stacktrace) 

230 elif stacktrace is not None: 

231 _safe_log_warning( 

232 "sentry_exception_stacktrace_unexpected_type", 

233 actual_type=type(stacktrace).__name__, 

234 action="skip_stacktrace_scrub", 

235 ) 

236 

237 # value / stacktrace 以外のトップレベルキーも機密判定する。 

238 _scrub_exception_value_item_extra_keys(scrubbed_value) 

239 

240 return scrubbed_value 

241 

242 

243if TYPE_CHECKING: 243 ↛ 244line 243 didn't jump to line 244 because the condition on line 243 was never true

244 from sentry_sdk.types import Event, Hint 

245 

246# 機密データキーのパターン(Security Auditor推奨を反映) 

247SENSITIVE_KEYS: frozenset[str] = frozenset( 

248 { 

249 # 認証系(基本) 

250 "password", 

251 "token", 

252 "secret", 

253 "api_key", 

254 "dsn", 

255 "authorization", 

256 "cookie", 

257 "session", 

258 "credential", 

259 # 認証系(拡張) 

260 "bearer", 

261 "jwt", 

262 "access_token", 

263 "refresh_token", 

264 "private_key", 

265 "client_secret", 

266 "x-api-key", 

267 "auth_token", 

268 "authtoken", # 複合語バリアント — 単語境界検出の false negative 補完 (#4) 

269 "usertoken", # 複合語バリアント 

270 "userpassword", # 複合語バリアント 

271 "passwd", 

272 # 暗号化 

273 "encryption_key", 

274 "cipher_key", 

275 # OAuth 

276 "oauth_token", 

277 # 二要素認証 

278 "otp", 

279 "mfa", 

280 "totp", 

281 # 個人情報 

282 "email", # GDPR/個人情報保護法: メールアドレスは個人識別情報 

283 "ip_address", # Sentry user.ip_address は個人識別情報として扱う 

284 "username", # Sentry user.username は個人識別情報として扱う 

285 "database_url", 

286 "ssn", 

287 "credit_card", 

288 "cvv", 

289 "card_number", 

290 # HTTPレスポンスプレビュー: _before_send は capture_exception / capture_message の 

291 # 両方で適用され、body_preview は extra 経由のペイロードをスクラブする。 

292 "body_preview", 

293 "access_key", 

294 "proxy-authorization", 

295 "set-cookie", 

296 "x-auth-token", 

297 "x-csrf-token", 

298 "csrf_token", 

299 "x-refresh-token", 

300 "x-access-token", 

301 }, 

302) 

303 

304# 遅延初期化フラグ 

305_sentry_initialized: bool = False 

306_sentry_init_lock = threading.Lock() 

307 

308 

309# 再帰制限のデフォルト値 

310MAX_SCRUB_DEPTH: int = 10 # 実測値 2-4、余裕値 10 は infinite recursion 防止用 

311 

312# before_send / before_send_transaction でスクラブ対象とするイベントフィールド 

313# (PII漏洩防止の対象集合)。error / transaction 双方が同一 _before_send を通る。 

314_SCRUBBED_EVENT_FIELDS: frozenset[str] = frozenset( 

315 { 

316 "extra", 

317 "user", 

318 "contexts", 

319 "tags", 

320 "breadcrumbs", 

321 # transaction イベント固有の子span配列(spans[*].data/description/tags)を 

322 # スクラブする。before_send_transaction 経路でのみ実在し、error イベントには 

323 # 当該キーが無いため _scrub_sentry_field の `if field in event_dict` で安全スキップ。 

324 "spans", 

325 # values[*].value は _is_sensitive_key の結果に関わらず 

326 # 無条件 [REDACTED] 置換(PII漏洩防止のため) 

327 "exception", 

328 } 

329) 

330 

331# defense-in-depth: ハイフン/アンダースコア表記揺れを吸収するため、 

332# SENSITIVE_KEYS と検査対象キー双方をハイフン→アンダースコアへ正規化してから 

333# 単語境界(先頭/末尾/アンダースコア)で判定する。これにより以下を同一視できる: 

334# - X-Auth-Token / x-auth-token / x_auth_token 

335# - Set-Cookie / set_cookie 

336# また composite key (例: user_password, email_address, session_id, auth_token_v2, 

337# customer_jwt) も単語単位で redact される一方、photo_url / prototype 等の 

338# unrelated substring は過剰 redact しない。 

339# 既知の false negative: ssnumber / cvvcode / foopassword 等の複合語は suffix が [a-z] のため 

340# suffix lookahead で非一致。頻出バリアントは SENSITIVE_KEYS に明示追加済み (#4)。 

341_NORMALIZED_SENSITIVE_KEYS: frozenset[str] = frozenset( 

342 sensitive.replace("-", "_") for sensitive in SENSITIVE_KEYS 

343) 

344# プレフィックス境界 `(?:^|[_\d])` はハイフンを含まないが、入力キーは _is_sensitive_key の 

345# ステップ4 (`key_norm.lower().replace("-", "_")`) でハイフンがアンダースコアへ正規化済みのため、 

346# `x-auth-token` → `x_auth_token` として `_` 境界で正しくマッチする。 

347# 左境界 `[_\d]` は数字を含むが小文字英字を含まないため非対称: `v2token` → True 

348# (数字 `2` が左境界)/ `footoken` → False(小文字 `o` は境界外)。これは単語先頭の 

349# 機密語のみを検出し、複合語中の偶発的な部分一致を避ける意図的設計(テストで担保。 

350_SENSITIVE_KEY_PATTERN: re.Pattern[str] = re.compile( 

351 r"(?:^|[_\d])(?:" 

352 + "|".join( 

353 re.escape(sensitive) 

354 for sensitive in sorted(_NORMALIZED_SENSITIVE_KEYS, key=len, reverse=True) 

355 ) 

356 + r")(?=[^a-z]|$)" # suffix-PII(ssnumber/cvvcode等)対応 

357) 

358# 全大文字命名 fallback 用: アンダースコア除去後の完全一致集合 

359# APIKEY / ACCESSTOKEN 等は ACRONYM 分割が効かないため別途事前計算 

360_COMPACT_SENSITIVE_KEYS: frozenset[str] = frozenset( 

361 sensitive.replace("_", "") for sensitive in _NORMALIZED_SENSITIVE_KEYS 

362) 

363 

364# camelCase / ACRONYM 分割用の事前コンパイル済みパターン 

365# 生リテラル re.sub の re._cache 依存を排し、設計を統一するため、 

366# すべての正規表現をモジュールレベルで事前コンパイルしています。 

367_ACRONYM_PATTERN: re.Pattern[str] = re.compile(r"([A-Z]+)([A-Z][a-z])") 

368_CAMEL_PATTERN: re.Pattern[str] = re.compile(r"(?<=[a-z])(?=[A-Z])") 

369# URL パスセグメント内のメールアドレス形式 PII を検出して [REDACTED] に置換する (#16) 

370_PATH_PII_PATTERN: re.Pattern[str] = re.compile(r"[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}") 

371 

372 

373# maxsize=512 — Sentry イベントが持つユニークキー名は典型的に 50〜200 程度。 

374# 512 はその 2〜10 倍のマージン。SENSITIVE_KEYS の要素数 (44) とは無関係。 

375@lru_cache(maxsize=512) 

376def _is_sensitive_key(key: str) -> bool: 

377 """機密キーかどうかを判定する(単語境界一致 + ハイフン/アンダースコア正規化)。 

378 

379 判定アルゴリズム: 

380 1. ACRONYMWord → ACRONYM_Word 変換(`APIKey` → `API_Key`) 

381 2. camelCase → snake_case 変換(`accessToken` → `access_Token`) 

382 3. key を lower-case 化 

383 4. ハイフン (``-``) およびドット (``.``) をアンダースコア (``_``) へ正規化 

384 (`database.url` → `database_url`、`x-auth-token` → `x_auth_token`) 

385 5. ``_NORMALIZED_SENSITIVE_KEYS`` の各要素を単語境界で検索 

386 6. 全大文字命名 fallback: アンダースコア除去後の完全一致 

387 (`APIKEY` → `apikey` == `api_key` compact → True) 

388 

389 これにより `user_password`, `email_address`, `X-Auth-Token` 等の 

390 composite key / HTTP header variant を redact しつつ、`photo_url`, 

391 `prototype`, `option` 等の unrelated substring は保持する。 

392 camelCase キー (`accessToken`, `apiKey`, `emailAddress`) も正規化後に 

393 snake_case として検出される。 

394 全大文字命名 (`APIKEY`, `ACCESSTOKEN`) は ACRONYM 分割が効かないため 

395 compact fallback(アンダースコア除去後の完全一致)で補完する。 

396 compact fallback は substring 一致ではなく完全一致のため、 

397 `PHOTOURL` (compact: photourl) が `url` にマッチして過剰 redact する問題は発生しない。 

398 

399 Args: 

400 key: 判定対象のキー文字列。 

401 

402 Returns: 

403 機密キーと判定された場合 True。 

404 

405 """ 

406 key_norm = _ACRONYM_PATTERN.sub(r"\1_\2", key) # ACRONYMWord → ACRONYM_Word 

407 key_norm = _CAMEL_PATTERN.sub("_", key_norm) # wordWord → word_Word(lower()前に分割) 

408 key_norm = key_norm.lower().replace("-", "_").replace(".", "_") 

409 if _SENSITIVE_KEY_PATTERN.search(key_norm) is not None: 

410 return True 

411 # 全大文字命名 fallback (APIKEY, ACCESSTOKEN 等): ACRONYM 分割が効かない場合に 

412 # アンダースコア除去後の完全一致で補完 

413 return key_norm.replace("_", "") in _COMPACT_SENSITIVE_KEYS 

414 

415 

416def _scrub_list_item(item: Any, _depth: int) -> Any: 

417 """tags 以外のフィールド(breadcrumbs / extra / contexts 等)向けの汎用 list 要素スクラブ。 

418 tags 専用の ``_scrub_tags_item`` と異なり、list[2] を (key, value) ペアとして扱わない 

419 ため、breadcrumbs 内の2要素 list を誤って tag pair 判定して過剰 redact するリスクがない。 

420 ``_scrub_sentry_field`` が field!="tags" の場合に本関数を呼ぶこと。 

421 

422 tags 専用の (key, value) ペア判定は ``_scrub_sentry_field`` の field 単位 

423 dispatch に集約し、本関数では tuple のみを tag pair として扱う 

424 (Sentry SDK が tags を list[tuple[str, str]] で渡す場合の互換性のため)。 

425 list[2] with str[0] のような汎用 list を tag pair と誤判定して 

426 breadcrumb 等の非 PII 2要素 list を過剰 redact する問題を避ける。 

427 """ 

428 if _depth >= MAX_SCRUB_DEPTH: 

429 _safe_log_warning("scrub_max_depth_exceeded", depth=_depth, max=MAX_SCRUB_DEPTH) 

430 return "[MAX_DEPTH_EXCEEDED]" 

431 if isinstance(item, tuple): 

432 if len(item) == 2: 432 ↛ 440line 432 didn't jump to line 440 because the condition on line 432 was always true

433 key = item[0] 

434 if isinstance(key, str) and _is_sensitive_key(key): 

435 return (key, "[REDACTED]") 

436 # len==2 非機密キー tuple を含む全 tuple の各要素を再帰スクラブし、 

437 # ネストされた dict/list 内の機密キーの漏洩を防ぐ。 

438 # 注: 素の str/数値要素はキーコンテキストを持たないため redact 対象外 

439 # (キーベース scrub の仕様限界)。 

440 return tuple(_scrub_list_item(elem, _depth + 1) for elem in item) 

441 if isinstance(item, dict): 

442 return _scrub_sensitive_data(item, _depth + 1) 

443 if isinstance(item, list): 

444 return [_scrub_list_item(child, _depth + 1) for child in item] 

445 return item 

446 

447 

448def _scrub_span_item(item: Any, _depth: int) -> Any: 

449 """Sentry transaction span の単一要素をスクラブする。""" 

450 if _depth >= MAX_SCRUB_DEPTH: 

451 _safe_log_warning("scrub_max_depth_exceeded", depth=_depth, max=MAX_SCRUB_DEPTH) 

452 return "[MAX_DEPTH_EXCEEDED]" 

453 if not isinstance(item, dict): 

454 return _scrub_list_item(item, _depth + 1) 

455 

456 scrubbed = _scrub_sensitive_data(item, _depth + 1) 

457 if not isinstance(scrubbed, dict): 457 ↛ 463line 457 didn't jump to line 463 because the condition on line 457 was never true

458 # _depth+1 が MAX_SCRUB_DEPTH に到達すると _scrub_sensitive_data は 

459 # "[MAX_DEPTH_EXCEEDED]" (str) を返す。str.get() による AttributeError → 

460 # _before_send の except 捕捉 → イベントのサイレントドロップを防ぐ防御ガード 

461 # 現呼び出し元 _scrub_sentry_field は _depth=0 固定のため到達 

462 # しないが、将来 _scrub_span_item が深い再帰文脈から呼ばれた場合の保険。 

463 return scrubbed 

464 description = scrubbed.get("description") 

465 if isinstance(description, str): 

466 method, separator, target = description.partition(" ") 

467 value_to_scrub = target if separator else description 

468 if "?" in value_to_scrub or "#" in value_to_scrub: 

469 scrubbed_description = _scrub_url(value_to_scrub) 

470 else: 

471 scrubbed_description = _PATH_PII_PATTERN.sub("[REDACTED]", value_to_scrub) 

472 scrubbed["description"] = ( 

473 f"{method} {scrubbed_description}" if separator else scrubbed_description 

474 ) 

475 return scrubbed 

476 

477 

478def _scrub_sensitive_data(data: Any, _depth: int = 0) -> Any: 

479 """機密データを再帰的にスクラブ 

480 

481 Args: 

482 data: スクラブ対象のデータ 

483 _depth: 現在の再帰深度(内部使用) 

484 

485 Returns: 

486 スクラブ済みデータ(元データは変更しない) 

487 

488 Note: 

489 MAX_SCRUB_DEPTH(デフォルト10)を超えると再帰を停止し、 

490 循環参照による無限ループを防止する。 

491 

492 """ 

493 if not isinstance(data, dict): 

494 # fail-open: 非dict入力はスクラブ不可。警告を残してそのまま返す。 

495 # _scrub_sentry_field の isinstance(value, dict) ガードと二重防御。 

496 _safe_log_warning( 

497 "scrub_sensitive_data_unexpected_type", 

498 actual_type=type(data).__name__, 

499 action="return_as_is", 

500 ) 

501 return data 

502 

503 # 再帰制限チェック(循環参照対策) 

504 if _depth >= MAX_SCRUB_DEPTH: 

505 _safe_log_warning("scrub_max_depth_exceeded", depth=_depth, max=MAX_SCRUB_DEPTH) 

506 return "[MAX_DEPTH_EXCEEDED]" 

507 

508 result: dict[str, Any] = {} 

509 for key, value in data.items(): 

510 if _is_sensitive_key(key): 

511 result[key] = "[REDACTED]" 

512 elif isinstance(value, dict): 

513 result[key] = _scrub_sensitive_data(value, _depth + 1) 

514 elif isinstance(value, list): 

515 result[key] = [_scrub_list_item(item, _depth + 1) for item in value] 

516 else: 

517 result[key] = value 

518 return result 

519 

520 

521def _scrub_request_field(value: Any) -> Any: 

522 """Sentry request.* フィールドを fail-closed でスクラブする。""" 

523 if isinstance(value, dict): 

524 return _scrub_sensitive_data(value) 

525 if isinstance(value, list): 525 ↛ 526line 525 didn't jump to line 526 because the condition on line 525 was never true

526 return [_scrub_list_item(item, _depth=0) for item in value] 

527 

528 _safe_log_warning( 

529 "sentry_request_field_type_unexpected", 

530 actual_type=type(value).__name__, 

531 action="replaced_with_redacted", 

532 ) 

533 return "[REDACTED]" 

534 

535 

536def _scrub_query_string(query_string: str) -> str: 

537 """重複キーを保持したままクエリ文字列をスクラブする。""" 

538 pairs = parse_qsl(query_string, keep_blank_values=True) 

539 scrubbed_pairs = [ 

540 (key, "[REDACTED]" if _is_sensitive_key(key) else value) for key, value in pairs 

541 ] 

542 return urlencode(scrubbed_pairs) 

543 

544 

545def _scrub_path_params(params: str) -> str: 

546 """RFC 2396 パスパラメータ (`;key=value` 形式) をスクラブする。 

547 

548 query string (_scrub_query_string) と同じ機密キー分類を使い、path 固有の 

549 email PII 除去も行う 

550 

551 アルゴリズム: 

552 - params を ";" で split。 

553 - 各 part に "=" を含む場合は partition で key を抽出。 

554 - `_is_sensitive_key(unquote(key))` が True → `{key}=[REDACTED]` に置換。 

555 - False → email PII (_PATH_PII_PATTERN) を [REDACTED] に置換して保持。 

556 - "=" を含まない part → email PII を [REDACTED] に置換して保持。 

557 - ";" で再結合して返す。 

558 

559 Args: 

560 params: urlparse の params フィールド(先頭の `;` を除いた文字列)。 

561 

562 Returns: 

563 スクラブ済みパスパラメータ文字列。 

564 """ 

565 scrubbed_parts: list[str] = [] 

566 for part in params.split(";"): 

567 if "=" in part: 

568 key, sep, val = part.partition("=") 

569 scrubbed_key = _PATH_PII_PATTERN.sub("[REDACTED]", key) 

570 if not key or _is_sensitive_key(unquote(key)): 

571 scrubbed_parts.append(f"{scrubbed_key}{sep}[REDACTED]") 

572 else: 

573 # 非機密キーでも値中の email PII は除去(防御維持) 

574 scrubbed_parts.append( 

575 f"{scrubbed_key}{sep}{_PATH_PII_PATTERN.sub('[REDACTED]', val)}" 

576 ) 

577 else: 

578 # "=" を含まないセグメントは email PII のみ除去 

579 scrubbed_parts.append(_PATH_PII_PATTERN.sub("[REDACTED]", part)) 

580 return ";".join(scrubbed_parts) 

581 

582 

583def _scrub_request_query_string(query_string: str | bytes) -> str: 

584 """Sentry request.query_string の str/bytes 値を安全にスクラブする。""" 

585 if isinstance(query_string, bytes): 

586 query_string = query_string.decode("utf-8", errors="ignore") 

587 return _scrub_query_string(query_string) 

588 

589 

590def _scrub_url(url: str) -> str: 

591 """URLのuserinfo/fragmentを除去し、query・pathのPIIをスクラブする。 

592 

593 - query: `_scrub_query_string` でキーベーススクラブ 

594 - path: メールアドレス形式のPII (`_PATH_PII_PATTERN`) を [REDACTED] に置換 (#16) 

595 - params: RFC 2396 パスパラメータ (`;key=value` 形式) を `_scrub_path_params` で 

596 キーベーススクラブ + email PII 除去 

597 - fragment: 完全除去(PII漏洩防止) 

598 """ 

599 parsed = urlparse(url) 

600 hostname = parsed.hostname 

601 if hostname is None: 

602 netloc = "" 

603 else: 

604 netloc = f"[{hostname}]" if ":" in hostname else hostname 

605 try: 

606 port = parsed.port 

607 except ValueError: 

608 port = None 

609 if port is not None: 

610 netloc = f"{netloc}:{port}" 

611 # tuple 6 要素を全フィールド明示で構築する 

612 # (vs `parsed._replace(...)`): ParseResult に新フィールドが追加された場合、 

613 # 暗黙保持で意図しないフィールドが残るリスクを排除する fail-safe 設計。 

614 # フィールド順は ParseResult 定義に従う: (scheme, netloc, path, params, query, fragment) 

615 return urlunparse( 

616 ( 

617 parsed.scheme, 

618 netloc, 

619 _PATH_PII_PATTERN.sub("[REDACTED]", parsed.path), 

620 # RFC 2396 パスパラメータ: キーベーススクラブ + email PII 除去 

621 _scrub_path_params(parsed.params) if parsed.params else "", 

622 _scrub_query_string(parsed.query) if parsed.query else "", 

623 "", # fragment を除去(PII 漏洩防止) 

624 ) 

625 ) 

626 

627 

628def _scrub_tags_item(item: Any, _depth: int = 0) -> Any: 

629 """tags フィールド専用の要素スクラブ。``_scrub_list_item`` との違いは (key, value) 

630 ペア判定を list[2] にも拡張している点で、``_scrub_sentry_field`` が field=="tags" 

631 の場合のみ本関数を呼ぶ。tags 以外のフィールドには ``_scrub_list_item`` を使用すること。 

632 

633 Sentry SDK 仕様: tags は ``dict[str, str]`` または ``list[tuple[str, str]]``。 

634 JSON roundtrip で tuple が list 化されるため list[2] も tag pair として扱う。 

635 加えて非標準だが custom before_send hook 等で生じうる dict / nested list 形態でも 

636 機密キーを redact する(defense-in-depth)。 

637 _depth による再帰深さ保護を行い、MAX_SCRUB_DEPTH 超過時は安全サイドに倒す。 

638 """ 

639 if _depth >= MAX_SCRUB_DEPTH: 

640 _safe_log_warning("scrub_max_depth_exceeded", depth=_depth, max=MAX_SCRUB_DEPTH) 

641 return "[MAX_DEPTH_EXCEEDED]" 

642 if isinstance(item, (tuple, list)) and len(item) == 2: 

643 key = item[0] 

644 scrubbed_value = ( 

645 "[REDACTED]" 

646 if (isinstance(key, str) and _is_sensitive_key(key)) 

647 else _scrub_list_item(item[1], _depth + 1) 

648 ) 

649 # type(item) は tuple/list 両対応の汎用ファクトリ(意図的な動的構築) 

650 return cast(Any, type(item))([key, scrubbed_value]) 

651 if isinstance(item, dict): 

652 return _scrub_sensitive_data(item, _depth + 1) 

653 if isinstance(item, list): 

654 # nested list は汎用 scrub にフォールバック(深さを引き継ぐ) 

655 return [_scrub_list_item(child, _depth=_depth + 1) for child in item] 

656 if isinstance(item, tuple): 656 ↛ 658line 656 didn't jump to line 658 because the condition on line 656 was never true

657 # len != 2 の tuple も _scrub_list_item で汎用スクラブ(list との一貫性) 

658 return tuple(_scrub_list_item(child, _depth + 1) for child in item) 

659 return item 

660 

661 

662def _scrub_exception_field(exception_value: dict[str, Any]) -> dict[str, Any]: 

663 """Sentry exception フィールドを構造検証付きでスクラブする(fail-open)。 

664 

665 Sentry の exception 構造: 

666 {values: [{type, value, stacktrace: {frames: [{vars: {...}}]}}]} 

667 

668 各階層(values→stacktrace→frames→vars)で型を isinstance で検証し、 

669 未知の構造では警告を残してスキップする(破壊しない)。 

670 vars 内の機密キーは _scrub_sensitive_data() で redact する。 

671 values[*].value は例外メッセージ文字列として全体を `[REDACTED]` に置換する。 

672 これは PII 保護を観測性より優先する意図的なトレードオフであり、 

673 PII を含まない FileNotFoundError 等の診断情報も失われる。将来改善する場合は、 

674 特定キーワードのみをマスクする selective redact へ移行する。 

675 

676 **values[*].type は意図的に redact しない**: 

677 `type` フィールドは例外クラス名 (`ValueError`, `KeyError` 等) を保持し、 

678 通常 PII を含まない。例外分類は Sentry UI / alert routing / metric 集計の 

679 primary key として機能するため、redact すると観測性が壊滅的に低下する。 

680 ただし将来 `type` に PII 文字列が現れる SDK 拡張 / custom exception 

681 命名規則が導入された場合は本方針を再評価する必要がある。 

682 

683 設計トレードオフ (fail-open): 

684 Sentry SDK の将来バージョンや custom integration により未知の exception 

685 構造が渡された場合、本関数は破壊せず通過させる(observability 優先)。 

686 この設計には残存リスクが1点存在する: 

687 **未知構造の内側に機密データが含まれていた場合、scrub されず Sentry に 

688 到達する可能性がある**。 

689 最外殻 ``_before_send`` の ``except Exception`` 二重防御は、本関数が 

690 ``MemoryError`` / ``RecursionError`` 以外の例外を送出した場合にのみ 

691 ``_emit_scrub_failure_to_sentry`` 経由で event drop に倒すが、 

692 fail-open ロジックは例外を送出せず正常完了するため、未知構造による 

693 PII 漏洩は二重防御では検出されない(許容された残存リスク)。 

694 fail-closed への変更を検討する場合は、観測性低下(正常な例外イベント 

695 まで drop される)とのトレードオフを評価すること。 

696 

697 Args: 

698 exception_value: Sentry イベントの "exception" 辞書 

699 

700 Returns: 

701 スクラブ済み exception 辞書(元データは変更しない) 

702 

703 """ 

704 values = exception_value.get("values") 

705 if isinstance(values, dict): 

706 # dict 型 values: _scrub_sensitive_data で内容をスクラブして返す 

707 result = dict(exception_value) 

708 result["values"] = _scrub_sensitive_data(values) 

709 return result 

710 if values is None: 

711 # "values" キー未存在は Sentry exception interface 仕様上の有効な構造 

712 # (getsentry/sentry interfaces/exception.py: get_path(data, "values", 

713 # default=[]) で空リスト扱い、Relay schema でも values は必須でない)。 

714 # よって誤検知 WARNING を出さずに、他キーをベストエフォートスクラブして返す 

715 # 正常構造に対するログノイズ抑制 

716 scrubbed = _scrub_sensitive_data(exception_value) 

717 # _depth=0 開始のため dict 以外("[MAX_DEPTH_EXCEEDED]")は構造上発生しないが、 

718 # cast による型隠蔽を避け isinstance ガードで型安全を明示する。 

719 return scrubbed if isinstance(scrubbed, dict) else {} 

720 if not isinstance(values, list): 

721 # str / int 等、None でも list/dict でもない真に予期しない型のみ WARNING。 

722 _safe_log_warning( 

723 "sentry_exception_values_unexpected_type", 

724 actual_type=type(values).__name__, 

725 action="exception_scrub_fallback", 

726 ) 

727 # 構造不明でも _scrub_sensitive_data でベストエフォートスクラブ 

728 scrubbed = _scrub_sensitive_data(exception_value) 

729 # _depth=0 開始のため dict 以外("[MAX_DEPTH_EXCEEDED]")は構造上発生しないが、 

730 # cast による型隠蔽を避け isinstance ガードで型安全を明示する。 

731 return scrubbed if isinstance(scrubbed, dict) else {} 

732 

733 result = dict(exception_value) 

734 result["values"] = [_scrub_exception_value_item(val) for val in values] 

735 return result 

736 

737 

738def _scrub_sentry_field(event_dict: dict[str, Any], field: str) -> None: 

739 """Sentryイベントの単一フィールドをスクラブする(_before_send 専用)。 

740 

741 dict型の場合は _scrub_sensitive_data() で再帰的にスクラブする 

742 (``exception`` は _scrub_exception_field で values[*].value REDACTION と 

743 stackframe scrub を適用)。 

744 list型の場合は field 単位で dispatch する: ``exception`` は各要素へ 

745 _scrub_exception_value_item を適用し(defense-in-depth)、``tags`` は 

746 (key, value) ペア形式、その他(breadcrumbs 等)は dict/list 要素形式として 

747 スクラブする。非PII のデバッグ情報(リクエストID等)は保持する 

748 (Sentry SDK 仕様: tags は ``list[tuple[str, str]]`` 形式も許容)。 

749 上記以外の型の場合は安全サイドに置換する(fail-closed): ``exception`` は 

750 Sentry exception interface 準拠の無害なプレースホルダ(``ScrubbedException``)へ、 

751 それ以外は空dictへ置換し、logger.warning を常時出力する(本番監視対応)。 

752 _scrub_sensitive_data の内部 non-dict ガードと二重防御を構成する。 

753 

754 Args: 

755 event_dict: Sentryイベント辞書(破壊的更新) 

756 field: スクラブ対象フィールド名 

757 

758 """ 

759 if field in event_dict: 

760 value = event_dict[field] 

761 if isinstance(value, dict): 

762 if field == "exception": 

763 event_dict[field] = _scrub_exception_field(value) 

764 else: 

765 event_dict[field] = _scrub_sensitive_data(value) 

766 elif isinstance(value, list): 

767 # field 単位 dispatch: tags は Sentry spec 上 list[tuple[str, str]] 

768 # (JSON経由で list[list[str, str]] にもなり得る)なので tag pair として処理。 

769 # 加えて非標準だが custom before_send hook 等で生じうる list[dict] / list[list] 

770 # 形態でも defense-in-depth で内部の機密キーを redact する 

771 # (PR #347 review reflexion iter2: tags-as-list-of-dicts 退行修正)。 

772 # 他フィールド(breadcrumbs/contexts/extra/user)は汎用要素として 

773 # 処理し、list[2] with str[0] の偶発的 tag-pair 誤判定で過剰 redact 

774 # しない(PR #347 review: KP-003 / T3 対応)。 

775 if field == "exception": 

776 # exception が list 形式(Sentry 標準は dict だが custom before_send 

777 # 等で生じうる)でも values[*].value の REDACTION と stackframe vars 

778 # scrub を適用し、PII(例外メッセージ・frame 変数)の素通りを防ぐ (defense-in-depth) # noqa: E501 

779 # dict 要素は exception 専用スクラブ、非 dict 要素(custom hook が 

780 # 生成しうる list/tuple/scalar)は汎用 _scrub_list_item で再帰スクラブ 

781 # する。これにより tags/spans/その他(L697)の list 分岐と同様に 

782 # 「全分岐で非 dict 要素も再帰スクラブ・素通しゼロ」を満たし、dispatch の 

783 # 一貫性保持のため、fail-openによる非対称を解消。 

784 event_dict[field] = [ 

785 _scrub_exception_value_item(item) 

786 if isinstance(item, dict) 

787 else _scrub_list_item(item, _depth=0) 

788 for item in value 

789 ] 

790 elif field == "tags": 

791 event_dict[field] = [_scrub_tags_item(item, _depth=0) for item in value] 

792 elif field == "spans": 

793 event_dict[field] = [_scrub_span_item(item, _depth=0) for item in value] 

794 else: 

795 event_dict[field] = [_scrub_list_item(item, _depth=0) for item in value] 

796 else: 

797 # dict/list以外はスクラブ不可能なため、安全サイドに倒して置換する(fail-closed)。 

798 # exception フィールドは Sentry exception interface 仕様に準拠した 

799 # 無害なプレースホルダ構造へ置換し PII 素通りを防ぐ。 

800 # 他フィールドは空 dict に置換する。 

801 if field == "exception": 

802 event_dict[field] = { 

803 "values": [ 

804 { 

805 "type": "ScrubbedException", 

806 "value": "[REDACTED: unscrubable exception structure]", 

807 } 

808 ] 

809 } 

810 _safe_log_warning( 

811 "sentry_field_type_unexpected", 

812 field=field, 

813 actual_type=type(value).__name__, 

814 action="replaced_with_safe_placeholder", 

815 event_id=event_dict.get("event_id"), 

816 ) 

817 else: 

818 event_dict[field] = {} 

819 _safe_log_warning( 

820 "sentry_field_type_unexpected", 

821 field=field, 

822 actual_type=type(value).__name__, 

823 action="replaced_with_empty_dict", 

824 event_id=event_dict.get("event_id"), 

825 ) 

826 

827 

828# fail-closed防御による退避後、スクラビング失敗イベントを安全にSentryへ通知するフェーズ。 

829# 以下の定数は、その内部イベント(_INTERNAL_TAG付与)専用のextra許可リストである。 

830 

831# このセットに含まれるキーのみ _set_internal_extras 経由で設定でき、_before_send の 

832# scrub をバイパスして Sentry に到達する。PII を含み得るキーを追加してはならない 

833# 各キーは「scrub バイパスを許可する」明示的な grant であり、セキュリティ境界そのもの。 

834# 値は _emit_scrub_failure_to_sentry が設定する extra キーと 1:1 で対応させる 

835# (不足 → 実行時 ValueError / 過剰 → PII 素通りの穴)。 

836_INTERNAL_EVENT_EXTRA_KEYS: frozenset[str] = frozenset( 

837 { 

838 "error_type", 

839 "error_module", 

840 "action", 

841 "event_id", 

842 } 

843) 

844 

845 

846def _set_internal_extras(scope: Any, extras: dict[str, str | None]) -> None: 

847 """scrub バイパス内部イベント専用の extra セッター。 

848 

849 許可リスト ``_INTERNAL_EVENT_EXTRA_KEYS`` 外のキーを拒否し、将来の変更者が 

850 新規 extra キーを無検証で追加して PII を scrub バイパスさせるリスクを 

851 コードレベルで防止する(コメント規約のみの保証を技術的強制へ格上げ)。 

852 

853 Args: 

854 scope: Sentry スコープ(``new_scope()`` の戻り値)。 

855 extras: 設定する extra キー・値のマッピング。 

856 

857 Raises: 

858 ValueError: 許可リスト外のキーが含まれる場合。 

859 """ 

860 for key in extras: 

861 if key not in _INTERNAL_EVENT_EXTRA_KEYS: 

862 raise ValueError(f"Unauthorized key in internal event extra: {key!r}") 

863 

864 for key, value in extras.items(): 

865 scope.set_extra(key, value) 

866 

867 

868def _emit_scrub_failure_to_sentry(exc: BaseException, event_id: str | None = None) -> None: 

869 """scrub 失敗を内部識別 tag 付きで Sentry へ通知する(fail-safe)。 

870 

871 sentry_sdk.capture_message は再度 _before_send を通るが、付与した 

872 _INTERNAL_TAG_KEY=_INTERNAL_TAG_VALUE を冒頭で検出して scrub をスキップ 

873 通過させるため無限再帰しない。内部 SDK 呼び出し自体が例外を投げた場合は 

874 最終フォールバックとして stderr へ出力し、メインプロセスをクラッシュさせない。 

875 

876 **SDK バージョン制約**: 

877 再帰防止ガード(_INTERNAL_TAG_KEY による scrub スキップ)は Sentry SDK 2.x 

878 の内部動作(``Scope._apply_tags_to_event``)に依存する。 

879 pyproject.toml の ``sentry-sdk<3.0.0`` 制約により 3.x 以降の 

880 内部変更から保護している。3.x へ移行する際は本ガードの動作を再検証すること。 

881 

882 **Security constraint**: ``extra`` フィールドに 

883 PII を含めてはならない。本制約に違反した場合、内部通知イベントが scrub を 

884 バイパスして PII がそのまま Sentry に到達する。 

885 """ 

886 try: 

887 sentry_sdk = sys.modules.get("sentry_sdk") 

888 if sentry_sdk is None: 

889 # SDK 未ロード(テスト・未インストール環境): stderr フォールバック 

890 print( 

891 f"[SENTRY_SCRUB_FAILED] sentry_sdk_not_loaded " 

892 f"error_type={type(exc).__qualname__} " 

893 f"error_module={type(exc).__module__}", 

894 file=sys.stderr, 

895 flush=True, 

896 ) 

897 return 

898 with sentry_sdk.new_scope() as scope: 

899 scope.set_tag(_INTERNAL_TAG_KEY, _INTERNAL_TAG_VALUE) 

900 scope.set_level("error") 

901 # SECURITY: extra は許可リスト(_INTERNAL_EVENT_EXTRA_KEYS)経由でのみ設定 — 

902 # _before_send scrub バイパスのため、PII を含み得るキーの素通りをコードで防止する。 

903 extras: dict[str, str | None] = { 

904 "error_type": type(exc).__qualname__, 

905 "error_module": type(exc).__module__, 

906 "action": "event_dropped", 

907 } 

908 # event_id は UUID 形式のため PII ではない。 

909 # None の場合は extra に含めない(従来挙動を維持)。 

910 if event_id is not None: 

911 extras["event_id"] = event_id 

912 _set_internal_extras(scope, extras) 

913 scope.capture_message("sentry_scrub_failed", level="error") 

914 except (MemoryError, RecursionError): # fmt: skip 

915 # MemoryError/RecursionError も Exception 派生のため、再raise しないと直下の 

916 # except Exception に捕捉されサイレント隠蔽される。OOM/再帰超過を握り潰さず 

917 # 即座に fail-fast 伝播させる。コードベース他6箇所の統一パターン 

918 # `except (MemoryError, RecursionError): raise` に集約し、将来の修正漏れを防ぐ 

919 # 分離記述から統合、挙動は不変。 

920 # 再raise後は呼び出し元 _before_send の emit 保護 try/except が捕捉し、明示的に 

921 # return None するため fail-closed(PII 非送信)は維持される。 

922 raise 

923 except Exception as inner_exc: # noqa: BLE001 

924 # Sentry 通知自体が失敗 → stderr へ最終フォールバック 

925 print( 

926 f"[SENTRY_SCRUB_FAILED] inner_error_type={type(inner_exc).__qualname__} " 

927 f"inner_error_module={type(inner_exc).__module__} " 

928 f"original_error_type={type(exc).__qualname__} " 

929 f"original_error_module={type(exc).__module__}", 

930 file=sys.stderr, 

931 flush=True, 

932 ) 

933 

934 

935def _before_send(event: Event, hint: Hint) -> Event | None: # noqa: ARG001, C901 

936 """Sentry送信前フック(機密データ除外) 

937 

938 再帰防止: scrub 失敗時に発火する内部通知イベント(_INTERNAL_TAG_KEY 

939 が付与されている)は冒頭で検出し、scrub をスキップして通過させる。 

940 これにより capture_message → _before_send → 例外 → capture_message 

941 の無限ループを遮断する(_emit_scrub_failure_to_sentry 参照)。 

942 

943 Args: 

944 event: Sentryイベント 

945 hint: 追加コンテキスト(未使用) 

946 

947 Returns: 

948 処理済みイベント、またはNone(送信キャンセル) 

949 

950 """ 

951 # 再帰防止ガード: 内部通知イベントは scrub をスキップして通過させる 

952 # dict / list[tuple] 両形式に対応 (defense-in-depth, _has_internal_tag 参照) 

953 if _has_internal_tag(event.get("tags")): 

954 return event 

955 

956 # scrub_exc は except 句で捕捉した例外を try ブロックの外へ持ち出すための変数。 

957 # _emit_scrub_failure_to_sentry を except コンテキスト外で呼ぶことで sys.exc_info() が 

958 # アクティブな状態を避け、Sentry SDK が元例外(PII 付き)を内部通知イベントに添付する 

959 # 事故を防ぐ (fail-closed 防御。詳細は下方 `except Exception` 節を参照) 

960 scrub_exc: Exception | None = None 

961 # cast は実行時 no-op(型システム専用)— try 外に置いても動作変化なし 

962 event_dict = cast(dict[str, Any], event) 

963 try: 

964 # リクエストデータのスクラブ。成功時だけ event["request"] を差し替える。 

965 # _scrub_sensitive_data / _scrub_query_string / _scrub_url は全て non-destructive で 

966 # 新しいオブジェクトを返すため、top-level dict の shallow copy で十分。 

967 # 元 request の non-mutation 契約は既存テスト 

968 # ``test_before_send_fail_closed_without_partial_request_mutation`` で継続検証。 

969 if "request" in event_dict: 

970 request = event_dict["request"] 

971 if isinstance(request, dict): 

972 scrubbed_request = dict(request) 

973 for req_field in ("headers", "data", "cookies", "env"): 

974 if req_field in scrubbed_request: 

975 scrubbed_request[req_field] = _scrub_request_field( 

976 scrubbed_request[req_field] 

977 ) 

978 if "query_string" in scrubbed_request and isinstance( 

979 scrubbed_request["query_string"], (str, bytes) 

980 ): 

981 scrubbed_request["query_string"] = _scrub_request_query_string( 

982 scrubbed_request["query_string"] 

983 ) 

984 if "url" in scrubbed_request and isinstance(scrubbed_request["url"], str): 

985 scrubbed_request["url"] = _scrub_url(scrubbed_request["url"]) 

986 event_dict["request"] = scrubbed_request 

987 else: 

988 event_dict["request"] = {} 

989 _safe_log_warning( 

990 "sentry_request_type_unexpected", 

991 actual_type=type(request).__name__, 

992 action="replaced_with_empty_dict", 

993 event_id=event_dict.get("event_id"), 

994 ) 

995 

996 # 追加データのスクラブ(2層防御) 

997 # _scrub_sentry_field: 非dict型フィールドを空dictに置換(型安全化・PII漏洩防止) 

998 # _scrub_sensitive_data: dict内の機密キーを [REDACTED] に置換(PII除外) 

999 for field in _SCRUBBED_EVENT_FIELDS: 

1000 _scrub_sentry_field(event_dict, field) 

1001 except (MemoryError, RecursionError): # fmt: skip 

1002 # システム異常(OOM・スタックオーバーフロー)は fail-closed で event を 

1003 # ドロップする。before_send からの例外は Sentry SDK が内部 catch し 

1004 # PII 付きイベントをそのまま送信し続けるリスクがあるため、stderr への 

1005 # 最小通知だけ残して return None で安全に遮断する(CWE-391 対策)。 

1006 try: 

1007 print( 

1008 "[SENTRY_SCRUB_FAILED] before_send system error: MemoryError or RecursionError", 

1009 file=sys.stderr, 

1010 flush=True, 

1011 ) 

1012 except Exception: # noqa: BLE001, S110 

1013 pass 

1014 return None 

1015 except Exception as exc: 

1016 # sys.exc_info() がアクティブな状態で _emit_scrub_failure_to_sentry を 

1017 # 呼ぶと、Sentry SDK が元例外(PII 付き)を内部通知イベントに添付する 

1018 # リスクがあるため、例外コンテキストの外で呼び出す(fail-closed防御) 

1019 scrub_exc = exc 

1020 

1021 if scrub_exc is not None: 

1022 try: 

1023 _logger.error( 

1024 "sentry_before_send_drop_event", 

1025 error_type=type(scrub_exc).__qualname__, 

1026 error_module=type(scrub_exc).__module__, 

1027 event_id=event_dict.get("event_id"), 

1028 ) 

1029 except Exception: # noqa: BLE001 

1030 # ロガー自体が失敗した場合のフォールバック(_safe_log_warningはフォールバック専用) 

1031 _safe_log_warning( 

1032 "sentry_before_send_drop_event", 

1033 error_type=type(scrub_exc).__qualname__, 

1034 error_module=type(scrub_exc).__module__, 

1035 ) 

1036 # emit 呼び出しを try/exceptで保護する 

1037 # _emit_scrub_failure_to_sentry は内部でシステム異常 (MemoryError/RecursionError) を 

1038 # re-raise する設計(emit 自身の except Exception による隠蔽回避のため)。その re-raise が 

1039 # ここで未捕捉のまま伝播すると下方の return None を飛び越え、scrubbing 失敗イベントの 

1040 # fail-closed ドロップが Sentry SDK の capture_internal_exceptions 挙動に依存してしまう。 

1041 # 呼び出し側で全 Exception を捕捉して return None を保証することで、line 881 の 

1042 # (MemoryError, RecursionError) → return None パターンと一貫した fail-closed を SDK 非依存で 

1043 # 確定させる。KeyboardInterrupt / SystemExit は BaseException 直系のため捕捉せず伝播させる。 

1044 try: 

1045 _emit_scrub_failure_to_sentry(scrub_exc, event_id=event_dict.get("event_id")) 

1046 except Exception: # noqa: BLE001 

1047 try: 

1048 print( 

1049 "[SENTRY_SCRUB_FAILED] emit failed; event dropped (fail-closed)", 

1050 file=sys.stderr, 

1051 flush=True, 

1052 ) 

1053 except Exception: # noqa: BLE001, S110 

1054 pass 

1055 return None 

1056 

1057 return event 

1058 

1059 

1060def init_sentry() -> bool: 

1061 """Sentry SDKをプロセス内で一度だけ初期化する。""" 

1062 with _sentry_init_lock: 

1063 if _sentry_initialized: 

1064 return True 

1065 return _init_sentry_unlocked() 

1066 

1067 

1068def _init_sentry_unlocked() -> bool: # noqa: C901 

1069 """Sentry SDK初期化(呼び出し側で _sentry_init_lock を保持していること)。 

1070 

1071 config/settings.pyのSentryConfigに基づいて初期化。 

1072 enabled=Falseまたは空DSNの場合はスキップ。 

1073 

1074 Returns: 

1075 True: 初期化成功 

1076 False: スキップまたは失敗 

1077 

1078 Example: 

1079 if init_sentry(): 

1080 logger.info("Sentry monitoring enabled") 

1081 

1082 """ 

1083 global _sentry_initialized # noqa: PLW0603 

1084 

1085 settings = get_settings() 

1086 sentry_config = settings.sentry 

1087 

1088 # 無効化チェック 

1089 if not sentry_config.enabled: 

1090 return False 

1091 

1092 # DSN取得・検証 

1093 dsn = sentry_config.dsn.get_secret_value() 

1094 if not dsn: 

1095 return False 

1096 

1097 # 環境名(フォールバック) 

1098 # Note: DSN検証はsentry_sdk.init()に委任(SDK内部でバリデーション実施) 

1099 environment = sentry_config.environment or settings.environment.value 

1100 

1101 # is_production_like() を関数先頭で一度だけ評価しローカル変数に保持。 

1102 # except 内で get_settings() を再呼び出しすると、環境変数変化や reload_settings() 

1103 # の race により ValidationError が発生し、元の例外(ImportError / 初期化失敗)を 

1104 # マスクしてデバッグを困難化するリスクがあるため (CWE-755 例外マスク防止)。 

1105 is_production_like = settings.is_production_like() 

1106 

1107 try: 

1108 import sentry_sdk 

1109 

1110 sentry_sdk.init( 

1111 dsn=dsn, 

1112 environment=environment, 

1113 traces_sample_rate=sentry_config.traces_sample_rate, 

1114 profiles_sample_rate=sentry_config.profiles_sample_rate, 

1115 send_default_pii=sentry_config.send_default_pii, 

1116 before_send=_before_send, 

1117 # transaction イベントは before_send を通らない(SDK仕様)。同一 scrub 経路へ 

1118 # 配線し、span data / WSGI-ASGI 由来の request を PII スクラブする 

1119 # per-event scrub コストは traces_sample_rate(既定 0.1 = 低サンプリング)で 

1120 # 上限が画定されるため、現設定では累積負荷は許容範囲 

1121 before_send_transaction=_before_send, 

1122 ) 

1123 

1124 _sentry_initialized = True 

1125 return True 

1126 

1127 except ImportError as exc: 

1128 # sentry-sdk未インストール 

1129 # 本番環境では必須 → Fail-Fast(依存関係漏れを即座に検出) 

1130 if is_production_like: 

1131 raise RuntimeError( 

1132 f"Sentry SDK not installed in production: {exc}. Add 'sentry-sdk' to dependencies.", 

1133 ) from exc 

1134 

1135 # 開発/テスト環境では許容(ログ警告のみ) 

1136 # warnings.warn は filterwarnings('error') 環境で UserWarning を raise し、 

1137 # __context__ 経由で DSN が漏洩するリスクがあるため _logger.warning に変更。 

1138 try: 

1139 _logger.warning( 

1140 "sentry_sdk_not_installed", 

1141 error_type=type(exc).__name__, 

1142 error_module=type(exc).__module__, 

1143 ) 

1144 except Exception as logger_exc: # noqa: BLE001 

1145 # ロガー失敗時は stderr へフォールバック(PII 非露出)。 

1146 # _emit_scrub_failure_to_sentry と同様にエラー型/モジュールを記録し設計を統一 

1147 print( 

1148 "[SENTRY_WARN] sentry_sdk_not_installed " 

1149 f"original_error_type={type(exc).__name__} " 

1150 f"original_error_module={type(exc).__module__} " 

1151 f"logger_error_type={type(logger_exc).__name__} " 

1152 f"logger_error_module={type(logger_exc).__module__}", 

1153 file=sys.stderr, 

1154 flush=True, 

1155 ) 

1156 return False 

1157 

1158 except Exception as exc: 

1159 # その他の初期化失敗 - 本番環境では例外を発生させる(Fail-Fast) 

1160 if is_production_like: 

1161 raise RuntimeError( 

1162 f"Sentry initialization failed in production: {type(exc).__name__}", 

1163 ) from exc 

1164 

1165 # 開発/テスト環境ではログ警告のみ 

1166 # warnings.warn は filterwarnings('error') 環境で UserWarning を raise し、 

1167 # __context__ 経由で DSN が漏洩するリスクがあるため _logger.warning に変更。 

1168 try: 

1169 _logger.warning( 

1170 "sentry_init_failed", 

1171 error_type=type(exc).__name__, 

1172 error_module=type(exc).__module__, 

1173 ) 

1174 except Exception as logger_exc: # noqa: BLE001 

1175 # ロガー失敗時は stderr へフォールバック(PII 非露出)。 

1176 # _emit_scrub_failure_to_sentry と同様にエラー型/モジュールを記録し設計を統一 

1177 print( 

1178 "[SENTRY_WARN] sentry_init_failed " 

1179 f"original_error_type={type(exc).__name__} " 

1180 f"original_error_module={type(exc).__module__} " 

1181 f"logger_error_type={type(logger_exc).__name__} " 

1182 f"logger_error_module={type(logger_exc).__module__}", 

1183 file=sys.stderr, 

1184 flush=True, 

1185 ) 

1186 return False 

1187 

1188 

1189def is_sentry_initialized() -> bool: 

1190 """Sentry初期化状態の確認 

1191 

1192 Returns: 

1193 True: 初期化済み 

1194 False: 未初期化 

1195 

1196 """ 

1197 with _sentry_init_lock: 

1198 return _sentry_initialized 

1199 

1200 

1201def reset_sentry_state() -> None: 

1202 """Sentry状態リセット(テスト用) 

1203 

1204 Warning: 

1205 本番コードでは使用しないでください。 

1206 

1207 """ 

1208 global _sentry_initialized # noqa: PLW0603 

1209 with _sentry_init_lock: 

1210 _sentry_initialized = False