"""The business owner's narrow view of a report they were assigned. Authorization here is **self-scope**: the caller must be the report's ``assignee_person_id``. There is no role gate — being named on the report IS the authorization, exactly like the access-review self-attestation surface. That means every function in this module must re-check ownership against the caller rather than trusting a router-level filter. What the owner sees is deliberately smaller than what the console shows, but it includes the whole technical picture, because the owner is the person being asked whether the finding is real: title, internal summary, reproduction steps, affected system/URLs, severity, remediation state, the fix deadline, the researcher's own write-up, the model's read of it, the files the researcher attached, and the `shared` slice of the timeline. What they never see: * the researcher's email / name / country / notes, * the reward amount, the payment state, the fiscal year, * `internal` timeline events. That split is enforced by ``serialize_report(include_sensitive=False)`` and ``list_events(internal=False)``, not by the frontend. ``raw_report`` used to be on the withheld list, on the reasoning that it carries the submitter's signature. It was moved across on 2026-08-26: 「这个白帽子的正文没有什么保密的,尤其是是说哪里有 bug 的那种」. Withholding it left the confirmation page holding a title and a deadline, and the banking details that were the real hazard are stripped at import. """ from __future__ import annotations import uuid from datetime import date from typing import Any from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.core.coordinators import system_registry_coordinator as registry from app.core.exceptions import AppException, NotFoundError 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 BountyReport from app.domains.bounty.serializers import ( serialize_attachment, serialize_event, serialize_report, ) from app.identity.person_models import IdentityPerson #: Reports the owner still has something to do about. A `confirmed` #: report stays here while its fix is outstanding — the owner's job is #: not over when the security team accepts the finding. _MY_OPEN_STATUSES: tuple[str, ...] = ( const.STATUS_AWAITING_OWNER, const.STATUS_IN_DISCUSSION, const.STATUS_CONFIRMED, ) async def load_my_report( db: AsyncSession, report_id: uuid.UUID, *, person_id: uuid.UUID, ) -> BountyReport: """Fetch a report the caller owns, or 404/403. 404 (not 403) when the report exists but belongs to someone else would leak that the id is real; 403 with a clear message is the honest answer for a stale Feishu card after a reassignment, which is the common case. So: 404 for "no such report", 403 for "not yours". """ report = ( await db.execute( select(BountyReport).where( BountyReport.id == report_id, BountyReport.deleted_at.is_(None), ), ) ).scalar_one_or_none() if report is None: raise NotFoundError("bounty report not found") if report.assignee_person_id != person_id: raise AppException( code="BOUNTY_NOT_ASSIGNEE", message=( "这条漏洞报告已不再指派给你(可能已改派给其他同事)。" "如果你认为应该由你处理,请联系 IT/安全团队。" ), status_code=403, ) return report async def list_my_reports( db: AsyncSession, *, person_id: uuid.UUID, include_closed: bool = False, ) -> dict[str, Any]: """Everything assigned to the caller, newest first, with the counters the todo hub badge needs.""" stmt = ( select(BountyReport) .where( BountyReport.assignee_person_id == person_id, BountyReport.deleted_at.is_(None), ) .order_by(BountyReport.reported_at.desc(), BountyReport.code.desc()) ) if not include_closed: stmt = stmt.where(BountyReport.status.in_(_MY_OPEN_STATUSES)) rows = list((await db.execute(stmt)).scalars().all()) tiers = await cfg.tier_map(db) systems = await _system_names(db, [r.system_id for r in rows if r.system_id]) awaiting = sum( 1 for r in rows if r.status == const.STATUS_AWAITING_OWNER ) fixing = sum( 1 for r in rows if r.status == const.STATUS_CONFIRMED and r.remediation_status not in const.REMEDIATION_SETTLED ) overdue = sum( 1 for r in rows if r.fix_due_date is not None and r.fix_due_date < date.today() and r.remediation_status not in const.REMEDIATION_SETTLED ) return { "items": [ serialize_report( r, researcher=None, tiers=tiers, include_sensitive=False, system=systems.get(str(r.system_id)) if r.system_id else None, ) for r in rows ], "counts": { "total": len(rows), "awaiting_confirmation": awaiting, "fix_outstanding": fixing, "overdue": overdue, # What the red badge should read: work the owner has not # answered yet. An in-progress fix is tracked, not nagged. "actionable": awaiting, }, } async def _system_names( db: AsyncSession, system_ids: list[uuid.UUID], ) -> dict[str, dict[str, Any]]: """Name + exposure of the linked system, and nothing else. The owner needs to recognise WHICH system this is about. Handing them the registry's owner emails and budget pot on their own confirmation page would widen what a single report exposes for no operational gain — the same reasoning as ``include_sensitive`` above. """ if not system_ids: return {} described = await registry.describe_systems(db, system_ids) return { key: { "id": value["id"], "slug": value["slug"], "name": value["name"], "kind": value["kind"], "exposure": value["exposure"], } for key, value in described.items() } async def my_open_count(db: AsyncSession, *, person_id: uuid.UUID) -> int: """Cheap counter for the todo hub — one COUNT, no serialisation.""" return int( ( await db.execute( select(func.count()) .select_from(BountyReport) .where( BountyReport.assignee_person_id == person_id, BountyReport.deleted_at.is_(None), BountyReport.status == const.STATUS_AWAITING_OWNER, ), ) ).scalar_one(), ) async def my_report_detail( db: AsyncSession, report_id: uuid.UUID, *, person: IdentityPerson, ) -> dict[str, Any]: report = await load_my_report(db, report_id, person_id=person.id) tiers = await cfg.tier_map(db) systems = await _system_names( db, [report.system_id] if report.system_id else [], ) payload = serialize_report( report, researcher=None, tiers=tiers, include_sensitive=False, system=( systems.get(str(report.system_id)) if report.system_id else None ), ) payload["events"] = [ serialize_event(e) for e in await svc.list_events(db, report.id, internal=False) ] # What the researcher attached. For a good share of these reports the # screenshot or the recording IS the evidence, and the owner is the # one being asked whether the finding is real — 「有些白帽子给了附件 # 这个才是重要的」. The rows carry no object key; downloads go through # the per-attachment endpoint, which re-checks this same scoping. payload["attachments"] = [ serialize_attachment(a) for a in await svc.list_attachments(db, report.id) ] payload["can_respond"] = report.status not in const.TERMINAL_STATUSES payload["settable_remediation"] = sorted( const.OWNER_SETTABLE_REMEDIATION, ) payload["severity_options"] = [ { "key": tier.key, "label_zh": tier.label_zh, "label_en": tier.label_en, "label_ja": tier.label_ja, # Reward amounts are IT-only: the owner grades impact, not price. "color": tier.color, } for tier in sorted( (t for t in tiers.values() if t.enabled), key=lambda t: t.sort_order, ) ] return payload async def respond( db: AsyncSession, report_id: uuid.UUID, *, person: IdentityPerson, decision: str, note: str | None, suggested_severity: str | None, fix_eta: date | None, remediation_status: str | None, ) -> BountyReport: """The owner's verdict, submitted by the owner themselves.""" report = await load_my_report(db, report_id, person_id=person.id) return await svc.record_owner_response( db, report, decision=decision, note=note, suggested_severity=suggested_severity, fix_eta=fix_eta, remediation_status=remediation_status, actor=person, ) async def update_remediation( db: AsyncSession, report_id: uuid.UUID, *, person: IdentityPerson, remediation_status: str, fix_eta: date | None, note: str | None, ) -> BountyReport: """Owner-side fix progress. ``owner_scope=True`` refuses the risk decisions (`wont_fix` / `risk_accepted`) that need the security team.""" report = await load_my_report(db, report_id, person_id=person.id) return await svc.set_remediation( db, report, target=remediation_status, note=note, fix_eta=fix_eta, actor=person, owner_scope=True, ) async def add_comment( db: AsyncSession, report_id: uuid.UUID, *, person: IdentityPerson, body: str, ) -> dict[str, Any]: """A comment from the owner. Always ``shared`` — an owner cannot write into the security team's internal-only channel, and everything they type is meant for the operator anyway.""" report = await load_my_report(db, report_id, person_id=person.id) event = await svc.add_comment( db, report, body=body, visibility=const.VISIBILITY_SHARED, actor=person, ) return serialize_event(event)