"""Wire shapes. ``include_sensitive`` is the security boundary of this domain, enforced here rather than in the frontend: * the researcher's identity (email, name, country, notes), * every money field (reward amount, payment status, references), * the correspondence log, are visible to the security team only. The assigned business owner gets the technical substance — title, internal summary, reproduction steps, the researcher's own write-up, affected system, severity, remediation state — and nothing that would let them contact the researcher or learn the payout. Why that split: the owner is a regular employee pulled in to answer one question, and handing them a stranger's address and a dollar figure widens a report's blast radius for no operational gain. The write-up is not in that category — 「这个白帽子的正文没有什么保密的,尤其是说哪里有 bug 的那种」. It describes OUR system's defect, which is exactly what the owner is being asked to judge, and withholding it left the confirmation page with a title and nothing else. Two guardrails remain on it. Banking details are stripped at import (``mail_intake.redact_payment_details``), so an invoice pasted into a first message never lands in the column. And ``summary`` stays in the security-team half despite sounding like the technical one: in this corpus it is the operator's own scratch column and carries lines like 「已向研究者索取发票,尚未付款」 — the payment state, in free text. """ from __future__ import annotations from datetime import date, datetime from decimal import Decimal from typing import Any from app.core import fiscal_year as fy_mod from app.domains.bounty import catalog from app.domains.bounty import constants as const from app.domains.bounty.models import ( BountyReplyTemplate, BountyReport, BountyReportEvent, BountyResearcher, BountySeverityTier, ) def _iso(value: datetime | date | None) -> str | None: if value is None: return None return value.isoformat() def _money(value: Decimal | None) -> float | None: if value is None: return None return float(Decimal(str(value))) def serialize_tier(tier: BountySeverityTier) -> dict[str, Any]: return { "id": str(tier.id), "key": tier.key, "label_zh": tier.label_zh, "label_en": tier.label_en, "label_ja": tier.label_ja, "default_reward_usd": _money(tier.default_reward_usd), "max_reward_usd": _money(tier.max_reward_usd), "fix_sla_days": tier.fix_sla_days, "color": tier.color, "sort_order": tier.sort_order, "enabled": tier.enabled, "is_builtin": tier.is_builtin, "updated_at": _iso(tier.updated_at), } def serialize_template( template: BountyReplyTemplate, *, include_body: bool = True, ) -> dict[str, Any]: payload: dict[str, Any] = { "id": str(template.id), "key": template.key, "name_zh": template.name_zh, "name_en": template.name_en, "name_ja": template.name_ja, "category": template.category, "subject": template.subject, "variables": list(template.variables or []), "sort_order": template.sort_order, "enabled": template.enabled, "is_builtin": template.is_builtin, "use_count": template.use_count, "updated_at": _iso(template.updated_at), } if include_body: payload["body"] = template.body return payload def serialize_researcher( researcher: BountyResearcher, *, stats: dict[str, Any] | None = None, ) -> dict[str, Any]: payload: dict[str, Any] = { "id": str(researcher.id), "email": researcher.email, "display_name": researcher.display_name, "handle": researcher.handle, "country": researcher.country, "notes": researcher.notes, "is_blocked": researcher.is_blocked, # An alias points at the person's canonical row. Reports keep # pointing here, so the console can still show which address a # report arrived on while attributing it to the right person. "merged_into_id": ( str(researcher.merged_into_id) if researcher.merged_into_id else None ), "first_report_at": _iso(researcher.first_report_at), "last_report_at": _iso(researcher.last_report_at), "created_at": _iso(researcher.created_at), } if stats is not None: total = stats.get("report_count") or 0 valid = stats.get("valid_count") or 0 payload["stats"] = { "report_count": total, "valid_count": valid, "rejected_count": stats.get("rejected_count") or 0, "paid_usd": _money(stats.get("paid_usd")) or 0.0, # Valid-rate over *decided* reports only; counting still-open # reports in the denominator would make an active submitter # look worse than a dormant one. "valid_rate": ( round(valid / total, 4) if total else None ), } return payload def _tier_label(tier: BountySeverityTier | None) -> dict[str, str] | None: if tier is None: return None return { "zh": tier.label_zh, "en": tier.label_en, "ja": tier.label_ja, "color": tier.color or "", } def serialize_report( report: BountyReport, *, researcher: BountyResearcher | None = None, tiers: dict[str, BountySeverityTier] | None = None, include_sensitive: bool, include_content: bool = True, system: dict[str, Any] | None = None, duplicate_of: dict[str, Any] | None = None, fy_start_month: int | None = None, mail_thread_url: str | None = None, ) -> dict[str, Any]: tiers = tiers or {} tier = tiers.get(report.severity) if report.severity else None suggested_tier = ( tiers.get(report.suggested_severity) if report.suggested_severity else None ) overdue = bool( report.fix_due_date and report.fix_due_date < date.today() and report.remediation_status not in const.REMEDIATION_SETTLED ) days_to_due = ( (report.fix_due_date - date.today()).days if report.fix_due_date else None ) payload: dict[str, Any] = { "id": str(report.id), "code": report.code, "title": report.title, "source": report.source, "reported_at": _iso(report.reported_at), "affected_system": report.affected_system, # The registry link, when the operator made one. ``system`` is # injected by the caller (through the registry coordinator) — # this domain does not import the registry. A linked-but-deleted # entry yields ``system: None`` and the free-text # ``affected_system`` carries the display, rather than a # fabricated name. # What this report duplicates, when it is one. ``duplicate_of`` # is injected by the caller (a second row in the same table); # ``duplicate_note`` covers an original that is not a Helix row # at all, which is the common case. "duplicate_of_id": ( str(report.duplicate_of_id) if report.duplicate_of_id else None ), "duplicate_of": duplicate_of, "duplicate_note": report.duplicate_note, "system_id": str(report.system_id) if report.system_id else None, "system": system, "affected_urls": list(report.affected_urls or []), "vuln_type": report.vuln_type, "cve_id": report.cve_id, # The security desk sees the arrival placeholder, because the desk # is who replaces it. Everyone else sees whether a judgement # exists: for an owner or a lead the tier is a fact about how bad # this is, and 「告知」 rendered onto an ungraded 7,735-record PII # leak states a fact that is false. The owner page already had a # 「待定级」 fallback — it just never fired, because the placeholder # is not null. Decided here, not in the component: which severity # is real is a decision, and the frontend is presentation. "severity": ( report.severity if include_sensitive else const.graded_severity(report) ), "severity_label": ( _tier_label(tier) if include_sensitive or not report.severity_is_default else None ), "suggested_severity": report.suggested_severity, "suggested_severity_label": _tier_label(suggested_tier), "status": report.status, # The one line a human reads. Derived from the axes above by the # single definition in ``constants`` — see there for why it is # derived rather than stored, and which two orderings real rows # forced. The axes stay on the wire for the security desk's own # controls; nothing else should be reading them to decide what to # show. "stage": const.lifecycle_stage(report), # Which letter to pre-select, resolved HERE by the one authoritative # function. The console used to re-derive it from two maps in # ``meta``, which — since neither map has a ``confirmed`` entry — # meant reading "the letter for a *new* report" as the fallback for # a confirmed-but-ungraded one. Those happen to be the same letter # today; the moment acknowledging receipt and "we are evaluating" # become two letters, an entirely ordinary edit, the console would # start offering the wrong one silently. Frontend is presentation # (CLAUDE.md); which letter to send is a decision. # A level the report merely ARRIVED with must not choose the # letter: offering a critical finding the $0 informational reward # letter is unrecoverable once sent. Until a person grades it, the # answer is "under evaluation". "suggested_template_key": catalog.suggested_template( report.status, const.graded_severity(report), ), "severity_is_default": bool(report.severity_is_default), "remediation_status": report.remediation_status, "assignee_person_id": ( str(report.assignee_person_id) if report.assignee_person_id else None ), "assignee_name": report.assignee_name, "assignee_department": report.assignee_department, "assigned_at": _iso(report.assigned_at), "last_nudged_at": _iso(report.last_nudged_at), "owner_decision": report.owner_decision, "owner_note": report.owner_note, "owner_responded_at": _iso(report.owner_responded_at), "fix_eta": _iso(report.fix_eta), "fix_due_date": _iso(report.fix_due_date), "fixed_at": _iso(report.fixed_at), "is_overdue": overdue, "days_to_fix_due": days_to_due, # How long the researcher has been waiting. The only dimension in # this corpus that can rank priority objectively (2 days … 295), # and computing it in three different clients drifts. "days_silent": ( (date.today() - report.reported_at.date()).days if report.reported_at else None ), "created_at": _iso(report.created_at), "updated_at": _iso(report.updated_at), } if include_content: payload["summary"] = report.summary payload["reproduction"] = report.reproduction # The researcher's own words about our defect. Not a security-team # field: it is the one artifact that answers 「哪里有 bug」, and the # owner cannot confirm a finding they are not allowed to read. # Banking details were stripped at import, and the correspondence # (our replies, the money talk) is a separate event stream that # stays behind include_sensitive. payload["raw_report"] = report.raw_report # Where the researcher's own words actually live. The FY27 import # kept correspondence in the mailbox on purpose, so for 149 of 149 # reports the technical detail is ONLY in the thread — a console # that cannot point at it is asking the operator to go and find it # by searching their inbox for the title. payload["mail_thread_url"] = mail_thread_url # The model's reading. Behind include_content with the rest of the # write-up, because it quotes it — an assessment is only meaningful # to a reader who can also see what was assessed. payload["ai_assessment"] = report.ai_assessment or None payload["ai_assessed_at"] = _iso(report.ai_assessed_at) if include_sensitive: payload.update({ "researcher": ( serialize_researcher(researcher) if researcher else None ), "researcher_id": str(report.researcher_id), "reward_amount_usd": _money(report.reward_amount_usd), "reward_is_override": report.reward_is_override, "reward_decided_at": _iso(report.reward_decided_at), "tier_default_reward_usd": ( _money(tier.default_reward_usd) if tier else None ), "tier_max_reward_usd": _money(tier.max_reward_usd) if tier else None, "budget_track": report.budget_track, "fiscal_year": report.fiscal_year, # The integer is the START year (2026 = 2026-04 ~ 2027-03); # the company says "FY27". One formatter for the platform is # why fiscal_year.fy_label exists — the drawer printing the # raw integer beside a budget bar labelled FY27 is exactly # the drift that change was made to stop. Callers that have # the start month pass it; the rest get None and the client # falls back to the integer rather than inventing a label. "fiscal_year_label": ( fy_mod.fy_label(report.fiscal_year, fy_start_month) if report.fiscal_year is not None and fy_start_month else None ), "payment_status": report.payment_status, "payment_requested_at": _iso(report.payment_requested_at), "paid_at": _iso(report.paid_at), "payment_reference": report.payment_reference, "payment_note": report.payment_note, "last_replied_at": _iso(report.last_replied_at), "last_reply_template_key": report.last_reply_template_key, "reply_count": report.reply_count, "closed_at": _iso(report.closed_at), "deleted_at": _iso(report.deleted_at), }) return payload def serialize_event(event: BountyReportEvent) -> dict[str, Any]: return { "id": str(event.id), "event_type": event.event_type, "actor_kind": event.actor_kind, "actor_person_id": ( str(event.actor_person_id) if event.actor_person_id else None ), "actor_name": event.actor_name, "visibility": event.visibility, "body": event.body, "meta": event.meta or {}, "created_at": _iso(event.created_at), } def serialize_budget(budget: dict[str, Any]) -> dict[str, Any]: """Money fields out of ``service.overview``/analytics as JSON floats.""" return { **budget, "budget_usd": _money(budget.get("budget_usd")) or 0.0, "paid_usd": _money(budget.get("paid_usd")) or 0.0, "committed_usd": _money(budget.get("committed_usd")) or 0.0, "total_usd": _money(budget.get("total_usd")) or 0.0, "remaining_usd": _money(budget.get("remaining_usd")) or 0.0, } def serialize_mail_intake( row: Any, *, suggested_report: dict[str, Any] | None = None, ) -> dict[str, Any]: """A queued mail. Carries a snippet, never the body: researchers send tax forms and bank details unprompted, and their correspondence stays in the mailbox by decision.""" return { "id": str(row.id), "message_id": row.message_id, "thread_id": row.thread_id, "from_email": row.from_email, "from_name": row.from_name, "subject": row.subject, "snippet": row.snippet, "received_at": _iso(row.received_at), "status": row.status, # The AI's reading of subject + snippet. Advisory: it is shown next # to the mail so a human can accept or reject it, never applied. "parsed": row.parsed or None, "report_id": str(row.report_id) if row.report_id else None, "decided_at": _iso(row.decided_at), "note": row.note, # Set by the router when the subject looks like a reply to a report # we already have. Gmail thread ids are per-account, so a chase on # any of the 149 imported reports cannot match by thread and lands # here looking like a new finding — this is what lets the operator # fold it in with one click instead of opening a duplicate. "suggested_report": suggested_report, } def mail_thread_url_from_events(events: Any) -> str | None: """Deep link to the security@ thread a report came from, read off its events. The id is stamped on the ``created`` event by the FY27 import (``meta.backfill = "gmail:"``) and by the mail-intake queue. It lives on the event rather than the report because the report table deliberately holds no correspondence — and the console still has to be able to point at where the researcher's own words are, or it is asking the operator to go and find the thread by searching their inbox for the title. Returns None when nothing wrote one: a hand-pasted report has no thread, and a dead link is worse than no link. The ``sav-`` ids the import used as placeholders for threads it could not resolve are excluded for the same reason. """ for event in events or []: if getattr(event, "event_type", None) != const.EVENT_CREATED: continue marker = str((getattr(event, "meta", None) or {}).get("backfill") or "") if not marker.startswith("gmail:"): continue thread = marker.split(":", 1)[1].strip() if thread and not thread.startswith("sav-"): return f"https://mail.google.com/mail/u/0/#all/{thread}" return None def serialize_attachment(row: Any) -> dict[str, Any]: """One attachment, without the object key. ``object_key`` never crosses the wire. A key is the one input to ``presign_get``, and this payload goes to business owners and leads — handing them the key alongside a download endpoint would let a guessed or edited key be signed on request. The id is the handle; the server resolves it to a key behind whichever ACL that surface already enforces. """ return { "id": str(row.id), "filename": row.filename, "content_type": row.content_type, "size_bytes": row.size_bytes, "is_inline": bool(row.is_inline), "source": row.source, "imported_at": _iso(row.imported_at), }