"""Pull what arrived on ``security@`` into a queue a person confirms. Every white-hat report is pasted in by hand today. That is the mechanical reason 118 of 149 never got an answer and the oldest waited 295 days: the work only happens when somebody opens the mailbox. Three decisions worth stating, because each one is a place this could have gone wrong: **It never creates a report by itself.** ``security@`` receives vendor pitches, internal mail and automated scan output. A poll writes rows into ``bounty_mail_intake`` and a human accepts or ignores each one. The precedent is concrete: an HRIS predicate slightly too loose created 26 accounts for system users because nothing asked a person first. **Follow-ups attach, they do not multiply.** Researchers chase their own threads — with 118 unanswered reports, a lot of chasing. A mail whose ``thread_id`` already produced a report is logged onto that report as inbound correspondence instead of becoming a second report. **The body is not stored.** Only a snippet. Researchers send tax forms and bank details unprompted, and the FY27 import already established that their correspondence stays in the mailbox. Requires ``gmail.readonly`` on the Workspace service account, which is a Google admin action. Until it is granted the poller reports a clear, actionable failure rather than a stack trace, and its scheduled job ships disabled so the failure notifier is not fired every fifteen minutes. """ from __future__ import annotations import asyncio import base64 import hashlib import html import logging import uuid from datetime import UTC, datetime from email.utils import parseaddr, parsedate_to_datetime from typing import Any from sqlalchemy import or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import AppException, ValidationError from app.domains.bounty import ai as bounty_ai from app.domains.bounty import config_service as cfg from app.domains.bounty import constants as const from app.domains.bounty import service as svc from app.domains.bounty.models import BountyMailIntake, BountyReport from app.identity.person_models import IdentityPerson logger = logging.getLogger(__name__) GMAIL_READ_SCOPE = "https://www.googleapis.com/auth/gmail.readonly" #: Addresses that are us. A thread that arrived by forward has one of these #: in ``From``, so attributing the report to it would file a researcher's #: finding against the security team itself. INTERNAL_SUFFIXES: tuple[str, ...] = ("@pingcap.com",) #: Default mailbox. Overridable per job config so a dedicated #: ``bounty-intake@`` forwarding address can be used instead of granting #: read access to the whole of ``security@``. DEFAULT_MAILBOX = "bounty-intake@pingcap.com" #: ⚠️ ``security@pingcap.com`` is a GROUP, not a mailbox. #: #: Domain-wide delegation can only impersonate a real Workspace USER. #: Verified 2026-08-26 against production: ``with_subject("security@…")`` #: fails with ``unauthorized_client`` for *every* scope — including #: ``gmail.send``, which Helix uses successfully all day — while the same #: credential impersonates real users fine. So a poller pointed at the #: group address can never work, no matter which scopes are granted. #: #: Two allowlists, because the two operations are not equally invasive. #: #: LISTING an inbox is broad: it walks whatever the query matches. That may #: only ever be a dedicated intake address whose entire purpose is receiving #: these reports — never a person's mailbox. #: #: FETCHING one thread by an id Helix already recorded is narrow: the id is #: on the report's own ``created`` event, so the read is bounded to #: correspondence the bounty ledger already knows about. That is what makes #: it acceptable against the archive mailbox below, which is where the #: security@ group mail actually lands. #: #: Both are code constants, not settings: ``mailbox`` reaches this module #: from a console-editable cron config, and unconstrained that would turn #: "can edit a cron config" into "can read any employee's mail". ALLOWED_POLL_MAILBOXES: frozenset[str] = frozenset({ # Does not exist yet — create it as a real USER (not a group) and # subscribe it to security@ to turn the poller on. See # docs/ops/gmail-readonly-delegation.md. "bounty-intake@pingcap.com", }) #: Mailboxes a recorded thread id may be fetched from. The platform owner #: is a member of the security@ group, so their mailbox is where the FY27 #: threads physically are — and Gmail thread ids are per-account, so it is #: the ONLY account in which those 149 ids resolve at all. ALLOWED_FETCH_MAILBOXES: frozenset[str] = ALLOWED_POLL_MAILBOXES | { "shuai.feng@pingcap.com", } #: Kept for the poller's default. Deliberately an address that does not #: exist yet rather than the group: a job that cannot work should fail #: saying so, not appear configured. ALLOWED_MAILBOXES = ALLOWED_POLL_MAILBOXES #: Where a recorded thread id is fetched from by default. The FY27 ids only #: resolve here, because Gmail thread ids are per-account. ARCHIVE_MAILBOX = "shuai.feng@pingcap.com" _MAX_PER_POLL = 50 def _looks_internal(email: str) -> bool: lowered = (email or "").lower() return any(lowered.endswith(suffix) for suffix in INTERNAL_SUFFIXES) def _header(payload: dict[str, Any], name: str) -> str | None: """One header value, as a string or nothing. The Google client hands back untyped dicts, so the value is narrowed here rather than at each call site — a non-string sneaking through would end up in a ``String(512)`` column. """ for header in (payload.get("headers") or []): if (header.get("name") or "").lower() == name.lower(): value = header.get("value") return str(value) if value is not None else None return None def _decode_snippet(raw: dict[str, Any]) -> str | None: """Gmail's own snippet, or the first text part, truncated. Truncated on purpose: this is a preview to help a human decide, not a copy of the correspondence. """ # Gmail returns ``snippet`` HTML-ESCAPED. Unescaped, a researcher # writing "it's still reproducible" reaches the desk's Feishu card and # the report's own timeline as "it's still reproducible" — seen on # the first real chase the live poller ingested (2026-08-28, # BB-2026-144). Cosmetic in the database, not cosmetic in a card # somebody is meant to read and act on. snippet = html.unescape(raw.get("snippet") or "").strip() if snippet: return snippet[:2000] payload = raw.get("payload") or {} for part in [payload, *(payload.get("parts") or [])]: if (part.get("mimeType") or "") != "text/plain": continue data = ((part.get("body") or {}).get("data") or "") if not data: continue try: text = base64.urlsafe_b64decode(data + "===").decode( "utf-8", errors="replace", ) except Exception: # noqa: BLE001 - a malformed part is not fatal continue return text.strip()[:2000] return None async def _service(db: AsyncSession, mailbox: str): """A read-only Gmail client impersonating ``mailbox``. Reuses the existing email integration's service-account key — the same credential that already sends mail as a Workspace user — rather than introducing a second secret. What it does NOT reuse is the scope: this asks for ``gmail.readonly``, which the Workspace admin has to grant in domain-wide delegation. Until they do, Google answers with a permission error, so the message here names the remediation: the credential is fine and the *scope grant* is what is missing, and "insufficient permission" on its own sends the reader to the wrong place. """ from app.integrations.service import build_primary_driver # noqa: PLC0415 try: driver = await build_primary_driver("email", session=db) except AppException as exc: raise AppException( code="BOUNTY_MAIL_NO_CREDENTIAL", message=( "没有可用的邮件集成,轮询 security@ 需要它的服务账号密钥。" f"原始原因:{exc.message}" ), status_code=409, ) from exc sa_json = (driver.config or {}).get("service_account_json") if not sa_json: raise AppException( code="BOUNTY_MAIL_NO_CREDENTIAL", message=( "邮件集成没有配置 service_account_json —— 当前的驱动可能是 " "SMTP。轮询 security@ 需要 Google 服务账号(域级委派)。" ), status_code=409, ) def _build(): import json from google.oauth2 import service_account # noqa: PLC0415 from googleapiclient.discovery import build # noqa: PLC0415 creds = service_account.Credentials.from_service_account_info( json.loads(str(sa_json)), scopes=[GMAIL_READ_SCOPE], ).with_subject(mailbox) return build("gmail", "v1", credentials=creds, cache_discovery=False) try: return await asyncio.to_thread(_build) except Exception as exc: # noqa: BLE001 raise AppException( code="BOUNTY_MAIL_SCOPE_MISSING", message=( "无法以只读方式打开 security@。最常见的原因是 Workspace " "域级委派里还没有给这个服务账号加 " f"{GMAIL_READ_SCOPE} —— 那是 Google 管理后台的动作," f"代码改不了。原始错误:{exc}" ), status_code=502, ) from exc async def poll( db: AsyncSession, *, config: dict[str, Any] | None = None, ) -> dict[str, Any]: """Fetch new mail into the confirmation queue. Never creates reports.""" config = config or {} mailbox = str(config.get("mailbox") or DEFAULT_MAILBOX).strip().lower() if mailbox not in ALLOWED_POLL_MAILBOXES: raise AppException( code="BOUNTY_MAIL_MAILBOX_NOT_ALLOWED", message=( f"不允许轮询 {mailbox}。域级委派可以冒充任意 Workspace 用户," "所以可读的邮箱是代码里的白名单,不是任务配置里的自由文本 —— " "要新增请改 ALLOWED_POLL_MAILBOXES 并走代码评审。当前允许:" + "、".join(sorted(ALLOWED_POLL_MAILBOXES)) + "。注意 security@pingcap.com 是**邮件组**,域级委派只能冒充" "真实用户,指向它永远会 unauthorized_client。" ), status_code=422, ) query = config.get("query") or "in:inbox newer_than:30d" limit = int(config.get("limit") or _MAX_PER_POLL) service = await _service(db, mailbox) def _list() -> list[dict[str, Any]]: result = service.users().messages().list( userId="me", q=query, maxResults=limit, ).execute() return result.get("messages") or [] listed = await asyncio.to_thread(_list) # Skip ids we have already offered before spending a `get` on each — # the poller runs every 15 minutes over a 30-day window, so almost # everything it lists is already known. gmail_ids = [m["id"] for m in listed if m.get("id")] known_threads: set[str] = set() seen_message_ids: set[str] = set() if gmail_ids: rows = ( await db.execute( select( BountyMailIntake.message_id, BountyMailIntake.thread_id, ), ) ).all() seen_message_ids = {r[0] for r in rows} known_threads = {r[1] for r in rows if r[1]} created = 0 attached = 0 skipped = 0 internal = 0 # The rows themselves, not just a count: the desk's card names the # sender and the subject of each, which is what makes it decidable # without opening the console. queued_rows: list[BountyMailIntake] = [] attached_codes: list[str] = [] for gmail_id in gmail_ids: def _get(mid: str = gmail_id) -> dict[str, Any]: fetched = service.users().messages().get( userId="me", id=mid, format="metadata", metadataHeaders=["From", "Subject", "Date", "Message-ID"], ).execute() # The client is untyped; a malformed response becomes an empty # dict here rather than an AttributeError three lines later. return dict(fetched) if isinstance(fetched, dict) else {} raw = await asyncio.to_thread(_get) payload = raw.get("payload") or {} message_id = _header(payload, "Message-ID") or f"gmail:{gmail_id}" if message_id in seen_message_ids: skipped += 1 continue from_name, from_email = parseaddr(_header(payload, "From") or "") if not from_email: skipped += 1 continue if _looks_internal(from_email): # Attributing a forwarded report to our own alias is worse than # not ingesting it: the ledger would show the security team # reporting to itself. internal += 1 continue date_header = _header(payload, "Date") try: received = ( parsedate_to_datetime(date_header) if date_header else datetime.now(UTC) ) except (TypeError, ValueError): received = datetime.now(UTC) if received.tzinfo is None: received = received.replace(tzinfo=UTC) thread_id = raw.get("threadId") subject = (_header(payload, "Subject") or "").strip() or None # A follow-up on a thread that already produced a report belongs on # that report, not in the queue as a new finding. existing_report_id: uuid.UUID | None = None if thread_id and thread_id in known_threads: existing = ( await db.execute( select(BountyMailIntake).where( BountyMailIntake.thread_id == thread_id, BountyMailIntake.report_id.is_not(None), ).limit(1), ) ).scalars().first() existing_report_id = existing.report_id if existing else None row = BountyMailIntake( message_id=message_id[:512], thread_id=thread_id, from_email=from_email.lower()[:320], from_name=(from_name or None) and from_name[:255], subject=subject[:500] if subject else None, snippet=_decode_snippet(raw), received_at=received, status="pending", ) db.add(row) await db.flush() seen_message_ids.add(message_id) if thread_id: known_threads.add(thread_id) if existing_report_id is not None and await attach_followup( db, row, existing_report_id, ): attached += 1 report = await db.get(BountyReport, existing_report_id) if report is not None: attached_codes.append(report.code) continue created += 1 queued_rows.append(row) # One AI call for everything new, on subject + snippet only. await _annotate_pending(db) if queued_rows: # One push per poll, not per mail — see notify_security_of_new_mail. # Failure to notify must not undo a successful poll: the rows are # already queued and visible in the console either way. try: from app.domains.bounty import notify as bnotify # noqa: PLC0415 # After ``_annotate_pending`` on purpose: the AI's first read of # each subject is part of what the card carries. await bnotify.notify_security_of_new_mail( db, rows=queued_rows, attached_codes=attached_codes, ) except Exception: # noqa: BLE001 logger.warning("bounty.mail: queue notification failed", exc_info=True) return { "listed": len(gmail_ids), "queued": created, "attached_to_existing": attached, "already_seen": skipped, "internal_sender_skipped": internal, "mailbox": mailbox, } def _aware(value: datetime) -> datetime: """Naive timestamps come out of MySQL; comparing one to an aware `now` raises. Mixed naive/aware arithmetic is a recurring bug class here.""" return value if value.tzinfo else value.replace(tzinfo=UTC) async def attach_followup( db: AsyncSession, row: BountyMailIntake, report_id: uuid.UUID, ) -> bool: """Log a follow-up onto the report its thread already created. Researchers chase their own threads, and 118 unanswered reports means a lot of chasing. Without this every chase becomes a second report for the same finding, which inflates the ledger and splits the conversation across two timelines. Module level rather than inline in ``poll`` so it is reachable without a Gmail client — the alternative was a test that asserted its own fixture. """ report = await db.get(BountyReport, report_id) if report is None: return False await svc.add_event( db, report=report, event_type=const.EVENT_INBOUND_LOGGED, actor=None, actor_kind="system", body=(row.snippet or "")[:4000] or None, meta={"via": "mail_intake", "message_id": row.message_id}, visibility="internal", ) row.status = "attached" row.report_id = report.id row.decided_at = datetime.now(UTC) await db.flush() # A follow-up on an open report is almost always a chase, and a chase # is the one event where the researcher is actively waiting on us. try: from app.domains.bounty import notify as bnotify # noqa: PLC0415 days = None if report.last_replied_at is None and report.reported_at is not None: days = (datetime.now(UTC) - _aware(report.reported_at)).days await bnotify.notify_security_of_researcher_followup( db, report, days_silent=days, excerpt=row.snippet, ) except Exception: # noqa: BLE001 — never lose the attachment over a push logger.warning("bounty.mail: follow-up notification failed", exc_info=True) return True async def _annotate_pending(db: AsyncSession, limit: int = 30) -> None: """Attach the AI's reading to queue rows that do not have one yet.""" rows = ( await db.execute( select(BountyMailIntake).where( BountyMailIntake.status == "pending", BountyMailIntake.parsed.is_(None), ).order_by(BountyMailIntake.received_at.asc()).limit(limit), ) ).scalars().all() if not rows: return tiers = await cfg.tier_map(db) # A SHORT key, not `str(row.id)`. `suggest_from_titles` narrows the code # the model echoes back with `_clean_str(..., limit=32)`, and a UUID # renders as 36 characters — so every returned key was a 32-char prefix # of the id we looked up, every `.get()` missed, and `parsed` stayed NULL # for every mail ever polled. Silent by construction: the AI call # succeeded, the model answered correctly, and nothing logged. Found # 2026-08-28 on the first real mail the live poller queued, by noticing # the card had no AI line. The limit is a guard against a model # returning prose where a code belongs and stays where it is; the # caller's job is to ask with a key that survives it. by_key = {row.id.hex[:16]: row for row in rows} suggestions = await bounty_ai.suggest_from_titles( db, items=[ { "code": key, "title": " · ".join( part for part in (row.subject, (row.snippet or "")[:200]) if part ), } for key, row in by_key.items() ], severity_keys=[k for k, tier in tiers.items() if tier.enabled], known_hosts=[], ) for key, row in by_key.items(): hint = suggestions.get(key) if hint: row.parsed = hint await db.flush() async def _claim_pending( db: AsyncSession, intake_id: Any, ) -> BountyMailIntake: """Lock one queued mail and assert it is still undecided. ``db.get()`` + ``if row.status != "pending"`` is a read-then-write race: a double-clicked 「接受」, or two operators working the queue at once, both read ``pending`` and both create a report for the same mail. In this ledger a duplicate is not cosmetic — every count, the budget attribution, and the reward owed are computed off these rows, and a duplicated finding can be paid twice. ``FOR UPDATE`` serialises the two callers; ``populate_existing=True`` is what makes the lock mean anything, because the session is ``expire_on_commit=False`` and would otherwise hand back the identity map's stale ``pending`` copy of the row it just locked. """ row = await db.scalar( select(BountyMailIntake) .where(BountyMailIntake.id == intake_id) .with_for_update() .execution_options(populate_existing=True), ) if row is None: raise ValidationError("mail intake row not found") if row.status != "pending": raise AppException( code="BOUNTY_MAIL_ALREADY_DECIDED", message=f"这封信已经处理过了(当前状态:{row.status})。", status_code=409, ) return row async def accept( db: AsyncSession, intake_id: Any, *, actor: IdentityPerson | None, title: str | None = None, affected_system: str | None = None, ) -> BountyReport: """Turn one queued mail into a report. The human's decision, not the poller's.""" row = await _claim_pending(db, intake_id) report = await svc.create_report( db, title=(title or row.subject or "(no subject)")[:200], researcher_email=row.from_email, researcher_name=row.from_name, reported_at=row.received_at, affected_system=affected_system, source="email", summary=None, actor=actor, ) # The AI's suggested severity is recorded as a SUGGESTION, never # applied: grading pulls a reward tier and a fix deadline, which is a # decision with money attached. suggested = (row.parsed or {}).get("severity") if suggested: report.suggested_severity = suggested row.status = "accepted" row.report_id = report.id row.decided_by = actor.id if actor else None row.decided_at = datetime.now(UTC) await db.flush() return report async def ignore( db: AsyncSession, intake_id: Any, *, actor: IdentityPerson | None, note: str | None = None, ) -> BountyMailIntake: """Not a researcher report. The row stays so the same mail is never offered again.""" row = await _claim_pending(db, intake_id) row.status = "ignored" row.note = (note or None) and note[:1000] row.decided_by = actor.id if actor else None row.decided_at = datetime.now(UTC) await db.flush() return row async def pending( db: AsyncSession, *, limit: int = 50, ) -> list[BountyMailIntake]: """Oldest first — the same ordering as everything else here, because the oldest is where the programme's credibility is being spent.""" return list( ( await db.execute( select(BountyMailIntake).where( BountyMailIntake.status == "pending", ).order_by(BountyMailIntake.received_at.asc()).limit(limit), ) ).scalars().all() ) #: Patterns that mean "this line is payment plumbing, not a vulnerability". #: The FY27 import kept bodies out of the database precisely because #: researchers attach these unprompted; importing bodies is only acceptable #: with them stripped on the way in. _PAYMENT_PATTERNS: tuple[tuple[str, str], ...] = ( (r"\b[A-Z]{2}\d{2}[A-Z0-9]{10,30}\b", "[IBAN 已移除]"), (r"\b[A-Z]{4}[A-Z]{2}[A-Z0-9]{2}(?:[A-Z0-9]{3})?\b(?=\s*(?:swift|bic))", "[SWIFT 已移除]"), (r"(?i)\b(?:account|acct|a/c|账号|账户|卡号)\s*(?:number|no\.?|#|:)?\s*" r"[:：]?\s*[0-9][0-9 \-]{7,25}", "[账号已移除]"), (r"(?i)\b(?:swift|bic|ifsc|routing|sort\s*code|upi\s*id)\s*[:：]\s*\S+", "[银行信息已移除]"), (r"\b(?:\d[ -]?){13,19}\b", "[卡号已移除]"), ) def redact_payment_details(text: str) -> str: """Strip banking details out of a researcher's message. Not a security control — it is a data-minimisation one. The reason the FY27 import stored no bodies is that researchers put IBANs, card numbers and tax IDs in these threads without being asked, and a bounty ledger that never held them cannot leak them. Importing the technical write-up is worth doing; importing somebody's bank account with it is not. Deliberately over-eager: a false positive costs a redacted string in a vulnerability description that is still readable in the original mail (one click away, on every report). A false negative puts a stranger's account number in our database permanently. """ import re # noqa: PLC0415 out = text or "" for pattern, replacement in _PAYMENT_PATTERNS: out = re.sub(pattern, replacement, out) return out def researcher_first_message( messages: list[dict[str, Any]], researcher_email: str, ) -> str | None: """The message that IS the vulnerability report. The first inbound message from the researcher, which is the write-up. Later messages are the conversation — chasing, negotiation, and the payment details that follow a reward being agreed — so taking only the first is what keeps this import to the technical content by construction, before the redaction pass has to catch anything. """ wanted = (researcher_email or "").strip().lower() def _when(message: dict[str, Any]) -> datetime: """Sort key. The Date header is RFC2822 (``Tue, 28 Jul 2026 …``), and sorting those as STRINGS orders them by weekday name — Fri < Mon < Sat < Thu < Tue — so "the first message" came out effectively random. Caught after the first import batch put a *follow-up* in one report and OUR OWN REPLY in another, both presented as the researcher's finding. """ raw = str(message.get("date") or "") try: parsed = parsedate_to_datetime(raw) except (TypeError, ValueError): return datetime.max.replace(tzinfo=UTC) if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=UTC) return parsed ordered = sorted(messages or [], key=_when) # OUR OWN messages are never the report. This has to hold in both # passes: when the address on file was missing, ``wanted`` was empty, # the guard below short-circuited, and the loop happily returned a # reply written by the security team — which then rendered on the page # as the researcher's finding. Worse than the empty page it replaced. for message in ordered: sender = str(message.get("sender") or "").strip().lower() if _looks_internal(sender): continue if wanted and wanted not in sender: continue body = str(message.get("plaintextBody") or "").strip() if body: return body # Nothing from the address on file — the thread may have arrived by # forward, or the researcher wrote from a second address. Take the # earliest message that is still not one of ours. for message in ordered: sender = str(message.get("sender") or "").strip().lower() if _looks_internal(sender): continue body = str(message.get("plaintextBody") or "").strip() if body: return body return None def _flatten_message(message: dict[str, Any]) -> dict[str, Any]: """Gmail's ``format=full`` shape reduced to what the extractor needs. The body is base64url in ``payload.body.data`` for a simple message and buried in ``payload.parts`` for a multipart one — the common case here, because researchers attach screenshots. Prefers ``text/plain``; falls back to the first part that has any data at all rather than returning nothing for an HTML-only message. """ payload = message.get("payload") or {} def _decode(node: dict[str, Any]) -> str: data = ((node.get("body") or {}).get("data") or "") if not data: return "" try: return base64.urlsafe_b64decode(data + "==").decode( "utf-8", errors="replace", ) except Exception: # noqa: BLE001 — a bad part must not lose the mail return "" def _walk(node: dict[str, Any], want: str | None) -> str: mime = str(node.get("mimeType") or "") if (want is None or mime == want) and not node.get("parts"): return _decode(node) for part in node.get("parts") or []: found = _walk(part, want) if found.strip(): return found return "" body = _walk(payload, "text/plain") or _walk(payload, None) return { "sender": parseaddr(_header(payload, "From") or "")[1], "date": _header(payload, "Date") or "", "plaintextBody": body, } async def _search_researcher_message(service: Any, email: str) -> str | None: """The researcher's earliest message to us, found by sender. Used only when the thread id on the report turns out to hold nothing from them. The query is pinned to one external address that the bounty ledger already records as a submitter — never a subject or free-text search, which would walk mail unrelated to any report. """ safe = (email or "").strip().lower() if not safe or "@" not in safe or any(ch in safe for ch in ' "\n'): return None def _find() -> list[dict[str, Any]]: listed = service.users().messages().list( userId="me", q=f"from:{safe}", maxResults=10, ).execute() out = [] for stub in (listed.get("messages") or []): full = service.users().messages().get( userId="me", id=stub["id"], format="full", ).execute() out.append(dict(full) if isinstance(full, dict) else {}) return out try: fetched = await asyncio.to_thread(_find) except Exception: # noqa: BLE001 — one bad search must not stop a batch logger.warning("bounty.mail: sender search failed", exc_info=True) return None messages = [_flatten_message(m) for m in fetched] # Same selection rules: earliest by real time, never one of ours. return researcher_first_message(messages, safe) #: Written on a report whose write-up could not be found, so the batch does #: not re-attempt it ahead of reports never tried at all. IMPORT_NOT_FOUND = "no_write_up_found" def _exists_import_attempt() -> Any: """A prior "looked and found nothing" marker on this report.""" from sqlalchemy import exists # noqa: PLC0415 from app.domains.bounty.models import ( # noqa: PLC0415 BountyReport, BountyReportEvent, ) return exists().where( BountyReportEvent.report_id == BountyReport.id, BountyReportEvent.event_type == const.EVENT_UPDATED, BountyReportEvent.meta["import"].as_string() == IMPORT_NOT_FOUND, ) async def import_body_for_report( db: AsyncSession, report: Any, *, mailbox: str = ARCHIVE_MAILBOX, ) -> bool: """Pull one report's technical write-up out of its mail thread. The thread id was stamped on the ``created`` event by the FY27 import and by this queue, and nothing had ever read it — so for 149 reports the console knew exactly where the write-up was and still showed a page with no vulnerability on it. Stores only the researcher's FIRST message, redacted. Returns False and changes nothing when there is no thread, no readable message, or the report already has a body a person could be relying on. """ from sqlalchemy import select as _select # noqa: PLC0415 from app.domains.bounty.models import BountyReportEvent # noqa: PLC0415 from app.domains.bounty.serializers import ( # noqa: PLC0415 mail_thread_url_from_events, ) if (report.raw_report or "").strip(): return False if mailbox not in ALLOWED_FETCH_MAILBOXES: raise AppException( code="BOUNTY_MAIL_MAILBOX_NOT_ALLOWED", message=( f"不允许从 {mailbox} 取邮件正文。可取的邮箱是代码里的白名单:" + "、".join(sorted(ALLOWED_FETCH_MAILBOXES)) ), status_code=422, ) events = ( await db.execute( _select(BountyReportEvent).where( BountyReportEvent.report_id == report.id, BountyReportEvent.event_type == const.EVENT_CREATED, ), ) ).scalars().all() # No recorded thread is NOT a dead end. The FY27 import left a `sav-` # placeholder on some reports, and the sender search below can still # find the write-up — returning early here left reports empty whose # researcher had five messages sitting in the mailbox. url = mail_thread_url_from_events(events) service = await _service(db, mailbox) messages: list[dict[str, Any]] = [] if url: thread_id = url.rsplit("/", 1)[-1] def _fetch() -> dict[str, Any]: fetched = service.users().threads().get( userId="me", id=thread_id, format="full", ).execute() return dict(fetched) if isinstance(fetched, dict) else {} raw = await asyncio.to_thread(_fetch) messages = [ _flatten_message(m) for m in (raw.get("messages") or []) ] # Queried, not read off ``report.researcher``: that relationship is not # eager-loaded here, so it came back None and left ``wanted`` empty — # which is how a security-team reply got imported as the finding. from app.domains.bounty.models import BountyResearcher # noqa: PLC0415 researcher_email = "" if report.researcher_id is not None: researcher_email = ( await db.scalar( _select(BountyResearcher.email).where( BountyResearcher.id == report.researcher_id, ), ) ) or "" body = researcher_first_message(messages, researcher_email) if not body and researcher_email: # The recorded thread holds only OUR reply. That is the common case: # the FY27 import stamped the thread of the message it was reading, # and Gmail threads an outbound reply separately from the inbound # report in the sender's own mailbox — so ~half of the ids point at # our side of the conversation. The researcher's write-up is still # in this mailbox, under a different thread. # # Pinned to ``from:``: that returns # only correspondence with a known bounty submitter, which is the # ledger's own subject matter. It is deliberately NOT a subject or # keyword search, which would be a general walk of somebody's mail. body = await _search_researcher_message(service, researcher_email) if not body: # Mark the attempt. Without this the batch selection — newest-first # among reports lacking a write-up — re-attempts the same failures # every run, two API round trips each, and throughput decays to # zero before it ever reaches importable reports further down. The # job has to converge on its own. await svc.add_event( db, report=report, event_type=const.EVENT_UPDATED, actor=None, actor_kind="system", body=None, meta={"field": "raw_report", "import": IMPORT_NOT_FOUND, "mailbox": mailbox}, visibility="internal", ) await db.flush() return False report.raw_report = redact_payment_details(body)[:200000] await db.flush() return True async def backfill_bodies_and_assess( db: AsyncSession, *, limit: int = 25, mailbox: str = ARCHIVE_MAILBOX, ) -> dict[str, Any]: """Import write-ups for reports that have none, then have the AI read them. Batched, and each report isolated in its own SAVEPOINT: this runs over 149 reports against two external services, and one thread that 404s or one model reply that comes back malformed must cost exactly that report — not the 24 after it. Re-running is the way to work through the rest, and a report that already has a body is skipped, so re-running is free. """ from sqlalchemy import select as _select # noqa: PLC0415 from app.core.txn import best_effort # noqa: PLC0415 from app.domains.bounty import service as bsvc # noqa: PLC0415 from app.domains.bounty.models import BountyReport # noqa: PLC0415 rows = ( await db.execute( _select(BountyReport) .where( BountyReport.deleted_at.is_(None), or_( BountyReport.raw_report.is_(None), BountyReport.raw_report == "", ), # Never tried, or tried and worth trying again. A report # whose mail simply is not in the archive mailbox must not # sit at the head of the queue re-costing two API calls a # run and starving everything behind it. Clearing the # marker (or importing the mail) makes it eligible again. ~_exists_import_attempt(), ) .order_by(BountyReport.reported_at.desc()) .limit(limit), ) ).scalars().all() imported = assessed = skipped = failed = 0 for report in rows: code = report.code async with best_effort(db, label=f"import body {code}") as outcome: got = await import_body_for_report(db, report, mailbox=mailbox) if got: await bsvc.assess_with_ai(db, report, actor=None) if not outcome.ok: failed += 1 continue if (report.raw_report or "").strip(): imported += 1 if report.ai_assessment: assessed += 1 else: skipped += 1 return { "considered": len(rows), "imported": imported, "assessed": assessed, "no_thread_or_body": skipped, "failed": failed, "remaining_hint": ( "re-run until considered is 0; each run takes the newest batch" ), } # ════════════════════════════════════════════════════════════════════ # Attachments — what the researcher actually sent # ════════════════════════════════════════════════════════════════════ # # The write-up import walked the MIME tree looking for text and stepped # past every attachment part. Measured against the real mailbox on # 2026-08-26: 319 files, 143.1 MiB, across 54 of 149 reports — 266 # images, 25 logs, 19 PDFs and 9 PoC screen recordings. 「有些白帽子给了 # 附件 这个才是重要的」: for a good share of these the screenshot IS the # evidence and the prose is its caption. # # Bytes go to S3 through ``core.storage``; the bucket and region already # live in ``platform_settings`` and the auditor domain's evidence files # already take that path. #: Per-file ceiling. The largest single attachment observed in this corpus #: is 16.3 MiB, so this clears the real data with room to spare while #: still refusing something pathological. A file over the cap is recorded #: as skipped rather than silently ignored — see ``ATTACH_TOO_LARGE``. MAX_ATTACHMENT_BYTES = 32 * 1024 * 1024 #: Per-report ceiling. The heaviest report here carries 27.5 MiB across 22 #: files; this stops one thread from consuming the bucket without cutting #: into anything real. MAX_ATTACHMENT_BYTES_PER_REPORT = 96 * 1024 * 1024 #: Written on a report whose attachments have been swept, so a re-run does #: not re-pay the API calls for a thread that had none. Same convergence #: device as ``IMPORT_NOT_FOUND``: without it the reports with no #: attachments sit at the head of the batch forever and starve the ones #: never tried. #: Object-key prefix. Must be a prefix the bucket policy grants — see the #: key construction below for what happens when it is not. ATTACHMENT_PREFIX = "bounty_attachment" ATTACH_SWEPT = "attachments_swept" def _attach_swept_marker() -> Any: """A prior 「已经扫过附件」 marker on this report.""" from sqlalchemy import exists # noqa: PLC0415 from app.domains.bounty.models import ( # noqa: PLC0415 BountyReport, BountyReportEvent, ) return exists().where( BountyReportEvent.report_id == BountyReport.id, BountyReportEvent.event_type == const.EVENT_UPDATED, BountyReportEvent.meta["import"].as_string() == ATTACH_SWEPT, ) def collect_attachment_parts(message: dict[str, Any]) -> list[dict[str, Any]]: """Every real attachment part in one Gmail message. A part counts when it has BOTH a filename and an ``attachmentId``: that pair is what distinguishes a file from the ``text/plain`` and ``text/html`` alternatives, which carry their bytes inline in ``body.data`` and no filename at all. ``is_inline`` records ``Content-Disposition: inline`` and nothing else. It is metadata, not a display rule: the first version also treated the presence of a ``Content-ID`` as inline, which Gmail sets on ordinary attachments too, so 103 of 125 real files — including every one of the 9 proof-of-concept recordings — were classified as mail furniture. Measured afterwards, this corpus has no furniture at all: the smallest image is a 10 KB screenshot named ``04-unauth-orgname-disclosure.png``. Deduping on the bytes is what actually collapses a noisy thread (one report went from 90 parts to 6); guessing at intent did not. """ out: list[dict[str, Any]] = [] message_id = str(message.get("id") or "") def _walk(node: dict[str, Any]) -> None: body = node.get("body") or {} filename = str(node.get("filename") or "").strip() attachment_id = body.get("attachmentId") if filename and attachment_id: disposition = (_header(node, "Content-Disposition") or "").lower() out.append({ "message_id": message_id, "attachment_id": str(attachment_id), "filename": filename[:255], "mime": str(node.get("mimeType") or "application/octet-stream")[:128], "size": int(body.get("size") or 0), # Disposition only. A Content-ID is not evidence of # furniture — Gmail sets one on real attachments. "is_inline": "inline" in disposition, }) for part in node.get("parts") or []: _walk(part) _walk(message.get("payload") or {}) return out async def _researcher_messages( db: AsyncSession, report: Any, *, mailbox: str, ) -> list[dict[str, Any]]: """The raw Gmail messages on this report that the RESEARCHER sent. Raw, not flattened: ``_flatten_message`` reduces a message to sender + date + text and throws the MIME tree away, which is precisely how the attachments were lost the first time. The internal-sender filter is the same rule the write-up import learned the hard way — a forwarded thread has our address in ``From``, and importing our own reply as the researcher's finding put our words on the page as their report. Here the failure would be quieter and worse: our own internal screenshots filed as a researcher's evidence. """ from app.domains.bounty.models import BountyReportEvent # noqa: PLC0415 from app.domains.bounty.serializers import ( # noqa: PLC0415 mail_thread_url_from_events, ) events = ( await db.execute( select(BountyReportEvent).where( BountyReportEvent.report_id == report.id, BountyReportEvent.event_type == const.EVENT_CREATED, ), ) ).scalars().all() url = mail_thread_url_from_events(events) if not url: return [] thread_id = url.rsplit("/", 1)[-1] if thread_id.startswith("sav-"): # A placeholder the FY27 backfill wrote, not a real thread id. return [] service = await _service(db, mailbox) def _fetch() -> dict[str, Any]: fetched = service.users().threads().get( userId="me", id=thread_id, format="full", ).execute() return dict(fetched) if isinstance(fetched, dict) else {} raw = await asyncio.to_thread(_fetch) messages: list[dict[str, Any]] = [] for message in (raw.get("messages") or []): payload = message.get("payload") or {} sender = parseaddr(_header(payload, "From") or "")[1] if _looks_internal(sender): continue messages.append(message) return messages async def import_attachments_for_report( db: AsyncSession, report: Any, *, mailbox: str = ARCHIVE_MAILBOX, ) -> dict[str, int]: """Pull this report's researcher-sent attachments into S3. Idempotent on ``(report_id, sha256)``: the bytes are hashed before the row is written, so a re-run over a thread already swept inserts nothing. That matters more than usual here — researchers resend threads, so the same PNG legitimately arrives several times inside one thread and must land once. """ if mailbox not in ALLOWED_FETCH_MAILBOXES: raise ValueError(f"mailbox not allowed for fetch: {mailbox}") from app.core import storage # noqa: PLC0415 from app.domains.bounty.models import ( # noqa: PLC0415 BountyReportAttachment, ) stats = {"found": 0, "stored": 0, "duplicate": 0, "too_large": 0, "failed": 0} messages = await _researcher_messages(db, report, mailbox=mailbox) if not messages: return stats known = { row[0] for row in ( await db.execute( select(BountyReportAttachment.sha256).where( BountyReportAttachment.report_id == report.id, ), ) ).all() if row[0] } budget = MAX_ATTACHMENT_BYTES_PER_REPORT service = await _service(db, mailbox) report_id = report.id code = report.code for message in messages: for part in collect_attachment_parts(message): stats["found"] += 1 if part["size"] > MAX_ATTACHMENT_BYTES or part["size"] > budget: stats["too_large"] += 1 continue def _get(p: dict[str, Any] = part) -> dict[str, Any]: fetched = service.users().messages().attachments().get( userId="me", messageId=p["message_id"], id=p["attachment_id"], ).execute() return dict(fetched) if isinstance(fetched, dict) else {} try: blob = await asyncio.to_thread(_get) raw = base64.urlsafe_b64decode( (blob.get("data") or "") + "==", ) except Exception as exc: # noqa: BLE001 — one part, not the batch stats["failed"] += 1 if stats["failed"] <= 3: logger.warning( "bounty attachment fetch failed for %s (%s): %s", code, part["filename"], exc, ) continue if not raw: stats["failed"] += 1 continue digest = hashlib.sha256(raw).hexdigest() if digest in known: stats["duplicate"] += 1 continue # Key is derived from the digest, so the same bytes never # occupy two objects even across reports, and the extension is # dropped from the key on purpose: the filename is data, and a # researcher-supplied name has no business shaping a path. # # The prefix is NOT free-form. ``HelixProdUploadsAccess`` # allowlists one prefix per feature (``inventory_evidence/``, # ``evidence_documents/``, ``self_check_evidence/`` …), so a # name outside that list is denied PutObject — correctly. The # first run of this importer used ``bounty/`` and every one of # 162 files came back AccessDenied. key = f"{ATTACHMENT_PREFIX}/{code}/{digest}" try: object_key = await storage.put_object_bytes( db, relative_key=key, content=raw, content_type=part["mime"], ) except Exception as exc: # noqa: BLE001 — one file, not the batch # Logged, not just counted. The first production run # reported "failed: 162" with no reason anywhere, and the # cause (a bucket-policy prefix that was never granted) # took a bespoke script to find. A counter without a # reason is not an error report. stats["failed"] += 1 if stats["failed"] <= 3: logger.warning( "bounty attachment upload failed for %s (%s): %s", code, part["filename"], exc, ) continue db.add(BountyReportAttachment( report_id=report_id, object_key=object_key, filename=part["filename"], content_type=part["mime"], size_bytes=len(raw), sha256=digest, is_inline=bool(part["is_inline"]), source="researcher_mail", gmail_message_id=part["message_id"][:128], gmail_attachment_id=part["attachment_id"], )) known.add(digest) budget -= len(raw) stats["stored"] += 1 await db.flush() return stats async def backfill_attachments( db: AsyncSession, *, limit: int = 10, mailbox: str = ARCHIVE_MAILBOX, ) -> dict[str, Any]: """Sweep reports that have never had their attachments looked at. Each report isolated in its own SAVEPOINT, for the same reason the write-up backfill is: this runs over 149 reports against two external services, and one 404 thread or one S3 timeout must cost exactly that report. ``limit`` is smaller than the body backfill's because a single report here can move 27 MiB. Every considered report gets the ``ATTACH_SWEPT`` marker whether or not it had attachments, which is what makes the batch converge — the body import originally lacked that and the reports with nothing to find sat at the head of the queue re-costing API calls every run. """ from app.core.txn import best_effort # noqa: PLC0415 from app.domains.bounty import service as bsvc # noqa: PLC0415 from app.domains.bounty.models import BountyReport # noqa: PLC0415 rows = ( await db.execute( select(BountyReport) .where( BountyReport.deleted_at.is_(None), ~_attach_swept_marker(), ) .order_by(BountyReport.reported_at.desc()) .limit(limit), ) ).scalars().all() totals = {"found": 0, "stored": 0, "duplicate": 0, "too_large": 0, "failed": 0} swept = errored = 0 for report in rows: code = report.code async with best_effort(db, label=f"attachments {code}") as outcome: stats = await import_attachments_for_report( db, report, mailbox=mailbox, ) for key, value in stats.items(): totals[key] += value # The marker is the convergence device, and writing it is a # claim that this report has been dealt with. Two conditions, # not one: # # * the sweep did not raise (checked below via ``outcome``), # so a crashed report stays eligible; and # * it did not find files and fail on every single one. # # The second condition is the one the first production run # taught. Every upload was denied by a bucket policy that had # never granted this prefix, ``import_attachments_for_report`` # counted the failures and returned normally, so the outcome # was "ok" and 60 reports were marked swept having stored # nothing — retiring them from the batch on a systemic error # that a re-run would have fixed. A wholesale failure is not a # sweep. wholesale_failure = ( stats["found"] > 0 and stats["stored"] == 0 and stats["duplicate"] == 0 and stats["failed"] == stats["found"] ) if wholesale_failure: raise RuntimeError( f"every attachment failed for {code} " f"({stats['failed']}/{stats['found']}) — not marking swept", ) await bsvc.add_event( db, report=report, event_type=const.EVENT_UPDATED, actor=None, actor_kind="system", body=( f"附件扫描完成:找到 {stats['found']} 个," f"入库 {stats['stored']} 个。" ), meta={"import": ATTACH_SWEPT, **stats}, visibility="internal", ) if outcome.ok: swept += 1 else: errored += 1 return { "considered": len(rows), "swept": swept, "errored": errored, **totals, "remaining_hint": "re-run until considered is 0", } #: Prefixes a mail client puts in front of a subject it is replying to. #: Matched case-insensitively and repeatedly — a thread three replies deep #: arrives as ``Re: Re: Fwd: ``. _REPLY_PREFIXES: tuple[str, ...] = ( "re:", "re :", "fwd:", "fw:", "答复:", "回复:", "转发:", ) def _normalize_subject(subject: str | None) -> str: """A subject reduced to what survives a reply chain. Used only to SUGGEST a report, never to attach one — the operator confirms. Two mails can share a subject and be different findings. """ text = (subject or "").strip().lower() changed = True while changed: changed = False for prefix in _REPLY_PREFIXES: if text.startswith(prefix): text = text[len(prefix):].lstrip() changed = True # Collapse whitespace so "IDOR - cross tenant" matches "IDOR - cross # tenant"; drop the bracketed programme tags researchers add and drop # again on later replies. return " ".join(text.split()) async def suggest_report_for( db: AsyncSession, row: BountyMailIntake, ) -> BountyReport | None: """The report this mail is most likely a follow-up on, or nothing. **Why this exists at all.** ``poll`` attaches a follow-up when the mail's ``thread_id`` already produced a report — and Gmail thread ids are PER ACCOUNT. The 149 FY27 reports carry thread ids from the platform owner's mailbox, while the live poller reads ``bounty-intake@``, so a chase on any of those reports can never match by thread and arrives looking like a brand-new finding. That is not hypothetical: the first mail the live poller queued (2026-08-28) was ``Re: [Bug Bounty] IDOR - Cross-Tenant Metrics Exposure…`` — a researcher asking for an update on BB-2026-144 after ten days. Subject-only, and a SUGGESTION: the operator presses 「挂到已有报告」. Guessing here would put a researcher's chase onto the wrong finding, which is worse than leaving it in the queue. """ normalized = _normalize_subject(row.subject) if len(normalized) < 12: # Short subjects ("update?", "hello") match half the ledger. return None # Only mails that LOOK like a reply get a suggestion. A first report # whose title happens to resemble an old one is a new finding, and # offering to fold it into the old one invites exactly that mistake. lowered = (row.subject or "").strip().lower() if not any(lowered.startswith(p) for p in _REPLY_PREFIXES): return None candidates = ( await db.execute( select(BountyReport).where( BountyReport.deleted_at.is_(None), ).order_by(BountyReport.created_at.desc()).limit(500), ) ).scalars().all() for report in candidates: if _normalize_subject(report.title) == normalized: return report return None async def attach_to_report( db: AsyncSession, intake_id: Any, *, actor: IdentityPerson | None, report_id: Any, ) -> BountyReport: """The operator's decision: this queued mail belongs to that report. The counterpart to ``accept``. Without it the queue offered only 「接受」(a second report for a conversation that already has one — and every count, the budget attribution and the reward owed are computed off these rows) or 「忽略」(the chase is silently dropped and the researcher keeps waiting). Reuses ``attach_followup``, so the mail is logged as ``inbound_logged`` on the report and the desk gets the same chase notification the automatic path sends. """ row = await _claim_pending(db, intake_id) report = await db.get(BountyReport, report_id) if report is None or report.deleted_at is not None: raise ValidationError("report not found") attached = await attach_followup(db, row, report.id) if not attached: # pragma: no cover - guarded by the load above raise ValidationError("report not found") if actor is not None: row.decided_by = actor.id await db.flush() return report