"""Outbound notifications for the bounty flow. Recipient discipline — this is the part that is easy to get wrong and expensive to get wrong: * The owner-facing pushes go to **exactly the person the security team named**. No manager, no skip-level, no walk-up the chain when the assignee is unreachable, no cron. One human action (assign / 催办) produces one push; 催办 is additionally throttled to 12h per report by ``service.record_nudge``. * The owner-verdict push goes to **IT only** (global_admin holders), and is deduped to the operators who can act on it. * Nothing here fires from a scheduled job, so deploying this cannot batch-notify a backlog of existing rows. Content discipline: the researcher's identity and the reward amount never enter an owner-facing payload. The recipient is a colleague pulled in to answer one technical question; handing them a stranger's email address and a dollar figure widens the report's blast radius for no operational gain. Every send is best-effort: a Feishu/inbox failure must not roll back the assignment that triggered it. The caller gets a bool back and the console reports it honestly rather than claiming "已通知". """ from __future__ import annotations import logging from collections.abc import Sequence from datetime import UTC, datetime from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.auth.models import AuthRole, AuthUserRole from app.domains.bounty import config_service as cfg from app.domains.bounty import constants as const from app.domains.bounty.models import BountyMailIntake, BountyReport from app.identity.person_models import IdentityPerson from app.notifications.service import ( CardAction, CardHighlight, LocalizedText, NotificationEnvelope, NotificationService, ) logger = logging.getLogger(__name__) #: Where an owner-facing push lands. Must be a page that carries the #: confirm CONTROLS, not a read-only detail view — a card whose landing #: page has nothing to click is why a worklist stays at zero rows. OWNER_PAGE = "/my/bounty/{report_id}" ADMIN_PAGE = "/bounty?report={report_id}" #: Cap on the summary excerpt in a card. The full text is on the page. _SUMMARY_CHARS = 400 def _naive(dt: datetime) -> datetime: """Drop tzinfo so a value compares against a ``DateTime(timezone=True)`` column, which round-trips naive on a fresh load.""" return dt.replace(tzinfo=None) if dt.tzinfo is not None else dt def _naive_opt(dt: datetime | None) -> datetime | None: return None if dt is None else _naive(dt) async def _severity_label(db: AsyncSession, key: str | None) -> str: if not key: return "—" tiers = await cfg.tier_map(db) tier = tiers.get(key) return tier.label_zh if tier is not None else key def _summary_block(report: BountyReport) -> str: """The internal Chinese summary, as a quoted block, or nothing. Deliberately falls back to **empty** rather than to the raw English mail: dumping a researcher's mail (with their signature and address in it) into a colleague's Feishu card is exactly the leak this module exists to avoid. """ summary = (report.summary or "").strip() if not summary: return "" excerpt = summary[:_SUMMARY_CHARS] if len(summary) > _SUMMARY_CHARS: excerpt = excerpt.rstrip() + "…" return f"{excerpt}\n\n" async def notify_owner_assigned( db: AsyncSession, report: BountyReport, *, owner: IdentityPerson, reminder: bool = False, ) -> bool: """Ask the named business owner to confirm the finding. Returns whether the notification was recorded. Never raises: the assignment it follows is already committed-worthy on its own. """ severity = await _severity_label( db, report.severity or report.suggested_severity, ) url = OWNER_PAGE.format(report_id=report.id) waiting_days = 0 assigned = _naive_opt(report.assigned_at) if assigned is not None: waiting_days = max( 0, (_naive(datetime.now(UTC)) - assigned).days, ) template = ( "bounty.owner_confirm_reminder" if reminder else "bounty.owner_confirm_requested" ) context = { "report_code": report.code, "report_title": report.title, "affected_system": report.affected_system or "—", "suggested_severity": severity, "summary_block": _summary_block(report), "waiting_days": waiting_days, } env = NotificationEnvelope( recipient=owner, title=( f"请确认安全漏洞报告 {report.code}" if not reminder else f"待确认:安全漏洞报告 {report.code}" ), body_text="", # rendered from the template channels=["feishu", "email"], template_key=template, context=context, url=url, related_resource_type="bounty_report", related_resource_id=report.id, card_subtitle=( "外部研究者报告了一个可能的漏洞,需要你判断是否真实存在。" ), card_highlights=[ CardHighlight(label="涉及系统", items=report.affected_system or "—"), CardHighlight(label="初步级别", items=severity), CardHighlight( label="需要你回答", items=[ "是不是真的问题", "影响面有多大", "预计什么时候能修完", ], ), ], card_actions=[ CardAction(label="打开确认页", url=url, style="primary"), ], ) try: result = await NotificationService(db).dispatch(env) except Exception: # Best effort by design: losing the ping is recoverable (the # operator can 催办), losing the assignment is not. logger.warning( "bounty: owner notification failed for %s", report.code, exc_info=True, ) return False # "Recorded in the outbox" is NOT "the owner was told": ``dispatch`` # persists the row and dead-letters a send that every channel # rejected, without raising. Returning True on that would make the # console show 已通知 for a card nobody received — the exact # dishonesty this flag exists to prevent. Report actual delivery. delivered = bool(result.delivered_channels) if not delivered: logger.warning( "bounty: owner notification for %s reached no channel " "(feishu=%s email=%s)", report.code, result.feishu_error, result.email_error, ) return delivered _DECISION_LABEL = { const.OWNER_DECISION_VALID: "确认是问题", const.OWNER_DECISION_FALSE_POSITIVE: "判定误报", const.OWNER_DECISION_NEED_INFO: "需要更多信息", } #: The DEFAULT alert audience when nobody has been named in the console: #: 安全员. This is the desk — the role exists precisely to triage inbound #: vulnerability reports. #: #: ``global_admin`` used to be in here, so these alerts fanned out to eight #: people in production when four of them do the work. IT holding a global #: admin seat is not a reason to receive every researcher chase, and an #: audience wider than the people who act on it is how a channel gets #: muted. Deliberately NOT ``auditor`` either: read-heavy governance seat, #: this is an operational hand-off. Anyone outside the default — including #: a global admin who does want them — is added by name in 配置. _DESK_ROLES: tuple[str, ...] = ("security_officer",) async def _role_default_operators(db: AsyncSession) -> list[IdentityPerson]: rows = ( await db.execute( select(IdentityPerson) .join(AuthUserRole, AuthUserRole.user_id == IdentityPerson.id) .join(AuthRole, AuthRole.id == AuthUserRole.role_id) .where( AuthRole.name.in_(_DESK_ROLES), IdentityPerson.employment_status == "active", IdentityPerson.deleted_at.is_(None), ), ) ).scalars().all() # A person can hold the role twice through legacy rows. seen: set = set() unique: list[IdentityPerson] = [] for person in rows: if person.id in seen: continue seen.add(person.id) unique.append(person) return unique async def _security_operators(db: AsyncSession) -> list[IdentityPerson]: """Who gets the desk's alerts: the configured list, else the role default. Configurable because the audience is an operational choice, not a property of the RBAC model — the security desk is four people while ``global_admin`` is another four who were being told about every researcher chase. Two guards, both about not silencing the channel by accident: * A configured recipient who has since left (or was soft-deleted) is dropped, exactly like a role holder would be. An offboarded person's row must not absorb an alert. * If the configured list resolves to NOBODY — everyone named has since left — the role default takes over rather than the desk going quiet. "The configured people are all gone" and "nobody should be told" are different states, and only one of them is ever intended. Muting is done by disabling the poll job, which is visible on the jobs page. """ configured = await cfg.get_alert_recipient_ids(db) if configured: rows = ( await db.execute( select(IdentityPerson).where( IdentityPerson.id.in_(configured), IdentityPerson.employment_status == "active", IdentityPerson.deleted_at.is_(None), ), ) ).scalars().all() if rows: # Preserve the configured order so the console's list and the # send order agree; a reader comparing the two should not have # to wonder whether a name was dropped. by_id = {p.id: p for p in rows} ordered = [by_id[pid] for pid in configured if pid in by_id] if len(ordered) < len(configured): logger.info( "bounty: %d configured alert recipient(s) skipped " "(left the company or deleted)", len(configured) - len(ordered), ) return ordered logger.warning( "bounty: every configured alert recipient is inactive — " "falling back to the %s role default so the desk is not silent", "/".join(_DESK_ROLES), ) return await _role_default_operators(db) async def notify_security_of_owner_response( db: AsyncSession, report: BountyReport, *, owner: IdentityPerson | None, ) -> int: """Push the owner's verdict back to the security team (IT-only). Returns how many recipients were notified. Without this the operator has to poll the console to find out that somebody answered. """ severity = await _severity_label(db, report.suggested_severity) note = (report.owner_note or "").strip() note_block = f" · 说明: {note[:300]}\n\n" if note else "" url = ADMIN_PAGE.format(report_id=report.id) owner_name = ( (owner.display_name or owner.work_email) if owner is not None else (report.assignee_name or "业务负责人") ) context = { "report_code": report.code, "owner_name": owner_name, "owner_department": report.assignee_department or "—", "decision_label": _DECISION_LABEL.get( report.owner_decision or "", report.owner_decision or "—", ), "suggested_severity": severity, "fix_eta": report.fix_eta.isoformat() if report.fix_eta else "未给出", "note_block": note_block, } service = NotificationService(db) sent = 0 for operator in await _security_operators(db): env = NotificationEnvelope( recipient=operator, title=f"{owner_name} 已确认 {report.code}", body_text="", channels=["feishu"], template_key="bounty.owner_responded", context=context, url=url, related_resource_type="bounty_report", related_resource_id=report.id, card_actions=[ CardAction(label="打开报告", url=url, style="primary"), ], ) try: # ``push_inbox_and_feishu`` writes the inbox row DELIVERED up # front by design — the operator's bell is the contract here, # and a Feishu hiccup does not lose the item. So a clean # return genuinely means "it is in their inbox". await service.push_inbox_and_feishu(env) sent += 1 except Exception: logger.warning( "bounty: security notification failed for %s → %s", report.code, operator.work_email, exc_info=True, ) return sent #: How many mails one card itemises. ``security@`` also receives vendor #: pitches and automated scan output, so a quiet week and a noisy hour #: produce very different queue sizes; past this the card says how many #: more are waiting rather than growing a list nobody reads to the end. _MAIL_ITEM_LIMIT = 5 #: Per-mail excerpt budget in the card. The full snippet is on the page. _MAIL_EXCERPT_CHARS = 160 #: Field labels for the itemised mail list, per language. The DATA (sender, #: subject, host) is language-independent; only these labels are not, and #: one envelope fans out to a desk that does not share a language. _MAIL_LABELS: dict[str, dict[str, str]] = { "zh": { "subject": "主题", "excerpt": "开头", "ai": "AI 初判", "confidence": "置信度", "unknown_sender": "(来信没有署名)", "no_subject": "(无主题)", "more": "还有 {n} 封在队列里", "thread_seen": "同一 thread 之前也来过信 —— 可能是跟进,不是新报告", "attached": "本轮另有 {n} 封挂到了已有报告({codes}),催办卡片另发", }, "en": { "subject": "Subject", "excerpt": "Opens with", "ai": "AI first read", "confidence": "confidence", "unknown_sender": "(unsigned)", "no_subject": "(no subject)", "more": "{n} more waiting in the queue", "thread_seen": ( "this thread has written before — likely a follow-up, not a new " "report" ), "attached": ( "{n} other mail(s) this round attached to existing reports " "({codes}); those get their own follow-up card" ), }, "ja": { "subject": "件名", "excerpt": "冒頭", "ai": "AI 初期判断", "confidence": "信頼度", "unknown_sender": "(署名なし)", "no_subject": "(件名なし)", "more": "残り {n} 件がキューにあります", "thread_seen": ( "同じスレッドから以前も受信 —— 新規報告ではなく追記の可能性" ), "attached": ( "今回はほかに {n} 件が既存の報告に紐付きました({codes})。" "そちらは別途カードを送ります" ), }, } def _mail_sender(row: Any, labels: dict[str, str]) -> str: """``Name ``, or just the address, or a marker for neither. The researcher's address DOES belong on this card: the recipients are the security desk, who answer these mails and already read the address in the console. The rule the module docstring states — no researcher identity — is about the OWNER-facing cards, where the reader is a colleague pulled in to answer one technical question. """ email = (row.from_email or "").strip() name = (row.from_name or "").strip() if name and email: return f"{name} <{email}>" return email or labels["unknown_sender"] def _mail_ai_line(row: Any, labels: dict[str, str]) -> str: """The AI's reading of subject + snippet, or nothing. Nothing rather than a placeholder: ``suggest_from_titles`` returns ``{}`` when no provider is configured, and a card line reading "AI 初判: —" on every mail teaches the reader to skip the whole block. """ parsed = row.parsed if isinstance(row.parsed, dict) else None if not parsed: return "" bits = [str(parsed[k]) for k in ("severity", "host") if parsed.get(k)] if not bits: return "" line = " · ".join(bits) confidence = parsed.get("confidence") if confidence: line = f"{line}({labels['confidence']} {confidence})" return f" {labels['ai']}: {line}\n" def _mail_items_block( rows: Sequence[Any], labels: dict[str, str], *, repeat_threads: set[str], ) -> str: """The itemised list that makes the card actionable. 「你觉得谁能看的懂这个推送?」 — the card used to carry a count and nothing else, so the desk had to open the console to learn whether the poll had found a real report or a vendor pitch. Sender, subject and the AI's first read are what decide whether somebody stops what they are doing, and they are all already on the row. """ lines: list[str] = [] for index, row in enumerate(rows[:_MAIL_ITEM_LIMIT], start=1): subject = (row.subject or "").strip() or labels["no_subject"] received = _naive(_aware_dt(row.received_at)).strftime("%m-%d %H:%M") lines.append( f" [b]{index}. {_mail_sender(row, labels)}[/b] · {received}\n" f" {labels['subject']}: {subject[:120]}\n" ) lines.append(_mail_ai_line(row, labels)) excerpt = (row.snippet or "").strip().replace("\n", " ") if excerpt: trimmed = excerpt[:_MAIL_EXCERPT_CHARS] if len(excerpt) > _MAIL_EXCERPT_CHARS: trimmed = trimmed.rstrip() + "…" lines.append(f" {labels['excerpt']}: {trimmed}\n") if row.thread_id and row.thread_id in repeat_threads: lines.append(f" [r]{labels['thread_seen']}[/r]\n") lines.append("\n") remaining = len(rows) - _MAIL_ITEM_LIMIT if remaining > 0: lines.append(f" {labels['more'].format(n=remaining)}\n\n") return "".join(lines) def _aware_dt(value: datetime | None) -> datetime: """``received_at`` is NOT NULL in the schema, but this renders a card — a malformed row must not be the reason nobody hears about a report.""" if value is None: return datetime.now(UTC) return value if value.tzinfo else value.replace(tzinfo=UTC) async def _repeat_threads(db: AsyncSession, rows: Sequence[Any]) -> set[str]: """Thread ids among ``rows`` that Helix has seen mail on before. A mail whose thread already produced a REPORT never reaches this card — ``poll`` attaches it as a follow-up instead. What is left is the middle case: a thread whose earlier mail is still sitting in the queue undecided. Flagging it stops the desk from opening two reports for one conversation, which is the specific way this ledger gets corrupted. """ thread_ids = {r.thread_id for r in rows if r.thread_id} if not thread_ids: return set() new_ids = {r.id for r in rows} seen = ( await db.execute( select(BountyMailIntake.thread_id).where( BountyMailIntake.thread_id.in_(thread_ids), BountyMailIntake.id.notin_(new_ids), ), ) ).scalars().all() return {t for t in seen if t} async def notify_security_of_new_mail( db: AsyncSession, *, rows: Sequence[Any] = (), attached_codes: Sequence[str] = (), ) -> int: """Tell the desk WHAT arrived on ``security@``, not just how many. 118 of 149 reports went unanswered, and the mechanical reason was that the work only happened when somebody remembered to open the mailbox. The queue removed the copy-paste; this removes the remembering. One notification per POLL, not per mail: fifteen-minute polling means a per-mail push would train the desk to mute the channel. But one per poll is only a reason to BATCH the content, not to omit it — the first version sent a bare count, which told the reader nothing they could act on and made the card indistinguishable from a vendor pitch arriving. """ rows = list(rows) if not rows: return 0 queued = len(rows) repeat = await _repeat_threads(db, rows) first = rows[0] def _labels(lang: str) -> dict[str, str]: return _MAIL_LABELS[lang] # The card TITLE is what shows in a Feishu chat list, so for the common # single-mail poll it carries the sender and subject rather than a # number the reader has to open the card to interpret. if queued == 1: subject = (first.subject or "").strip() headline_zh = ( f"{_mail_sender(first, _labels('zh'))}" + (f" · {subject[:60]}" if subject else "") ) headline = LocalizedText( zh=headline_zh, en=( f"{_mail_sender(first, _labels('en'))}" + (f" · {subject[:60]}" if subject else "") ), ja=( f"{_mail_sender(first, _labels('ja'))}" + (f" · {subject[:60]}" if subject else "") ), ) title = f"白帽子来信:{headline_zh}"[:180] else: headline = LocalizedText( zh=f"{queued} 封待确认", en=f"{queued} waiting to be confirmed", ja=f"{queued} 件が確認待ち", ) title = f"security@ 有 {queued} 封新的白帽子来信待确认" items_block = LocalizedText( zh=_mail_items_block(rows, _labels("zh"), repeat_threads=repeat), en=_mail_items_block(rows, _labels("en"), repeat_threads=repeat), ja=_mail_items_block(rows, _labels("ja"), repeat_threads=repeat), ) attached_codes = [c for c in attached_codes if c][:3] if attached_codes: codes = "、".join(attached_codes) attached_block: LocalizedText | str = LocalizedText( zh=" " + _labels("zh")["attached"].format( n=len(attached_codes), codes=codes) + "\n\n", en=" " + _labels("en")["attached"].format( n=len(attached_codes), codes=", ".join(attached_codes)) + "\n\n", ja=" " + _labels("ja")["attached"].format( n=len(attached_codes), codes=codes) + "\n\n", ) else: attached_block = "" service = NotificationService(db) # ?intake=1 opens the confirmation queue itself. Landing on /bounty # left the queue behind one more click in the page header — a card whose # button says 「去确认」 has to arrive AT the thing it names. url = "/bounty?intake=1" sent = 0 for operator in await _security_operators(db): env = NotificationEnvelope( recipient=operator, title=title, body_text="", channels=["feishu"], template_key="bounty.mail_queued", context={ "queued": str(queued), "headline": headline, "items_block": items_block, "attached_block": attached_block, }, url=url, related_resource_type="bounty_mail_intake", related_resource_id=None, card_actions=[ CardAction( label=LocalizedText( zh="去确认", en="Confirm now", ja="確認する", ), url=url, style="primary", ), ], ) try: await service.push_inbox_and_feishu(env) sent += 1 except Exception: logger.warning( "bounty: new-mail notification failed for %s", operator.work_email, exc_info=True, ) return sent async def notify_security_of_researcher_followup( db: AsyncSession, report: Any, *, days_silent: int | None = None, excerpt: str | None = None, ) -> int: """Tell the desk a researcher has come back to us on an open report. A follow-up is almost always a chase, and a chase is the programme's credibility being spent — the oldest report here waited 295 days. It is also the one event where the researcher is actively waiting, so it is worth interrupting somebody for. """ service = NotificationService(db) url = f"/bounty/reports/{report.code}" waited = ( f",已静默 {days_silent} 天" if days_silent and days_silent > 0 else "" ) # The silence line is a whole BLOCK rather than a bare ``{days_silent}`` # placeholder: the poller passes ``None`` whenever we have replied at # least once, and a template that interpolated the number directly would # render "已静默: 天" on exactly those rows. Localized because one # envelope fans out to the whole desk and they do not share a language — # a Chinese fragment spliced into an English card is the same defect as # a Chinese-only template. silence_block: LocalizedText | str = "" if days_silent and days_silent > 0: silence_block = LocalizedText( zh=f" · 已静默: {days_silent} 天\n", en=f" · Silent for: {days_silent} day(s)\n", ja=f" · 沈黙期間: {days_silent} 日\n", ) # What they actually wrote this time. 「不给说出重要信息 就是一个提醒?」 # — a chase card that names only the report makes the reader open the # console to find out whether the researcher answered a question, sent # a PoC or asked about the reward. The desk is the recipient here, so # the mail's own opening line is theirs to read. excerpt_block: LocalizedText | str = "" text = (excerpt or "").strip().replace("\n", " ") if text: trimmed = text[:_MAIL_EXCERPT_CHARS] if len(text) > _MAIL_EXCERPT_CHARS: trimmed = trimmed.rstrip() + "…" excerpt_block = LocalizedText( zh=f" · 这次说了什么: {trimmed}\n", en=f" · This message: {trimmed}\n", ja=f" · 今回の内容: {trimmed}\n", ) sent = 0 for operator in await _security_operators(db): env = NotificationEnvelope( recipient=operator, title=f"白帽子催办 {report.code}{waited}", body_text="", channels=["feishu"], template_key="bounty.researcher_followup", context={ "code": report.code, "title": (report.title or "—")[:120], "days_silent": str(days_silent or ""), "silence_block": silence_block, "excerpt_block": excerpt_block, }, url=url, related_resource_type="bounty_report", related_resource_id=report.id, card_actions=[ CardAction( label=LocalizedText( zh="打开报告", en="Open the report", ja="報告を開く", ), url=url, style="primary", ), ], ) try: await service.push_inbox_and_feishu(env) sent += 1 except Exception: logger.warning( "bounty: follow-up notification failed for %s", operator.work_email, exc_info=True, ) return sent