"""Report lifecycle: intake, assignment, triage, severity, remediation, money, correspondence. Every mutation here does three things, always in this order: 1. validate the transition against the machines in ``constants``, 2. mutate the report, 3. append a ``BountyReportEvent``. Step 3 is not optional bookkeeping — the detail page's timeline, the business owner's visible history and the "what did we tell the researcher" record are all reads of that table. A mutation that skips it is a silent state change. The domain never imports another domain. Notifying the assigned owner (PR2) goes through the notification service from the router layer, not from here. """ from __future__ import annotations import uuid from datetime import UTC, date, datetime, timedelta from decimal import Decimal from typing import Any from sqlalchemy import Select, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from app.core.exceptions import AppException, NotFoundError, ValidationError from app.core.txn import best_effort 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.models import ( BountyReplyTemplate, BountyReport, BountyReportAttachment, BountyReportEvent, BountyResearcher, BountySeverityTier, ) from app.domains.bounty.rendering import ( RenderedReply, build_context, declared_tokens, render, ) from app.identity.person_models import IdentityPerson def _now() -> datetime: return datetime.now(UTC) def _naive(dt: datetime) -> datetime: """Drop tzinfo so a value can be compared against a ``DateTime(timezone=True)`` column, which round-trips **naive** on a fresh load. Mixing an aware ``datetime.now(UTC)`` with a freshly loaded column value raises ``TypeError`` at runtime, so every comparison in this module goes through here.""" 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) # ════════════════════════════════════════════════════════════════════ # Researchers # ════════════════════════════════════════════════════════════════════ def normalize_email(email: str) -> str: cleaned = (email or "").strip().lower() if len(cleaned) > 320: raise ValidationError("email address is too long") if cleaned.count("@") != 1 or any(char.isspace() for char in cleaned): raise ValidationError(f"'{email}' is not a valid email address") local, domain = cleaned.split("@", 1) if ( not local or not domain or "." not in domain or domain.startswith(".") or domain.endswith(".") ): raise ValidationError(f"'{email}' is not a valid email address") return cleaned async def upsert_researcher( db: AsyncSession, *, email: str, display_name: str | None = None, handle: str | None = None, country: str | None = None, notes: str | None = None, actor_id: uuid.UUID | None = None, ) -> BountyResearcher: """Find-or-create by email. An existing row's fields are only FILLED, never overwritten: the operator may have curated the display name, and a later mail signed differently must not clobber it.""" normalized = normalize_email(email) row = ( await db.execute( select(BountyResearcher).where(BountyResearcher.email == normalized), ) ).scalar_one_or_none() if row is None: row = BountyResearcher( email=normalized, display_name=(display_name or None), handle=(handle or None), country=(country or None), notes=(notes or None), created_by=actor_id, ) db.add(row) await db.flush() return row if display_name and not row.display_name: row.display_name = display_name if handle and not row.handle: row.handle = handle if country and not row.country: row.country = country await db.flush() return row async def get_researcher(db: AsyncSession, researcher_id: uuid.UUID) -> BountyResearcher: row = await db.get(BountyResearcher, researcher_id) if row is None: raise NotFoundError("researcher not found") return row async def load_assignable_person( db: AsyncSession, person_id: uuid.UUID, ) -> IdentityPerson: """Load the prospective business owner with ``org_memberships`` eager. ``IdentityPerson.department`` is a *property* over that relationship; touching it on a lazily-loaded instance raises ``MissingGreenlet`` under asyncio. Assignment snapshots the department, so this loader is the only sanctioned way to fetch an assignee. """ person = ( await db.execute( select(IdentityPerson) .options(selectinload(IdentityPerson.org_memberships)) .where( IdentityPerson.id == person_id, IdentityPerson.deleted_at.is_(None), ), ) ).scalar_one_or_none() if person is None: raise NotFoundError("person not found") if person.employment_status != "active": raise AppException( code="BOUNTY_ASSIGNEE_INACTIVE", message=( f"{person.display_name or person.work_email} 已不是在职员工," "不能作为漏洞确认责任人。" ), status_code=409, ) return person async def update_researcher( db: AsyncSession, researcher_id: uuid.UUID, *, patch: dict[str, Any], ) -> BountyResearcher: row = await get_researcher(db, researcher_id) if "email" in patch and patch["email"]: new_email = normalize_email(patch["email"]) if new_email != row.email: clash = ( await db.execute( select(BountyResearcher.id).where( BountyResearcher.email == new_email, BountyResearcher.id != row.id, ), ) ).scalar_one_or_none() if clash is not None: raise AppException( code="BOUNTY_RESEARCHER_EXISTS", message=f"另一位白帽子已使用邮箱 {new_email}", status_code=409, ) row.email = new_email for field_name in ("display_name", "handle", "country", "notes", "is_blocked"): if field_name in patch: value = patch[field_name] setattr( row, field_name, value if field_name == "is_blocked" else (value or None), ) await db.flush() return row async def researcher_stats( db: AsyncSession, researcher_ids: list[uuid.UUID], ) -> dict[uuid.UUID, dict[str, Any]]: """Per-researcher aggregates: reports, valid, paid, total USD paid. One grouped query rather than a per-row count — the researcher list is small but this also backs the leaderboard, which is not. **Merged addresses aggregate together.** One person with two addresses was counted as two people, which made every figure here wrong for them — and these are the figures that answer "how many has he sent us" and "how much have we paid him". Reports keep pointing at the address that sent them; only the totals resolve, and they resolve onto BOTH the canonical id and each alias id, so a caller holding either one gets the person's real numbers. """ if not researcher_ids: return {} # Resolve aliases in two BOUNDED queries. The first version read every # researcher row on every call, which is nothing at 56 rows and a # full-table scan behind the leaderboard — a function whose own # docstring says it has to scale should not do that. # # 1. the canonical row for each id the caller asked about asked = list(dict.fromkeys(researcher_ids)) canonical_of: dict[uuid.UUID, uuid.UUID] = {} for rid, merged_into in ( await db.execute( select(BountyResearcher.id, BountyResearcher.merged_into_id) .where(BountyResearcher.id.in_(asked)), ) ).all(): canonical_of[rid] = merged_into or rid wanted = {canonical_of.get(rid, rid) for rid in asked} # 2. every address belonging to those people. Without this the alias's # reports are missing from the canonical row's totals — which is the # entire bug this resolution exists to fix. for rid, merged_into in ( await db.execute( select(BountyResearcher.id, BountyResearcher.merged_into_id) .where( or_( BountyResearcher.merged_into_id.in_(wanted), BountyResearcher.id.in_(wanted), ), ), ) ).all(): canonical_of[rid] = merged_into or rid researcher_ids = [ rid for rid, canonical in canonical_of.items() if canonical in wanted ] or asked stmt = ( select( BountyReport.researcher_id, func.count().label("total"), func.sum( func.if_(BountyReport.status == const.STATUS_CONFIRMED, 1, 0), ).label("valid"), func.sum( func.if_( BountyReport.status.in_(( const.STATUS_FALSE_POSITIVE, const.STATUS_OUT_OF_SCOPE, const.STATUS_DUPLICATE, )), 1, 0, ), ).label("rejected"), func.coalesce( func.sum( func.if_( BountyReport.payment_status == const.PAYMENT_PAID, BountyReport.reward_amount_usd, 0, ), ), 0, ).label("paid_usd"), func.max(BountyReport.reported_at).label("last_report_at"), ) .where( BountyReport.researcher_id.in_(researcher_ids), BountyReport.deleted_at.is_(None), ) .group_by(BountyReport.researcher_id) ) merged: dict[uuid.UUID, dict[str, Any]] = {} for row in (await db.execute(stmt)).all(): key = canonical_of.get(row.researcher_id, row.researcher_id) entry = merged.setdefault(key, { "report_count": 0, "valid_count": 0, "rejected_count": 0, "paid_usd": Decimal("0"), "last_report_at": None, }) entry["report_count"] += int(row.total or 0) entry["valid_count"] += int(row.valid or 0) entry["rejected_count"] += int(row.rejected or 0) entry["paid_usd"] += Decimal(str(row.paid_usd or 0)) if row.last_report_at and ( entry["last_report_at"] is None or row.last_report_at > entry["last_report_at"] ): entry["last_report_at"] = row.last_report_at # File the person's totals under every id that is them, so a caller # holding an alias id is not told the person has zero reports. out: dict[uuid.UUID, dict[str, Any]] = {} for rid, canonical in canonical_of.items(): if canonical in merged: out[rid] = merged[canonical] return out # ════════════════════════════════════════════════════════════════════ # Report codes # ════════════════════════════════════════════════════════════════════ CODE_PREFIX = "BB" async def next_report_code(db: AsyncSession, *, when: datetime | None = None) -> str: """``BB-2026-001``, sequential within the calendar year. Derived from the max existing code for that year rather than a counter row, so it is correct after a manual data fix and cannot drift. The unique constraint on ``code`` is the real guard: on the (vanishingly rare) concurrent-intake collision the insert fails and the caller retries — see ``create_report``. """ year = (when or _now()).year prefix = f"{CODE_PREFIX}-{year}-" last = ( await db.execute( select(func.max(BountyReport.code)).where( BountyReport.code.like(f"{prefix}%"), ), ) ).scalar_one_or_none() seq = 1 if last: tail = last[len(prefix):] if tail.isdigit(): seq = int(tail) + 1 return f"{prefix}{seq:03d}" # ════════════════════════════════════════════════════════════════════ # Events # ════════════════════════════════════════════════════════════════════ async def add_event( db: AsyncSession, *, report: BountyReport, event_type: str, actor: IdentityPerson | None = None, actor_kind: str = const.ACTOR_USER, body: str | None = None, meta: dict[str, Any] | None = None, visibility: str = const.VISIBILITY_INTERNAL, ) -> BountyReportEvent: if event_type not in const.EVENT_TYPES: raise ValidationError(f"unknown event type '{event_type}'") if visibility not in const.VISIBILITIES: raise ValidationError(f"unknown visibility '{visibility}'") row = BountyReportEvent( report_id=report.id, event_type=event_type, actor_kind=actor_kind, actor_person_id=actor.id if actor is not None else None, actor_name=( (actor.display_name or actor.work_email) if actor is not None else None ), visibility=visibility, body=body, meta=meta, ) db.add(row) await db.flush() return row async def list_events( db: AsyncSession, report_id: uuid.UUID, *, internal: bool, ) -> list[BountyReportEvent]: stmt = ( select(BountyReportEvent) .where(BountyReportEvent.report_id == report_id) .order_by(BountyReportEvent.created_at.asc(), BountyReportEvent.id.asc()) ) if not internal: stmt = stmt.where( BountyReportEvent.visibility == const.VISIBILITY_SHARED, ) return list((await db.execute(stmt)).scalars().all()) # ════════════════════════════════════════════════════════════════════ # Reports — intake # ════════════════════════════════════════════════════════════════════ async def create_report( db: AsyncSession, *, title: str, researcher_email: str, researcher_name: str | None = None, researcher_handle: str | None = None, researcher_country: str | None = None, reported_at: datetime | None = None, raw_report: str | None = None, summary: str | None = None, reproduction: str | None = None, affected_system: str | None = None, system_id: uuid.UUID | None = None, affected_urls: list[str] | None = None, vuln_type: str | None = None, cve_id: str | None = None, suggested_severity: str | None = None, budget_track: str = const.DEFAULT_BUDGET_TRACK, source: str = const.SOURCE_EMAIL, actor: IdentityPerson | None = None, ) -> BountyReport: title = (title or "").strip() if not title: raise ValidationError("report title is required") if len(title) > const.MAX_TITLE_CHARS: raise ValidationError( f"report title exceeds {const.MAX_TITLE_CHARS} characters", ) if raw_report and len(raw_report) > const.MAX_RAW_REPORT_CHARS: raise ValidationError( f"pasted report exceeds {const.MAX_RAW_REPORT_CHARS} characters; " "trim quoted history / attachments before pasting", ) if source not in const.SOURCES: raise ValidationError(f"unknown source '{source}'") if budget_track not in const.BUDGET_TRACKS: raise ValidationError(f"unknown budget track '{budget_track}'") if vuln_type is not None and vuln_type not in const.VULN_TYPES: raise ValidationError(f"unknown vulnerability type '{vuln_type}'") if suggested_severity is not None: await cfg.get_tier(db, suggested_severity) researcher = await upsert_researcher( db, email=researcher_email, display_name=researcher_name, handle=researcher_handle, country=researcher_country, actor_id=actor.id if actor else None, ) when = reported_at or _now() report = BountyReport( code=await next_report_code(db, when=when), title=title, researcher_id=researcher.id, source=source, reported_at=when, raw_report=raw_report or None, summary=(summary or None), reproduction=(reproduction or None), # Stripped, because it is the grouping key for the matrix and its # lane, and it arrives by paste out of a researcher's mail. A # padded " www.pingcap.com " grouped as its own ownership row and # (before the lane's SQL also trimmed) showed a count the # drill-down could not produce a single card for. affected_system=((affected_system or "").strip() or None), system_id=system_id, affected_urls=_clean_urls(affected_urls), vuln_type=vuln_type, cve_id=(cve_id or None), suggested_severity=suggested_severity, budget_track=budget_track, status=const.STATUS_NEW, remediation_status=const.REMEDIATION_NOT_APPLICABLE, payment_status=const.PAYMENT_NOT_APPLICABLE, created_by=actor.id if actor else None, ) db.add(report) await db.flush() # Researcher activity window — drives the "repeat submitter" signal. # Widened, never narrowed: a back-dated report must be able to move # ``first_report_at`` earlier without disturbing ``last_report_at``. reported_naive = _naive(when) first_seen = _naive_opt(researcher.first_report_at) if first_seen is None or reported_naive < first_seen: researcher.first_report_at = when last_seen = _naive_opt(researcher.last_report_at) if last_seen is None or reported_naive > last_seen: researcher.last_report_at = when # Arrive at the LOWEST tier rather than ungraded. Grading gated # everything downstream (reward tier, fix deadline, reply letter), so # 130 reports could not move until a person picked a level — which is # why the queue never drained. ``informational`` commits $0 and sets no # deadline, and the flag remembers that nobody chose it, so the reply # composer stays on "under evaluation" and 待定级 still counts it. if report.severity is None: default_tier = await cfg.lowest_enabled_tier(db) if default_tier is not None: report.severity = default_tier.key report.severity_is_default = True # Born linked to a system that has an owner? Then it already knows # who answers for it; see auto_assign_from_system. if report.system_id is not None: await auto_assign_from_system(db, report, actor=actor) await add_event( db, report=report, event_type=const.EVENT_CREATED, actor=actor, meta={"code": report.code, "source": source}, visibility=const.VISIBILITY_INTERNAL, ) await db.flush() return report def _clean_urls(urls: list[str] | None) -> list[str] | None: if not urls: return None cleaned = [] for raw in urls: value = (raw or "").strip() if not value: continue if len(value) > 2000: value = value[:2000] cleaned.append(value) if len(cleaned) >= 20: break return cleaned or None async def update_report( db: AsyncSession, report: BountyReport, *, patch: dict[str, Any], actor: IdentityPerson | None, ) -> BountyReport: """Edit the descriptive fields. Status / severity / money / assignment each have their own dedicated transition function — they are NOT reachable from here, so no caller can sidestep a state machine by PATCHing a status column.""" changed: dict[str, Any] = {} if "title" in patch: title = (patch["title"] or "").strip() if not title: raise ValidationError("report title is required") if len(title) > const.MAX_TITLE_CHARS: raise ValidationError( f"report title exceeds {const.MAX_TITLE_CHARS} characters", ) if title != report.title: changed["title"] = title report.title = title if "raw_report" in patch: raw = patch["raw_report"] or None if raw and len(raw) > const.MAX_RAW_REPORT_CHARS: raise ValidationError( f"pasted report exceeds {const.MAX_RAW_REPORT_CHARS} characters", ) report.raw_report = raw changed["raw_report"] = "updated" for field_name in ("summary", "reproduction", "affected_system", "cve_id"): if field_name in patch: value = (patch[field_name] or None) if field_name == "affected_system" and value is not None: # See create_report: this is a grouping key, not prose. value = value.strip() or None if value != getattr(report, field_name): changed[field_name] = "updated" setattr(report, field_name, value) if "system_id" in patch: # Existence is validated at the router (the registry lives in # another domain); here the column is simply re-pointed. Clearing # it back to free text is legitimate — an operator who linked the # wrong system must be able to undo it. new_system_id = patch["system_id"] if new_system_id != report.system_id: changed["system_id"] = ( str(new_system_id) if new_system_id else None ) report.system_id = new_system_id if new_system_id is not None: # Same reasoning as the bulk path: attribution and routing are # one intention, so linking a system routes the report to its # owner rather than parking it in 待指派 for a second pass. await db.flush() await auto_assign_from_system(db, report, actor=actor) if "affected_urls" in patch: report.affected_urls = _clean_urls(patch["affected_urls"]) changed["affected_urls"] = "updated" if "vuln_type" in patch: vuln_type = patch["vuln_type"] or None if vuln_type is not None and vuln_type not in const.VULN_TYPES: raise ValidationError(f"unknown vulnerability type '{vuln_type}'") if vuln_type != report.vuln_type: changed["vuln_type"] = vuln_type report.vuln_type = vuln_type if "suggested_severity" in patch: suggested = patch["suggested_severity"] or None if suggested is not None: await cfg.get_tier(db, suggested) report.suggested_severity = suggested changed["suggested_severity"] = suggested if "reported_at" in patch and patch["reported_at"]: report.reported_at = patch["reported_at"] changed["reported_at"] = patch["reported_at"].isoformat() if "source" in patch and patch["source"]: if patch["source"] not in const.SOURCES: raise ValidationError(f"unknown source '{patch['source']}'") report.source = patch["source"] changed["source"] = patch["source"] if changed: await add_event( db, report=report, event_type=const.EVENT_UPDATED, actor=actor, meta={"fields": sorted(changed.keys())}, ) await db.flush() return report async def soft_delete_report( db: AsyncSession, report: BountyReport, *, actor: IdentityPerson | None, reason: str | None, ) -> None: """Hide a mis-keyed report. The ``code`` is never recycled and the row stays queryable for audit.""" if report.payment_status == const.PAYMENT_PAID: raise AppException( code="BOUNTY_REPORT_PAID", message="已付款的报告不能删除 —— 它是这笔支出的凭证。", status_code=409, ) report.deleted_at = _now() await add_event( db, report=report, event_type=const.EVENT_UPDATED, actor=actor, body=reason, meta={"deleted": True}, ) await db.flush() # ════════════════════════════════════════════════════════════════════ # Reports — reads # ════════════════════════════════════════════════════════════════════ async def get_report( db: AsyncSession, report_id: uuid.UUID, *, include_deleted: bool = False, ) -> BountyReport: stmt = select(BountyReport).where(BountyReport.id == report_id) if not include_deleted: stmt = stmt.where(BountyReport.deleted_at.is_(None)) row = (await db.execute(stmt)).scalar_one_or_none() if row is None: raise NotFoundError("bounty report not found") return row async def get_report_by_code(db: AsyncSession, code: str) -> BountyReport: row = ( await db.execute( select(BountyReport).where( BountyReport.code == code.strip().upper(), BountyReport.deleted_at.is_(None), ), ) ).scalar_one_or_none() if row is None: raise NotFoundError("bounty report not found") return row def _apply_report_filters( stmt: Select[Any], *, statuses: list[str] | None = None, severities: list[str] | None = None, remediation_statuses: list[str] | None = None, payment_statuses: list[str] | None = None, assignee_person_id: uuid.UUID | None = None, researcher_id: uuid.UUID | None = None, fiscal_year: int | None = None, vuln_type: str | None = None, system_id: uuid.UUID | None = None, department: str | None = None, unassigned: bool | None = None, ungraded: bool | None = None, never_replied: bool | None = None, overdue_only: bool = False, search: str | None = None, reported_from: datetime | None = None, reported_to: datetime | None = None, ) -> Select[Any]: if statuses: stmt = stmt.where(BountyReport.status.in_(statuses)) if severities: stmt = stmt.where(BountyReport.severity.in_(severities)) if remediation_statuses: stmt = stmt.where(BountyReport.remediation_status.in_(remediation_statuses)) if payment_statuses: stmt = stmt.where(BountyReport.payment_status.in_(payment_statuses)) if assignee_person_id is not None: stmt = stmt.where(BountyReport.assignee_person_id == assignee_person_id) if researcher_id is not None: stmt = stmt.where(BountyReport.researcher_id == researcher_id) if fiscal_year is not None: stmt = stmt.where(BountyReport.fiscal_year == fiscal_year) if vuln_type: stmt = stmt.where(BountyReport.vuln_type == vuln_type) if department: stmt = stmt.where(BountyReport.assignee_department == department) if system_id is not None: stmt = stmt.where(BountyReport.system_id == system_id) if unassigned is True: stmt = stmt.where(BountyReport.assignee_person_id.is_(None)) elif unassigned is False: stmt = stmt.where(BountyReport.assignee_person_id.is_not(None)) # A real predicate rather than severity="__none__": an IN-list filter # with a sentinel value matches nothing, which on screen is # indistinguishable from a filter that is broken. if ungraded is True: stmt = stmt.where(BountyReport.severity.is_(None)) elif ungraded is False: stmt = stmt.where(BountyReport.severity.is_not(None)) # 118 open reports have never had an answer, the oldest for 295 days. # Bulk replying was ruled out, so the queue that works through them one # at a time needs to be able to ask for exactly this set. if never_replied is True: stmt = stmt.where( or_( BountyReport.reply_count == 0, BountyReport.reply_count.is_(None), ), ) elif never_replied is False: stmt = stmt.where(BountyReport.reply_count > 0) if overdue_only: stmt = stmt.where( BountyReport.fix_due_date.is_not(None), BountyReport.fix_due_date < date.today(), BountyReport.remediation_status.not_in(tuple(const.REMEDIATION_SETTLED)), ) if reported_from is not None: stmt = stmt.where(BountyReport.reported_at >= reported_from) if reported_to is not None: stmt = stmt.where(BountyReport.reported_at <= reported_to) if search: needle = f"%{search.strip()}%" # The researcher lives on another table, reached by subquery so # that ``_apply_report_filters`` stays join-free. (A join would be # correct on the count too — ``researcher_id`` is many-to-one, so # it cannot duplicate rows — but every filter here composes onto # one statement, and a join added by one branch is a trap for the # next one that wants the same table.) # # Searching the researcher is not a widening: the list endpoint # already serialises reports with ``include_sensitive=True``, so # anyone who can read the list can already see these addresses. # Without it, "which reports did this person send us?" needed the # researcher tab and a copied UUID. # Match by any address of the person, not just the row that # literally matched: once two addresses are recorded as one person, # searching either one has to find everything they sent, or the # merge is invisible where it matters most. direct = select(BountyResearcher.id).where( or_( BountyResearcher.email.like(needle), BountyResearcher.display_name.like(needle), BountyResearcher.handle.like(needle), ), ).subquery() canonical_ids = select( func.coalesce( BountyResearcher.merged_into_id, BountyResearcher.id, ), ).where(BountyResearcher.id.in_(select(direct.c.id))) researcher_match = select(BountyResearcher.id).where( or_( BountyResearcher.id.in_(select(direct.c.id)), BountyResearcher.id.in_(canonical_ids), BountyResearcher.merged_into_id.in_(canonical_ids), ), ) stmt = stmt.where( or_( BountyReport.code.like(needle), BountyReport.title.like(needle), BountyReport.affected_system.like(needle), BountyReport.cve_id.like(needle), BountyReport.assignee_name.like(needle), BountyReport.researcher_id.in_(researcher_match), ), ) return stmt async def list_reports( db: AsyncSession, *, page: int = 1, page_size: int = 20, sort: str = "reported_at", order: str = "desc", **filters: Any, ) -> tuple[list[BountyReport], int]: base = select(BountyReport).where(BountyReport.deleted_at.is_(None)) base = _apply_report_filters(base, **filters) count_stmt = select(func.count()).select_from(base.subquery()) total = (await db.execute(count_stmt)).scalar_one() sort_col = { "reported_at": BountyReport.reported_at, "created_at": BountyReport.created_at, "updated_at": BountyReport.updated_at, "code": BountyReport.code, "severity": BountyReport.severity, "reward_amount_usd": BountyReport.reward_amount_usd, "fix_due_date": BountyReport.fix_due_date, }.get(sort, BountyReport.reported_at) direction = sort_col.desc() if order.lower() != "asc" else sort_col.asc() stmt = ( base.order_by(direction, BountyReport.code.desc()) .offset(max(0, (page - 1)) * page_size) .limit(page_size) ) rows = list((await db.execute(stmt)).scalars().all()) return rows, int(total) async def load_report_refs( db: AsyncSession, reports: list[BountyReport], ) -> tuple[dict[uuid.UUID, BountyResearcher], dict[str, BountySeverityTier]]: """Batch-load the researcher rows + tier map a list serialisation needs, so the list endpoint is 3 queries regardless of page size.""" tiers = await cfg.tier_map(db) ids = {r.researcher_id for r in reports} researchers: dict[uuid.UUID, BountyResearcher] = {} if ids: rows = ( await db.execute( select(BountyResearcher).where(BountyResearcher.id.in_(ids)), ) ).scalars().all() researchers = {r.id: r for r in rows} return researchers, tiers # ════════════════════════════════════════════════════════════════════ # Assignment # ════════════════════════════════════════════════════════════════════ async def auto_assign_from_system( db: AsyncSession, report: BountyReport, *, actor: IdentityPerson | None, ) -> bool: """Route a report to its system's owner without asking anyone. 「系统都有负责人了,为什么还需要指派一下」 — the register exists to answer "who is accountable for this system", so once a report is linked to a system that has one, the assignment carries no judgement. Making the security desk press 指派 on every report was a click that could only ever have one outcome, and 15 reports sat in 待指派 because of it. Deliberately does NOT notify. The push to the named owner stays an explicit act at the router (``notify=true``) — auto-routing 100 backfilled reports would have fired 100 unrequested DMs at employees, which the platform does not do. This records WHO owns it so the report leaves the 待指派 column; the nudge is still somebody's decision. Skips a report that already has an assignee: a human's routing decision outranks the register's default. """ if report.assignee_person_id is not None or report.system_id is None: return False if report.deleted_at is not None: return False from app.core.coordinators import ( # noqa: PLC0415 system_registry_coordinator as registry, ) owner_id = await registry.suggested_owner(db, report.system_id) if owner_id is None: # No owner, or only a departed one. Leaving it unassigned is the # honest state — it stays visible in 待指派 and in the gap panel. return False person = await load_assignable_person(db, owner_id) await assign_report( db, report, person=person, note=None, actor=actor, ) await add_event( db, report=report, event_type=const.EVENT_UPDATED, actor=actor, actor_kind="system", body=None, meta={"field": "assignee_person_id", "via": "system_owner", "system_id": str(report.system_id)}, visibility="internal", ) return True async def assign_report( db: AsyncSession, report: BountyReport, *, person: IdentityPerson, note: str | None, actor: IdentityPerson | None, ) -> BountyReport: """Point the report at a business owner and snapshot their attribution. Re-assigning an already-assigned report is allowed and logged as a reassignment (people change teams, guesses are wrong).""" if report.deleted_at is not None: raise ValidationError("cannot assign a deleted report") if report.status in const.TERMINAL_STATUSES: raise AppException( code="BOUNTY_REPORT_TERMINAL", message=( "该报告已结案,先「重新打开」再指派。" ), status_code=409, ) previous_id = report.assignee_person_id previous_name = report.assignee_name report.assignee_person_id = person.id report.assignee_name = person.display_name or person.work_email report.assignee_department = person.department report.assigned_at = _now() report.assigned_by = actor.id if actor else None # A new owner has not answered yet — clear the previous owner's # verdict so the console never shows a decision attributed to # someone who no longer owns the report. if previous_id != person.id: report.owner_decision = None report.owner_responded_at = None report.owner_note = None report.last_nudged_at = None if report.status == const.STATUS_NEW: report.status = const.STATUS_AWAITING_OWNER await add_event( db, report=report, event_type=( const.EVENT_ASSIGNED if previous_id is None else const.EVENT_ASSIGNED ), actor=actor, body=note, meta={ "assignee": report.assignee_name, "assignee_person_id": str(person.id), "department": person.department, "previous_assignee": previous_name, "reassigned": previous_id is not None and previous_id != person.id, }, visibility=const.VISIBILITY_SHARED, ) await db.flush() return report async def unassign_report( db: AsyncSession, report: BountyReport, *, actor: IdentityPerson | None, reason: str | None, ) -> BountyReport: if report.assignee_person_id is None: return report previous_name = report.assignee_name report.assignee_person_id = None report.assignee_name = None report.assignee_department = None report.assigned_at = None report.assigned_by = None report.owner_decision = None report.owner_responded_at = None report.owner_note = None report.last_nudged_at = None if report.status == const.STATUS_AWAITING_OWNER: report.status = const.STATUS_NEW await add_event( db, report=report, event_type=const.EVENT_UNASSIGNED, actor=actor, body=reason, meta={"previous_assignee": previous_name}, ) await db.flush() return report NUDGE_COOLDOWN = timedelta(hours=12) async def record_nudge( db: AsyncSession, report: BountyReport, *, actor: IdentityPerson | None, ) -> BountyReport: """Stamp a reminder. Throttled: an owner who has not answered is chased at most twice a day, no matter how often the button is clicked.""" if report.assignee_person_id is None: raise ValidationError("report has no assignee to remind") last = _naive_opt(report.last_nudged_at) if last is not None and _naive(_now()) - last < NUDGE_COOLDOWN: raise AppException( code="BOUNTY_NUDGE_THROTTLED", message="12 小时内已催办过一次,请稍后再试。", status_code=429, ) report.last_nudged_at = _now() await add_event( db, report=report, event_type=const.EVENT_NUDGED, actor=actor, meta={"assignee": report.assignee_name}, visibility=const.VISIBILITY_SHARED, ) await db.flush() return report # ════════════════════════════════════════════════════════════════════ # Owner response (the business owner's confirmation) # ════════════════════════════════════════════════════════════════════ async def record_owner_response( db: AsyncSession, report: BountyReport, *, decision: str, note: str | None, suggested_severity: str | None, fix_eta: date | None, remediation_status: str | None, actor: IdentityPerson | None, ) -> BountyReport: """The assignee's verdict. Deliberately does NOT set a final triage status: it records the decision and parks the report in ``in_discussion`` for the security team to rule on. Whether a finding is in scope and what it is worth is not the owner's call.""" if decision not in const.OWNER_DECISIONS: raise ValidationError(f"unknown owner decision '{decision}'") if report.status in const.TERMINAL_STATUSES: raise AppException( code="BOUNTY_REPORT_TERMINAL", message="该报告已结案,无法再提交确认结论。", status_code=409, ) if suggested_severity is not None: await cfg.get_tier(db, suggested_severity) if remediation_status is not None: if remediation_status not in const.OWNER_SETTABLE_REMEDIATION: raise ValidationError( f"'{remediation_status}' is not a remediation state the " "business owner may set", ) report.owner_decision = decision report.owner_note = (note or None) report.owner_responded_at = _now() if suggested_severity is not None: report.suggested_severity = suggested_severity if fix_eta is not None: report.fix_eta = fix_eta if decision == const.OWNER_DECISION_VALID: # The owner says it is real: remediation starts tracking even # before the security team finalises severity, so the fix clock # is not gated on a triage meeting. if remediation_status is not None: report.remediation_status = remediation_status elif report.remediation_status == const.REMEDIATION_NOT_APPLICABLE: report.remediation_status = const.REMEDIATION_PENDING if report.remediation_status == const.REMEDIATION_FIXED: report.fixed_at = report.fixed_at or _now() if report.status in (const.STATUS_NEW, const.STATUS_AWAITING_OWNER): report.status = const.STATUS_IN_DISCUSSION await add_event( db, report=report, event_type=const.EVENT_OWNER_RESPONDED, actor=actor, body=note, meta={ "decision": decision, "suggested_severity": suggested_severity, "fix_eta": fix_eta.isoformat() if fix_eta else None, "remediation_status": report.remediation_status, }, visibility=const.VISIBILITY_SHARED, ) await db.flush() return report # ════════════════════════════════════════════════════════════════════ # Triage / severity / remediation / money # ════════════════════════════════════════════════════════════════════ def _assert_status_transition(current: str, target: str) -> None: if current == target: return allowed = const.STATUS_TRANSITIONS.get(current, frozenset()) if target not in allowed: raise AppException( code="BOUNTY_INVALID_TRANSITION", message=( f"状态不能从 '{current}' 直接变为 '{target}';" f"可选:{', '.join(sorted(allowed)) or '(无)'}" ), status_code=409, ) async def _apply_duplicate_of( db: AsyncSession, report: BountyReport, *, duplicate_of_id: uuid.UUID | None, duplicate_note: str | None, ) -> None: """Record WHAT this report duplicates. Deliberately not mandatory. The original is frequently not another Helix row — both production cases point at an email that predates this domain — and a required FK would be answered with whatever row is nearest, which is worse than a sentence of prose. But one of the two must be present: "duplicate" with no referent at all is the dead end this replaces, and it is precisely what a researcher disputes. """ note = (duplicate_note or "").strip() or None if note and len(note) > 500: raise ValidationError("duplicate reference exceeds 500 characters") original: BountyReport | None = None if duplicate_of_id is not None: if duplicate_of_id == report.id: raise ValidationError("a report cannot duplicate itself") original = await get_report(db, duplicate_of_id) if original.duplicate_of_id == report.id: # A two-cycle makes "which one is the original" unanswerable # in both directions, which is the whole point of the field. raise AppException( code="BOUNTY_DUPLICATE_CYCLE", message=( f"{original.code} 已标记为重复于本报告,不能反过来再指向它。" ), status_code=409, ) if original is None and note is None: raise AppException( code="BOUNTY_DUPLICATE_REFERENCE_REQUIRED", message=( "标记为重复时必须说明重复于什么:选择一条已有报告," "或写明外部来源(内部 issue、commit、更早的邮件)。" ), status_code=409, ) report.duplicate_of_id = original.id if original else None report.duplicate_note = note async def set_status( db: AsyncSession, report: BountyReport, *, target: str, note: str | None, actor: IdentityPerson | None, duplicate_of_id: uuid.UUID | None = None, duplicate_note: str | None = None, ) -> BountyReport: if target not in const.STATUSES: raise ValidationError(f"unknown status '{target}'") _assert_status_transition(report.status, target) previous = report.status if previous == target: return report if target == const.STATUS_CONFIRMED and _decided_severity(report) is None: raise AppException( code="BOUNTY_SEVERITY_REQUIRED", message="确认为有效漏洞前必须先定级(级别决定赏金额度与修复时限)。", status_code=409, ) if target == const.STATUS_CLOSED: _assert_closable(report) if target == const.STATUS_DUPLICATE: await _apply_duplicate_of( db, report, duplicate_of_id=duplicate_of_id, duplicate_note=duplicate_note, ) else: # Leaving the duplicate verdict drops the pointer with it — a # stale "duplicate of BB-x" on a report we later accepted reads # as a contradiction, and it is the kind of thing a researcher # quotes back. report.duplicate_of_id = None report.duplicate_note = None report.status = target if target == const.STATUS_CONFIRMED and ( report.remediation_status == const.REMEDIATION_NOT_APPLICABLE ): # Accepting a finding starts the fix clock, symmetrically with the # owner's "yes it is real" path. Without this, a report the # security team confirms directly — after a phone call, or because # no business owner was needed — keeps ``not_applicable``, which is # in ``REMEDIATION_SETTLED``: its SLA deadline is stamped but the # row is invisible to the overdue counter and to every fix # worklist. Caught 2026-08-24 while writing the SLA report. report.remediation_status = const.REMEDIATION_PENDING event_type = const.EVENT_STATUS_CHANGED if target == const.STATUS_CLOSED: report.closed_at = _now() event_type = const.EVENT_CLOSED else: report.closed_at = None if previous in const.TERMINAL_STATUSES and target not in const.TERMINAL_STATUSES: event_type = const.EVENT_REOPENED # Leaving the "valid" lane means no money and no fix is owed. Money # already paid is never silently reset — that would erase a real # expense from the budget report. if target in ( const.STATUS_FALSE_POSITIVE, const.STATUS_OUT_OF_SCOPE, const.STATUS_DUPLICATE, ): if report.payment_status in const.PAYMENT_COMMITTED: report.payment_status = const.PAYMENT_NOT_APPLICABLE report.payment_requested_at = None if report.remediation_status in ( const.REMEDIATION_PENDING, const.REMEDIATION_IN_PROGRESS, ): report.remediation_status = const.REMEDIATION_NOT_APPLICABLE meta: dict[str, Any] = {"from": previous, "to": target} if target == const.STATUS_DUPLICATE: # The referent goes on the timeline entry too, not just the # column: "why was mine rejected" is asked months later, and the # answer should sit at the moment the decision was made. original = ( await get_report(db, report.duplicate_of_id) if report.duplicate_of_id else None ) meta["duplicate_of_code"] = original.code if original else None meta["duplicate_note"] = report.duplicate_note await add_event( db, report=report, event_type=event_type, actor=actor, body=note, meta=meta, visibility=const.VISIBILITY_SHARED, ) await db.flush() return report def _assert_closable(report: BountyReport) -> None: """Closing must not strand money. A confirmed report with a reward still pending is a promise we made to a researcher; the operator has to either pay it or explicitly waive it first.""" if ( report.status == const.STATUS_CONFIRMED and report.payment_status in const.PAYMENT_COMMITTED ): raise AppException( code="BOUNTY_PAYMENT_OPEN", message=( "该报告的赏金还在「待付款/付款中」,不能结案:请先标记已付款," "或明确标记为不发奖(waived)。" ), status_code=409, ) def _decided_severity(report: BountyReport) -> str | None: """Alias of ``const.graded_severity`` — see there for the rule. Kept as a name because two call sites below read better with it, but the rule itself must exist once: it now also decides what an owner and a lead are shown, and a copy of it here drifted from the copy in the serializer the first time that display question came up. """ return const.graded_severity(report) async def set_severity( db: AsyncSession, report: BountyReport, *, severity: str, note: str | None, apply_default_reward: bool, actor: IdentityPerson | None, ) -> BountyReport: """Final severity. Stamps the remediation SLA deadline and, unless told otherwise, pulls the tier's default reward onto the report.""" tier = await cfg.get_tier(db, severity) if not tier.enabled and report.severity != severity: raise ValidationError( f"severity tier '{severity}' is disabled and cannot be assigned", ) previous = report.severity report.severity = severity # A person picked it — including when they pick the same level the # report arrived with. That is a decision, not the default standing. report.severity_is_default = False # SLA deadline is anchored on when the report arrived, not on when we # got around to triaging it — otherwise a slow triage silently buys # the fix more time. if tier.fix_sla_days is not None: anchor = _naive_opt(report.reported_at) or _naive(_now()) report.fix_due_date = ( anchor + timedelta(days=tier.fix_sla_days) ).date() else: report.fix_due_date = None if apply_default_reward: await _set_reward( db, report, amount=Decimal(str(tier.default_reward_usd)), is_override=False, actor=actor, note=None, tier=tier, ) await add_event( db, report=report, event_type=const.EVENT_SEVERITY_SET, actor=actor, body=note, meta={ "from": previous, "to": severity, "fix_due_date": ( report.fix_due_date.isoformat() if report.fix_due_date else None ), }, visibility=const.VISIBILITY_SHARED, ) await db.flush() return report async def set_remediation( db: AsyncSession, report: BountyReport, *, target: str, note: str | None, fix_eta: date | None, actor: IdentityPerson | None, owner_scope: bool = False, ) -> BountyReport: if target not in const.REMEDIATION_STATUSES: raise ValidationError(f"unknown remediation status '{target}'") if owner_scope and target not in const.OWNER_SETTABLE_REMEDIATION: raise ValidationError( f"'{target}' is not a remediation state the business owner may set", ) previous = report.remediation_status report.remediation_status = target if fix_eta is not None: report.fix_eta = fix_eta if target == const.REMEDIATION_FIXED: report.fixed_at = report.fixed_at or _now() else: # Re-opening a fix clears the completion stamp, otherwise the # "average time to fix" metric keeps counting the first attempt. report.fixed_at = None await add_event( db, report=report, event_type=const.EVENT_REMEDIATION_CHANGED, actor=actor, body=note, meta={ "from": previous, "to": target, "fix_eta": fix_eta.isoformat() if fix_eta else None, }, visibility=const.VISIBILITY_SHARED, ) await db.flush() return report async def _set_reward( db: AsyncSession, report: BountyReport, *, amount: Decimal, is_override: bool, actor: IdentityPerson | None, note: str | None, tier: BountySeverityTier | None, ) -> None: if amount < 0: raise ValidationError("reward must not be negative") if amount > const.MAX_REWARD_USD: raise ValidationError( f"reward exceeds the {const.MAX_REWARD_USD} USD sanity cap", ) if ( tier is not None and tier.max_reward_usd is not None and amount > Decimal(str(tier.max_reward_usd)) ): raise AppException( code="BOUNTY_REWARD_ABOVE_TIER_CAP", message=( f"金额 {amount} USD 超过「{tier.label_zh}」级别的上限 " f"{tier.max_reward_usd} USD;请调整金额或先改级别配置。" ), status_code=409, ) previous = report.reward_amount_usd report.reward_amount_usd = amount report.reward_is_override = is_override report.reward_decided_at = _now() if report.fiscal_year is None: report.fiscal_year = await cfg.current_fy(db) # A zero reward is not a pending payment — it is nothing owed. if amount == 0: if report.payment_status in const.PAYMENT_COMMITTED: report.payment_status = const.PAYMENT_NOT_APPLICABLE report.payment_requested_at = None elif report.payment_status == const.PAYMENT_NOT_APPLICABLE: report.payment_status = const.PAYMENT_PENDING await add_event( db, report=report, event_type=const.EVENT_REWARD_SET, actor=actor, body=note, meta={ "from": str(previous) if previous is not None else None, "to": str(amount), "is_override": is_override, "fiscal_year": report.fiscal_year, }, visibility=const.VISIBILITY_INTERNAL, ) async def set_budget_track( db: AsyncSession, report: BountyReport, *, track: str, note: str | None, actor: IdentityPerson | None, ) -> BountyReport: """Move a report between the two funding pots. Allowed after payment as well: which budget an already-settled reward was charged to is a bookkeeping fact that finance sometimes corrects, and refusing the edit would leave the ledger wrong forever. The move is logged either way. """ if track not in const.BUDGET_TRACKS: raise ValidationError(f"unknown budget track '{track}'") previous = report.budget_track if previous == track: return report report.budget_track = track await add_event( db, report=report, event_type=const.EVENT_UPDATED, actor=actor, body=note, meta={ "field": "budget_track", "from": previous, "to": track, "amount_usd": ( str(report.reward_amount_usd) if report.reward_amount_usd is not None else None ), "already_paid": report.payment_status == const.PAYMENT_PAID, }, visibility=const.VISIBILITY_INTERNAL, ) await db.flush() return report async def set_reward( db: AsyncSession, report: BountyReport, *, amount: Decimal, note: str | None, actor: IdentityPerson | None, ) -> BountyReport: """Set the payout explicitly. ``reward_is_override`` records whether the figure differs from the severity tier's default, so the budget report can explain a number that does not match the ladder.""" decided = _decided_severity(report) if decided is None: raise AppException( code="BOUNTY_SEVERITY_REQUIRED", message="请先定级,再设置赏金金额。", status_code=409, ) if report.payment_status == const.PAYMENT_PAID: raise AppException( code="BOUNTY_ALREADY_PAID", message="已付款的报告不能改金额;如需更正请在付款备注中记录并联系财务。", status_code=409, ) tier = await cfg.get_tier(db, decided) is_override = amount != Decimal(str(tier.default_reward_usd)) await _set_reward( db, report, amount=amount, is_override=is_override, actor=actor, note=note, tier=tier, ) await db.flush() return report async def set_payment_status( db: AsyncSession, report: BountyReport, *, target: str, reference: str | None, note: str | None, paid_at: datetime | None, actor: IdentityPerson | None, ) -> BountyReport: if target not in const.PAYMENT_STATUSES: raise ValidationError(f"unknown payment status '{target}'") previous = report.payment_status if previous != target: allowed = const.PAYMENT_TRANSITIONS.get(previous, frozenset()) if target not in allowed: raise AppException( code="BOUNTY_INVALID_PAYMENT_TRANSITION", message=( f"付款状态不能从 '{previous}' 直接变为 '{target}';" f"可选:{', '.join(sorted(allowed)) or '(无)'}" ), status_code=409, ) if target in (const.PAYMENT_REQUESTED, const.PAYMENT_PAID): if report.status != const.STATUS_CONFIRMED: raise AppException( code="BOUNTY_NOT_CONFIRMED", message="只有已确认为有效漏洞的报告才能发起付款。", status_code=409, ) if not report.reward_amount_usd or Decimal( str(report.reward_amount_usd), ) <= 0: raise AppException( code="BOUNTY_REWARD_REQUIRED", message="请先设置赏金金额(>0),再发起付款。", status_code=409, ) if report.fiscal_year is None: report.fiscal_year = await cfg.current_fy(db) report.payment_status = target if target == const.PAYMENT_REQUESTED: report.payment_requested_at = report.payment_requested_at or _now() report.paid_at = None elif target == const.PAYMENT_PAID: report.payment_requested_at = report.payment_requested_at or _now() report.paid_at = paid_at or _now() elif target == const.PAYMENT_WAIVED: report.payment_requested_at = None report.paid_at = None else: report.payment_requested_at = None report.paid_at = None if reference is not None: report.payment_reference = reference.strip() or None if note is not None: report.payment_note = note.strip() or None await add_event( db, report=report, event_type=const.EVENT_PAYMENT_CHANGED, actor=actor, body=note, meta={ "from": previous, "to": target, "amount_usd": ( str(report.reward_amount_usd) if report.reward_amount_usd is not None else None ), "reference": report.payment_reference, "fiscal_year": report.fiscal_year, }, visibility=const.VISIBILITY_INTERNAL, ) await db.flush() return report # ════════════════════════════════════════════════════════════════════ # Correspondence # ════════════════════════════════════════════════════════════════════ async def add_comment( db: AsyncSession, report: BountyReport, *, body: str, visibility: str, actor: IdentityPerson | None, ) -> BountyReportEvent: text = (body or "").strip() if not text: raise ValidationError("comment body is required") if len(text) > const.MAX_COMMENT_CHARS: raise ValidationError( f"comment exceeds {const.MAX_COMMENT_CHARS} characters", ) if visibility not in const.VISIBILITIES: raise ValidationError(f"unknown visibility '{visibility}'") return await add_event( db, report=report, event_type=const.EVENT_COMMENT, actor=actor, body=text, visibility=visibility, ) def _assert_money_letter_has_money( *, template: BountyReplyTemplate, report: BountyReport, amount_override: Decimal | None, rendered_body: str, ) -> None: """Refuse to produce a reward letter with no reward in it. The renderer reports a missing token as advisory, which was right for a signature line and wrong for money: the operator copies the preview into their mail client, so an empty ``{amount_usd}`` becomes a letter that says "your reward for this report is USD ." — and the single most common real complaint about this feature was that the reply never mentions the money at all. Checked against the TEMPLATE's declared tokens, not the rendered text: once substitution has happened the amount is indistinguishable from prose, and an operator who typed the figure by hand into a free-form body must not be blocked. """ declared = declared_tokens(template.subject, template.body) if "amount_usd" not in declared: return amount = amount_override if amount_override is not None else report.reward_amount_usd if amount is not None: return # A hand-edited body that no longer references the token is the # operator having taken responsibility for the wording. if "{amount_usd}" not in (rendered_body or "") and "USD" not in ( rendered_body or "" ): return raise AppException( code="BOUNTY_REPLY_AMOUNT_REQUIRED", message=( "这封信要写明赏金金额,但这条报告还没有定赏金。" "先定级(档位会带出默认金额)或直接在「赏金」里填金额,再生成回信。" ), status_code=422, ) async def preview_reply( db: AsyncSession, report: BountyReport, *, template_key: str, amount_override: Decimal | None = None, subject_override: str | None = None, body_override: str | None = None, ) -> tuple[RenderedReply, BountyReplyTemplate]: """Render a reply without recording anything. ``*_override`` lets the console preview operator edits before they are logged, so what the operator copies is exactly what gets stored. """ template = await cfg.get_template(db, template_key) researcher = await get_researcher(db, report.researcher_id) tier = None if report.severity: tiers = await cfg.tier_map(db) tier = tiers.get(report.severity) original = ( await get_report(db, report.duplicate_of_id) if report.duplicate_of_id else None ) context = build_context( report=report, researcher=researcher, severity_label=tier.label_en if tier else None, signature=await cfg.get_signature(db), amount_usd=amount_override, duplicate_of_code=original.code if original else None, ) source_body = ( body_override if body_override is not None else template.body ) rendered = render( subject=subject_override if subject_override is not None else template.subject, body=source_body, context=context, ) _assert_money_letter_has_money( template=template, report=report, amount_override=amount_override, rendered_body=source_body, ) return rendered, template async def log_reply( db: AsyncSession, report: BountyReport, *, template_key: str | None, subject: str, body: str, actor: IdentityPerson | None, ) -> BountyReportEvent: """Record that a reply was sent to the researcher. Helix does not send the mail (decision 2026-08-24: the operator keeps the thread in their own mailbox), so this is an attestation — the operator confirms "I sent this text". Storing the verbatim body is the point: months later, "what exactly did we promise this person" must be answerable without digging through a personal inbox. """ text = (body or "").strip() if not text: raise ValidationError("reply body is required") if len(text) > const.MAX_RAW_REPORT_CHARS: raise ValidationError("reply body is too long") template: BountyReplyTemplate | None = None if template_key: template = await cfg.get_template(db, template_key) template.use_count = (template.use_count or 0) + 1 report.last_replied_at = _now() report.last_reply_template_key = template_key report.reply_count = (report.reply_count or 0) + 1 event = await add_event( db, report=report, event_type=const.EVENT_REPLY_LOGGED, actor=actor, body=text, meta={ "template_key": template_key, "template_name": template.name_zh if template else None, "subject": (subject or "").strip(), "direction": "outbound", }, visibility=const.VISIBILITY_INTERNAL, ) await db.flush() return event async def log_inbound( db: AsyncSession, report: BountyReport, *, body: str, received_at: datetime | None, actor: IdentityPerson | None, ) -> BountyReportEvent: """Record a follow-up mail FROM the researcher, pasted in by the operator, so the thread in Helix stays a complete record.""" text = (body or "").strip() if not text: raise ValidationError("message body is required") if len(text) > const.MAX_RAW_REPORT_CHARS: raise ValidationError("message body is too long") event = await add_event( db, report=report, event_type=const.EVENT_INBOUND_LOGGED, actor=actor, body=text, meta={ "direction": "inbound", "received_at": (received_at or _now()).isoformat(), }, visibility=const.VISIBILITY_INTERNAL, ) await db.flush() return event # ════════════════════════════════════════════════════════════════════ # Console overview # ════════════════════════════════════════════════════════════════════ async def overview(db: AsyncSession) -> dict[str, Any]: """Counters for the console's KPI strip + the budget widget. Budget maths, and why ``committed`` exists: a burn number that counts only ``paid`` reads healthy right up to the moment finance settles six pending rewards. ``committed`` (pending + requested) is money we have already promised a researcher in writing, so the widget shows both and the warning threshold is measured on the sum. """ status_rows = ( await db.execute( select(BountyReport.status, func.count()) .where(BountyReport.deleted_at.is_(None)) .group_by(BountyReport.status), ) ).all() by_status = {row[0]: int(row[1]) for row in status_rows} severity_rows = ( await db.execute( select(BountyReport.severity, func.count()) .where( BountyReport.deleted_at.is_(None), BountyReport.severity.is_not(None), ) .group_by(BountyReport.severity), ) ).all() by_severity = {row[0]: int(row[1]) for row in severity_rows} remediation_rows = ( await db.execute( select(BountyReport.remediation_status, func.count()) .where(BountyReport.deleted_at.is_(None)) .group_by(BountyReport.remediation_status), ) ).all() by_remediation = {row[0]: int(row[1]) for row in remediation_rows} fiscal_year = await cfg.current_fy(db) budget = await cfg.budget_for_fy(db, fiscal_year) # both pots combined money_rows = ( await db.execute( select( BountyReport.payment_status, func.coalesce(func.sum(BountyReport.reward_amount_usd), 0), func.count(), ) .where( BountyReport.deleted_at.is_(None), BountyReport.fiscal_year == fiscal_year, ) .group_by(BountyReport.payment_status), ) ).all() paid = Decimal("0") committed = Decimal("0") paid_count = 0 committed_count = 0 for status, amount, count in money_rows: amount_dec = Decimal(str(amount or 0)) if status in const.PAYMENT_SETTLED_SPENT: paid += amount_dec paid_count += int(count) elif status in const.PAYMENT_COMMITTED: committed += amount_dec committed_count += int(count) total_committed = paid + committed ratio = ( float(total_committed / budget) if budget and budget > 0 else (1.0 if total_committed > 0 else 0.0) ) overdue = ( await db.execute( select(func.count()) .select_from(BountyReport) .where( BountyReport.deleted_at.is_(None), BountyReport.fix_due_date.is_not(None), BountyReport.fix_due_date < date.today(), BountyReport.remediation_status.not_in( tuple(const.REMEDIATION_SETTLED), ), ), ) ).scalar_one() awaiting_owner = by_status.get(const.STATUS_AWAITING_OWNER, 0) unassigned = ( await db.execute( select(func.count()) .select_from(BountyReport) .where( BountyReport.deleted_at.is_(None), BountyReport.assignee_person_id.is_(None), BountyReport.status.in_(tuple(const.OPEN_STATUSES)), ), ) ).scalar_one() total_reports = sum(by_status.values()) decided = sum( by_status.get(s, 0) for s in ( const.STATUS_CONFIRMED, const.STATUS_FALSE_POSITIVE, const.STATUS_OUT_OF_SCOPE, const.STATUS_DUPLICATE, const.STATUS_CLOSED, ) ) # "Was this report valid" survives closing, which ``status`` does not # (``closed`` covers both a settled valid finding and a filed # rejection). A decided severity is the durable marker: only the # valid lane ever gets one. confirmed = ( await db.execute( select(func.count()) .select_from(BountyReport) .where( BountyReport.deleted_at.is_(None), BountyReport.severity.is_not(None), BountyReport.status.in_( (const.STATUS_CONFIRMED, const.STATUS_CLOSED), ), ), ) ).scalar_one() return { "total_reports": total_reports, "open_reports": sum( by_status.get(s, 0) for s in const.OPEN_STATUSES ), "by_status": by_status, "by_severity": by_severity, "by_remediation": by_remediation, "awaiting_owner": int(awaiting_owner), "unassigned_open": int(unassigned), "overdue_remediation": int(overdue), "decided_reports": decided, "confirmed_reports": int(confirmed), "budget": { "fiscal_year": fiscal_year, "fiscal_year_label": await cfg.fy_label(db, fiscal_year), "budget_usd": budget, "paid_usd": paid, "committed_usd": committed, "total_usd": total_committed, "remaining_usd": budget - total_committed, "burn_ratio": round(ratio, 4), "paid_count": paid_count, "committed_count": committed_count, "level": _budget_level(ratio), }, } def _budget_level(ratio: float) -> str: if ratio >= const.BUDGET_OVER_RATIO: return "over" if ratio >= const.BUDGET_WARN_RATIO: return "warn" return "ok" async def distinct_departments(db: AsyncSession) -> list[str]: """Departments that actually appear on reports — the filter's option list must come from the reports, not from the org chart, or it offers 150 departments with zero findings.""" rows = ( await db.execute( select(BountyReport.assignee_department) .where( BountyReport.deleted_at.is_(None), BountyReport.assignee_department.is_not(None), ) .group_by(BountyReport.assignee_department) .order_by(BountyReport.assignee_department), ) ).all() return [row[0] for row in rows if row[0]] # ════════════════════════════════════════════════════════════════════ # Backfill operations — internal fields only, in bulk # ════════════════════════════════════════════════════════════════════ # # Added 2026-08-25. Bulk is allowed here and deliberately NOT allowed for # replies: grading a report and linking it to a system are internal # bookkeeping that a human can see and correct, whereas a reply leaves # the building. One mis-worded letter sent to thirty researchers cannot # be corrected. # # Every one of these writes the report's own timeline, so a bulk action # is as traceable afterwards as thirty single ones would have been. async def bulk_set_severity( db: AsyncSession, *, report_ids: list[uuid.UUID], severity_by_id: dict[uuid.UUID, str], actor: IdentityPerson | None, ) -> dict[str, Any]: """Grade many reports at once. Each grade must be supplied per report — there is no single severity for a batch, and accepting one would be a footgun disguised as a convenience.""" applied: list[str] = [] skipped: list[dict[str, str]] = [] for report_id in report_ids: severity = severity_by_id.get(report_id) if not severity: continue report = await get_report(db, report_id) if report.severity: # Never overwrite a human grade with a suggestion. skipped.append({"code": report.code, "reason": "already_graded"}) continue # SAVEPOINT per row. ``set_severity`` flushes (it pulls the tier's # default reward and writes an event), and a failed flush # deactivates the WHOLE transaction — so catching and continuing # without isolation would turn one bad row into every later row # failing with PendingRollbackError while the caller was told they # were merely "skipped". code = report.code async with best_effort(db, label=f"grade {code}") as outcome: await set_severity( db, report, severity=severity, note=None, apply_default_reward=True, actor=actor, ) if not outcome.ok: skipped.append({ "code": code, "reason": getattr(outcome.error, "code", "grade_failed"), }) continue applied.append(code) return {"applied": applied, "skipped": skipped} async def bulk_link_system( db: AsyncSession, *, system_by_id: dict[uuid.UUID, uuid.UUID], actor: IdentityPerson | None, ) -> dict[str, Any]: """Link many reports to registry systems. Skips reports already linked so a re-run cannot silently re-point somebody's earlier decision.""" applied: list[str] = [] skipped: list[dict[str, str]] = [] for report_id, system_id in system_by_id.items(): report = await get_report(db, report_id) if report.system_id is not None: skipped.append({"code": report.code, "reason": "already_linked"}) continue code = report.code async with best_effort(db, label=f"link {code}") as outcome: report.system_id = system_id report.system_not_applicable = False await db.flush() await add_event( db, report=report, event_type=const.EVENT_UPDATED, actor=actor, actor_kind="user" if actor else "system", body=None, meta={"field": "system_id", "to": str(system_id), "via": "backfill_bulk"}, visibility="internal", ) # Linking is the moment "who answers for this" becomes known — # the operator should not have to make a second pass to route # what they just attributed. await auto_assign_from_system(db, report, actor=actor) if not outcome.ok: skipped.append({"code": code, "reason": "link_failed"}) continue applied.append(code) return {"applied": applied, "skipped": skipped} async def bulk_mark_no_system( db: AsyncSession, *, report_ids: list[uuid.UUID], actor: IdentityPerson | None, note: str | None = None, ) -> dict[str, Any]: """Record that somebody looked and no internal system applies. This is the only way a report leaves the "unlinked" gap without being linked, which is why it writes an event carrying the operator's note: six months from now the question will be "who decided this was an open-source finding", and a boolean cannot answer it. """ applied: list[str] = [] skipped: list[dict[str, str]] = [] for report_id in report_ids: report = await get_report(db, report_id) if report.system_id is not None: skipped.append({"code": report.code, "reason": "already_linked"}) continue if report.system_not_applicable: skipped.append({"code": report.code, "reason": "already_marked"}) continue code = report.code async with best_effort(db, label=f"mark no-system {code}") as outcome: report.system_not_applicable = True await db.flush() await add_event( db, report=report, event_type=const.EVENT_UPDATED, actor=actor, actor_kind="user" if actor else "system", body=note or None, meta={"field": "system_not_applicable", "to": True, "via": "backfill_bulk"}, visibility="internal", ) if not outcome.ok: skipped.append({"code": code, "reason": "mark_failed"}) continue applied.append(code) return {"applied": applied, "skipped": skipped} async def reports_needing_backfill( db: AsyncSession, *, kind: str, limit: int = 50, ) -> list[BountyReport]: """Open reports missing a grade (``kind='severity'``) or a system link (``kind='system'``), oldest first — the oldest is where the programme's credibility is being spent.""" stmt = select(BountyReport).where( BountyReport.deleted_at.is_(None), BountyReport.status.in_(tuple(const.OPEN_STATUSES)), ) if kind == "severity": stmt = stmt.where(BountyReport.severity.is_(None)) else: stmt = stmt.where( BountyReport.system_id.is_(None), BountyReport.system_not_applicable.is_(False), ) stmt = stmt.order_by(BountyReport.reported_at.asc()).limit(limit) return list((await db.execute(stmt)).scalars().all()) # ════════════════════════════════════════════════════════════════════ # Researcher identity — one person, several addresses # ════════════════════════════════════════════════════════════════════ # # Confirmed in production: suhaspandya55@ and suhaspandya124@ are one # person, counted as two. Every per-researcher figure was therefore wrong # for anyone who has ever used a second address. # # Reports are never re-pointed. The alias row keeps receiving them, so a # reply goes back to the address the mail arrived on and "which address # did this come in on" stays answerable. Only the aggregation resolves. async def alias_group( db: AsyncSession, researcher_id: uuid.UUID, ) -> list[uuid.UUID]: """Every researcher row that is the same person as this one. One hop in each direction, because ``merge_researchers`` refuses to make an alias an alias target — so this can never need recursion and a cycle cannot make it hang. """ row = await db.get(BountyResearcher, researcher_id) if row is None: return [researcher_id] canonical = row.merged_into_id or row.id aliases = ( await db.execute( select(BountyResearcher.id).where( BountyResearcher.merged_into_id == canonical, ), ) ).scalars().all() return [canonical, *aliases] async def merge_researchers( db: AsyncSession, *, source_id: uuid.UUID, target_id: uuid.UUID, ) -> BountyResearcher: """Record that ``source`` is another address of ``target``. Refused when it would create a chain or a cycle: an alias may not be a merge target, and a researcher that already has aliases may not itself become an alias. Both would force alias resolution to recurse, and a recursive resolve over operator-entered data is how a report page ends up hanging. """ if source_id == target_id: raise ValidationError("a researcher cannot be merged into itself") source = await get_researcher(db, source_id) target = await get_researcher(db, target_id) if target.merged_into_id is not None: raise AppException( code="BOUNTY_MERGE_TARGET_IS_ALIAS", message=( "目标本身已经是别名了。请合并到那个人的主记录上," "不要串成链。" ), status_code=409, ) existing_aliases = ( await db.execute( select(func.count()).select_from(BountyResearcher).where( BountyResearcher.merged_into_id == source.id, ), ) ).scalar_one() if existing_aliases: raise AppException( code="BOUNTY_MERGE_SOURCE_HAS_ALIASES", message=( f"这个记录本身已经有 {existing_aliases} 个别名指向它," "不能再把它并到别人名下。请反向操作。" ), status_code=409, ) if source.merged_into_id == target.id: return source source.merged_into_id = target.id # The canonical row inherits the earliest first-seen and the latest # last-seen, so "since when have we known this person" stays true. if source.first_report_at and ( target.first_report_at is None or source.first_report_at < target.first_report_at ): target.first_report_at = source.first_report_at if source.last_report_at and ( target.last_report_at is None or source.last_report_at > target.last_report_at ): target.last_report_at = source.last_report_at await db.flush() return source async def unmerge_researcher( db: AsyncSession, researcher_id: uuid.UUID, ) -> BountyResearcher: """Undo a merge. Operator-entered identity claims are wrong sometimes, and a one-way door on a guess is a trap.""" row = await get_researcher(db, researcher_id) row.merged_into_id = None await db.flush() return row async def assess_with_ai( db: AsyncSession, report: BountyReport, *, actor: IdentityPerson | None, lang: str | None = None, ) -> dict[str, Any] | None: """Run the triage read and store it on the report. Fills the empty STRUCTURED fields as a side effect — `vuln_type`, `affected_urls`, `reproduction` — because those are extraction, not judgement: they are in the researcher's own text and having them in columns is what makes the console searchable. It will not touch a field a person has already filled. What it deliberately does NOT do: set the severity, move the status, or influence the reply. The model's level lands in ``suggested_severity`` (advisory, already the column's meaning) and its verdict lands in ``ai_assessment`` where the page labels it as the model's opinion. An LLM writing into a ledger that authorises payments is not a feature. """ tiers = await cfg.tier_map(db) assessment = await bounty_ai.assess_report( db, title=report.title or "", raw_text=report.raw_report or "", severity_keys=[k for k, tier in tiers.items() if tier.enabled], vuln_types=list(const.VULN_TYPES), lang=lang, ) if assessment is None: return None report.ai_assessment = assessment report.ai_assessed_at = _now() # Extraction, not judgement — and only where nobody has filled it in. if not report.vuln_type and assessment.get("vuln_type"): report.vuln_type = assessment["vuln_type"] if not report.affected_urls and assessment.get("affected_urls"): report.affected_urls = assessment["affected_urls"][:10] if not report.reproduction and assessment.get("reproduction"): report.reproduction = assessment["reproduction"] if not report.suggested_severity and assessment.get("suggested_severity"): report.suggested_severity = assessment["suggested_severity"] await add_event( db, report=report, event_type=const.EVENT_UPDATED, actor=actor, actor_kind="system", body=None, meta={"field": "ai_assessment", "verdict": assessment["verdict"], "model": assessment.get("model")}, visibility="internal", ) await db.flush() return assessment async def list_attachments( db: AsyncSession, report_id: uuid.UUID, ) -> list[BountyReportAttachment]: """This report's attachments, real files before inline ones. Ordering is the whole point: one report in this corpus carries 90 parts, most of them signature logos, and a gallery that leads with those buries the proof-of-concept screenshot the researcher actually sent. Real attachments first, then inline, each oldest-first so a numbered series ("step1.png", "step2.png") keeps its order. """ return list( ( await db.execute( select(BountyReportAttachment) .where(BountyReportAttachment.report_id == report_id) .order_by( BountyReportAttachment.is_inline.asc(), BountyReportAttachment.imported_at.asc(), BountyReportAttachment.filename.asc(), ), ) ).scalars().all(), ) async def attachment_download_url( db: AsyncSession, *, report_id: uuid.UUID, attachment_id: uuid.UUID, ) -> str | None: """A short-lived presigned GET, or ``None`` if it is not on this report. ``report_id`` is not decoration: the caller has already established that the viewer may see THAT report, so scoping the lookup to it is what stops an attachment id from another report being signed. Returns ``None`` rather than raising so each surface can answer with its own shape (404 for the lead, whose endpoint must not confirm existence). """ from app.core import storage row = ( await db.execute( select(BountyReportAttachment).where( BountyReportAttachment.id == attachment_id, BountyReportAttachment.report_id == report_id, ), ) ).scalars().first() if row is None: return None # Ten minutes: long enough to click, short enough that a URL pasted # into a chat is dead before it is useful. The URL bypasses Helix auth # once minted — that is why the ACL happens above this call, not here. return await storage.presign_get(db, object_key=row.object_key, expires_in=600)