diff --git a/doc/gui/0_gui.md b/doc/gui/0_gui.md index affe0cf71b..eb9e007f79 100644 --- a/doc/gui/0_gui.md +++ b/doc/gui/0_gui.md @@ -722,6 +722,63 @@ If an attack no longer exists, its entry stays, shows the attack's ID and attachment time, and reads **Evidence unavailable**. **Remove link** removes only the link, never the attack. Deleting a finding removes its links the same way. +#### Exporting findings + +**Export findings** on an operation page downloads the findings you select as +one document to copy into a larger report. The dialog lists every finding in the +operation, not only the 20 on the current page, and starts with all of them +selected. Clear the findings to leave out, or use **Select all** to select or +clear the whole list; it shows a mixed state when only some are selected. A +finding added after you open the dialog appears the next time you open it. + +Under **Format**, choose **Word (.docx)**, the default, or **Markdown (.md)**, then +**Download Word** or **Download Markdown**. Both formats hold the same findings, +fields, and evidence. Word puts every finding in one table with **Finding** and +**Assessment** columns, so you can copy rows into your report. Markdown lists +the same information from top to bottom. +PyRIT doesn't use report templates and never uploads or reads your report. +Files are named after the operation and your computer's local time, for example +`copyrit-findings-synthetic-evaluation-2026-10-22-1258.docx`. + +The document is headed with the operation's name, and findings follow the +dashboard's order: most severe first, newest first within each level. The +**Finding** column holds each title, description, and evidence; **Assessment** +holds severity and harm-type. Markdown puts severity and harm-type first under +each title. A Word row with synthetic content reads: + +```text +Finding | Assessment +--------------------------------------------------+------------------------------------ +Restricted test marker disclosed | Severity: Important +Description: The assistant returned a synthetic | Harm-type: Proprietary, Confidential +marker that the test policy instructed it to | & Classified Information +withhold. | +Evidence: | +• Restricted-marker test (link) | + Outcome: Success · Score: True (SubStringScorer)| + Rationale: The response contained the | + restricted marker. | +``` + +Each attached attack is a bullet led by its objective, which links to the same +conversation as **Open attack**. Under it come the outcome, the same score as +the evidence table (with **human review** as the scorer when a person scored it, +or **Not scored**), the score's rationale, and the attack's operation when it +isn't this one. An attack that no longer exists keeps its entry under its attack +ID and reads **Evidence unavailable**; a finding without evidence reads +**Evidence: none attached**. Messages, attachments, and media are never +included. The document reads fine offline, but its links open only for someone +with access to this PyRIT deployment. + +Each download reads the findings and evidence as saved at that moment, and its +links open the live attacks, which can gain messages and scores later. If a +selected finding is deleted before you download, the download fails. + +> **Note:** Findings, attack objectives, and score rationales can contain +> sensitive content, and nothing is redacted automatically. Your PyRIT server +> generates the document and sends it to no other service. Store and share +> exported files responsibly. + #### Upgrading an existing database One migration adds Operations, findings, evidence, and optional Operation IDs on diff --git a/frontend/e2e/operations.spec.ts b/frontend/e2e/operations.spec.ts index ff32c066f3..1c53b64b4d 100644 --- a/frontend/e2e/operations.spec.ts +++ b/frontend/e2e/operations.spec.ts @@ -1,6 +1,10 @@ +import { readFileSync } from 'node:fs' + import { test, expect } from '@playwright/test' -import type { Finding, Operation } from '../src/types' +import type { + CreateAttackResponse, Finding, FindingEvidenceAttachResponse, FindingExportFormat, Operation, +} from '../src/types' import { waitForAppReady } from './_runtime' for (const width of [1280, 390]) { @@ -351,3 +355,157 @@ test('selects saved operations inline and revisits ID-scoped attack history @see await expect(historyLink).toBeVisible() await expect(page.getByRole('heading', { name: 'Independent human finding', exact: true })).toBeVisible() }) + +test('exports selected findings as Word and Markdown @seeded', async ({ page, request }) => { + test.setTimeout(90_000) + const version = await (await request.get('/api/version')).json() + const headers = { 'PyRIT-Compatibility-ID': version.compatibility_id } + const operationResponse = await request.post('/api/operations', { + headers, data: { name: `Findings export ${Date.now()}` }, + }) + expect(operationResponse.status()).toBe(201) + const operation: Operation = await operationResponse.json() + const description = 'Ünïcødé 😀 中文 first line\n```\n# Not a finding\nkept literal\nlast line' + const longTitle = `Beyond the first page ${'unbroken'.repeat(30)}` + // Saved out of severity order, so the document's order can only come from severity. + const seeds = [ + { title: 'Export low', severity: 'low' }, + { title: 'Export critical', severity: 'critical', description }, + { title: longTitle, severity: 'informational' }, + ...Array.from({ length: 17 }, (_, index) => ({ + title: `Export moderate ${String(index).padStart(2, '0')}`, severity: 'moderate', + })), + { title: 'Export important', severity: 'important' }, + ] + const findings = new Map() + for (const seed of seeds) { + const response = await request.post(`/api/operations/${operation.id}/findings`, { headers, data: seed }) + expect(response.status()).toBe(201) + findings.set(seed.title, await response.json()) + } + const findingId = (title: string): string => { + const finding = findings.get(title) + if (!finding) throw new Error(`Expected a saved finding titled ${title}.`) + return finding.id + } + const attackResponse = await request.post('/api/attacks', { + headers, data: { + target_registry_name: 'browser_offline', + operation_id: operation.id, + prepended_conversation: [{ role: 'user', pieces: [{ original_value: 'Saved offline evidence.' }] }], + }, + }) + expect(attackResponse.ok()).toBeTruthy() + const attack: CreateAttackResponse = await attackResponse.json() + const attachResponse = await request.post( + `/api/operations/${operation.id}/findings/${findingId('Export critical')}/evidence`, + { headers, data: { attack_result_id: attack.attack_result_id } }, + ) + expect(attachResponse.ok()).toBeTruthy() + const evidence: FindingEvidenceAttachResponse = await attachResponse.json() + const selectedTitles = ['Export critical', 'Export moderate 03', 'Export low', longTitle] + const excludedTitles = seeds.map(seed => seed.title).filter(title => !selectedTitles.includes(title)) + + await page.goto(`/operations/${operation.id}`) + await waitForAppReady(page) + const region = page.getByRole('region', { name: 'Operation', exact: true }) + await expect(region.getByRole('listitem')).toHaveCount(20) + await expect(region.getByRole('heading', { name: longTitle })).toHaveCount(0) + const origin = new URL(page.url()).origin + const trigger = page.getByRole('button', { name: 'Export findings', exact: true }) + const dialog = page.getByRole('dialog', { name: 'Export findings' }) + const list = dialog.getByRole('list') + const selectAll = dialog.getByRole('checkbox', { name: 'Select all', exact: true }) + const format = dialog.getByRole('combobox', { name: 'Format' }) + const openWithEverySelected = async (): Promise => { + await trigger.click() + await expect(list.getByRole('checkbox', { checked: true })).toHaveCount(seeds.length) + await expect(selectAll).toBeChecked() + // The sensitive-content warning wraps, and its bar must grow to hold it. + const warning = dialog.locator('.fui-MessageBar').filter({ hasText: 'Nothing is redacted automatically.' }) + expect(await warning.evaluate(element => element.scrollHeight <= element.clientHeight)).toBeTruthy() + } + const selectSubset = async (): Promise => { + await selectAll.click() + await expect(list.getByRole('checkbox', { checked: true })).toHaveCount(0) + for (const title of selectedTitles) { + await list.getByRole('checkbox', { name: title, exact: true }).check() + } + await expect(selectAll).toBeChecked({ indeterminate: true }) + } + const download = async ( + button: string, expectedFormat: FindingExportFormat, + ): Promise<{ filename: string; content: Buffer }> => { + const exportRequest = page.waitForRequest(sent => + sent.url().endsWith(`/api/operations/${operation.id}/findings/export`) && sent.method() === 'POST') + const downloaded = page.waitForEvent('download') + await dialog.getByRole('button', { name: button, exact: true }).click() + expect((await exportRequest).postDataJSON()).toEqual({ + finding_ids: selectedTitles.map(findingId), format: expectedFormat, dashboard_base_url: `${origin}/`, + }) + const file = await downloaded + await expect(dialog).toHaveCount(0) + await expect(trigger).toBeFocused() + return { filename: file.suggestedFilename(), content: readFileSync(await file.path()) } + } + + await openWithEverySelected() + await expect(list.getByRole('checkbox').first()).toHaveAccessibleName('Export critical') + await expect(list.getByRole('checkbox').last()).toHaveAccessibleName(longTitle) + await selectAll.click() + await expect(list.getByRole('checkbox', { checked: true })).toHaveCount(0) + await expect(selectAll).not.toBeChecked() + await expect(dialog.getByRole('button', { name: 'Download Word', exact: true })).toBeDisabled() + await selectAll.click() + await expect(list.getByRole('checkbox', { checked: true })).toHaveCount(seeds.length) + await selectSubset() + await expect(format).toHaveValue('docx') + await expect(format.locator('option:checked')).toHaveText('Word (.docx)') + const word = await download('Download Word', 'docx') + // Named after the Operation and the browser's local time. + const filenameStart = `^copyrit-findings-${operation.name.toLowerCase().replace(/ /g, '-')}-\\d{4}-\\d{2}-\\d{2}-\\d{4}` + expect(word.filename).toMatch(new RegExp(`${filenameStart}\\.docx$`)) + expect(word.content.subarray(0, 4).toString('latin1')).toBe('PK\x03\x04') + expect(word.content.includes('word/document.xml')).toBeTruthy() + + // Reopening starts again from the whole Operation. + await openWithEverySelected() + await selectSubset() + await format.selectOption({ label: 'Markdown (.md)' }) + await expect(list.getByRole('checkbox', { checked: true })).toHaveCount(selectedTitles.length) + const markdownFile = await download('Download Markdown', 'markdown') + expect(markdownFile.filename).toMatch(new RegExp(`${filenameStart}\\.md$`)) + const markdown = markdownFile.content.toString('utf8') + expect(markdown.startsWith(`# Findings - ${operation.name}\n\n`)).toBeTruthy() + expect(markdown.match(/^## .*$/gm)).toEqual(selectedTitles.map(title => `## ${title}`)) + for (const title of excludedTitles) { + expect(markdown).not.toContain(title) + } + expect(markdown).toContain(`**Description**\n\n\`\`\`\`text\n${description}\n\`\`\`\``) + const link = `${origin}/attacks/${attack.attack_result_id}/conversations/${attack.conversation_id}` + + `?findingEvidenceId=${evidence.item.id}` + // As in Word, the evidence bullet's objective is the link text. + expect(markdown.split('\n').filter(line => line.startsWith('- [') && line.endsWith(`](<${link}>)`))).toHaveLength(1) + + await page.setViewportSize({ width: 390, height: 844 }) + expect(await page.evaluate(() => document.documentElement.scrollWidth <= window.innerWidth)).toBeTruthy() + await openWithEverySelected() + const content = dialog.locator('.fui-DialogContent') + expect(await content.evaluate(element => element.scrollHeight > element.clientHeight + && ['auto', 'scroll'].includes(getComputedStyle(element).overflowY))).toBeTruthy() + expect(await content.evaluate(element => element.scrollWidth <= element.clientWidth)).toBeTruthy() + expect(await dialog.evaluate(element => element.scrollWidth <= element.clientWidth)).toBeTruthy() + const longLabel = dialog.locator('label').filter({ hasText: longTitle }) + expect(await longLabel.evaluate(element => element.scrollWidth <= element.clientWidth)).toBeTruthy() + const longBox = await longLabel.boundingBox() + const shortBox = await dialog.locator('label').filter({ hasText: /^Export low$/ }).boundingBox() + if (!longBox || !shortBox) throw new Error('Expected visible finding labels.') + expect(longBox.height).toBeGreaterThan(2 * shortBox.height) + expect(await page.evaluate(() => document.documentElement.scrollWidth <= window.innerWidth)).toBeTruthy() + await dialog.getByRole('button', { name: 'Cancel', exact: true }).click() + await expect(dialog).toHaveCount(0) + await expect(trigger).toBeFocused() + + await page.goto(link) + await expect(page.getByText('Saved offline evidence.', { exact: true })).toBeVisible() +}) diff --git a/frontend/src/components/Operations/FindingExportDialog.styles.ts b/frontend/src/components/Operations/FindingExportDialog.styles.ts new file mode 100644 index 0000000000..0093513e51 --- /dev/null +++ b/frontend/src/components/Operations/FindingExportDialog.styles.ts @@ -0,0 +1,13 @@ +import { makeStyles, tokens } from '@fluentui/react-components' + +import { mobileTouchTargetHeight } from '@/styles/touchTargets' + +export const useFindingExportDialogStyles = makeStyles({ + content: { display: 'flex', flexDirection: 'column', gap: tokens.spacingVerticalM }, + messageBar: { flexShrink: 0 }, + list: { margin: 0, padding: 0, listStyleType: 'none' }, + row: { display: 'flex', flexWrap: 'wrap', alignItems: 'center', columnGap: tokens.spacingHorizontalS, minWidth: 0 }, + checkbox: { flex: '1 1 12rem', minWidth: 0 }, + label: { minWidth: 0, overflowWrap: 'anywhere' }, + button: { ...mobileTouchTargetHeight }, +}) diff --git a/frontend/src/components/Operations/FindingExportDialog.test.tsx b/frontend/src/components/Operations/FindingExportDialog.test.tsx new file mode 100644 index 0000000000..30cef36a60 --- /dev/null +++ b/frontend/src/components/Operations/FindingExportDialog.test.tsx @@ -0,0 +1,407 @@ +import { act, render, screen, waitFor, within } from '@testing-library/react' +import userEvent from '@testing-library/user-event' +import { FluentProvider, webLightTheme } from '@fluentui/react-components' + +import { operationsApi } from '@/services/api' +import { deferred } from '@/test-utils/deferred' +import type { FindingListItem, FindingListResponse, Operation } from '@/types' +import { downloadFile } from '@/utils/downloadFile' +import FindingExportDialog from './FindingExportDialog' + +jest.mock('@/services/api', () => ({ operationsApi: { listFindings: jest.fn(), exportFindings: jest.fn() } })) +jest.mock('@/utils/downloadFile', () => ({ downloadFile: jest.fn() })) + +const OPERATION: Operation = { + id: '6b1f0c3a-2d4e-4f5a-8b6c-7d8e9f0a1b2c', name: 'Synthetic evaluation', created_at: '2026-10-21T12:00:00Z', +} +const OTHER_OPERATION: Operation = { + id: '1c2d3e4f-5a6b-4c7d-8e9f-0a1b2c3d4e5f', name: 'Another evaluation', created_at: '2026-10-21T12:00:00Z', +} + +function syntheticFinding(index: number, overrides: Partial = {}): FindingListItem { + return { + id: `00000000-0000-4000-8000-${String(index).padStart(12, '0')}`, operation_id: OPERATION.id, + title: `Synthetic finding ${index}`, description: '', severity: 'moderate', + created_at: '2026-10-21T12:00:00Z', evidence_count: 0, ...overrides, + } +} + +function onePage(items: FindingListItem[]): FindingListResponse { + return { items, has_more: false, next_offset: null } +} + +const FIRST_PAGE: FindingListResponse = { + items: Array.from({ length: 100 }, (_, index) => syntheticFinding(index + 1)), has_more: true, next_offset: 100, +} +const LAST_PAGE_FINDING = syntheticFinding(101, { + title: 'Final-page finding', severity: 'other', severity_other: 'Team-specific severity', +}) +const LAST_PAGE = onePage([LAST_PAGE_FINDING]) +const ALPHA = syntheticFinding(1, { title: 'Alpha', severity: 'critical' }) +const BETA = syntheticFinding(2, { title: 'Beta', severity: 'low' }) +const GAMMA = syntheticFinding(3, { title: 'Gamma', severity: 'informational' }) +const BLOB = new Blob(['synthetic document']) +const FILENAME_TIME = '\\d{4}-\\d{2}-\\d{2}-\\d{4}' + +function renderDialog(operation: Operation = OPERATION): ReturnType { + return render( + + + , + ) +} + +async function openDialog(user: ReturnType): Promise { + // Tabster restores the background's accessibility after a deferred update once a dialog closes. + await user.click(await screen.findByRole('button', { name: 'Export findings' })) + return screen.getByRole('dialog', { name: 'Export findings' }) +} + +beforeEach(() => { + jest.clearAllMocks() + jest.mocked(operationsApi.listFindings).mockResolvedValue(onePage([ALPHA, BETA, GAMMA])) + jest.mocked(operationsApi.exportFindings).mockResolvedValue(BLOB) +}) + +afterEach(() => { + delete process.env.BASE_URL +}) + +describe('FindingExportDialog', () => { + it('loads every page before enabling download and sends exactly the selected final-page finding', async () => { + const user = userEvent.setup() + const lastPage = deferred() + jest.mocked(operationsApi.listFindings).mockImplementation( + async (_, { offset }) => (offset === 0 ? FIRST_PAGE : lastPage.promise), + ) + renderDialog() + const dialog = await openDialog(user) + await waitFor(() => expect(operationsApi.listFindings).toHaveBeenCalledTimes(2)) + expect(operationsApi.listFindings).toHaveBeenNthCalledWith(1, OPERATION.id, { limit: 100, offset: 0 }) + expect(operationsApi.listFindings).toHaveBeenNthCalledWith(2, OPERATION.id, { limit: 100, offset: 100 }) + expect(within(dialog).queryByRole('checkbox')).not.toBeInTheDocument() + expect(within(dialog).getByRole('button', { name: 'Download Word' })).toBeDisabled() + + await act(async () => { lastPage.resolve(LAST_PAGE) }) + const selectAll = within(dialog).getByRole('checkbox', { name: 'Select all' }) + const rows = within(dialog).getAllByRole('checkbox').filter(checkbox => checkbox !== selectAll) + expect(rows).toHaveLength(101) + for (const row of rows) expect(row).toBeChecked() + expect(selectAll).toBeChecked() + expect(within(dialog).getByText('Team-specific severity')).toBeInTheDocument() + + await user.click(selectAll) + expect(within(dialog).queryAllByRole('checkbox', { checked: true })).toHaveLength(0) + expect(within(dialog).getByRole('button', { name: 'Download Word' })).toBeDisabled() + + const lastRow = within(dialog).getByRole('checkbox', { name: LAST_PAGE_FINDING.title }) + await user.click(lastRow) + expect(selectAll).toBePartiallyChecked() + await user.selectOptions(within(dialog).getByRole('combobox', { name: 'Format' }), 'Markdown (.md)') + expect(lastRow).toBeChecked() + await user.click(within(dialog).getByRole('button', { name: 'Download Markdown' })) + + await waitFor(() => expect(downloadFile).toHaveBeenCalledTimes(1)) + expect(operationsApi.exportFindings).toHaveBeenCalledWith( + OPERATION.id, + { + finding_ids: [LAST_PAGE_FINDING.id], + format: 'markdown', + dashboard_base_url: 'http://localhost/', + }, + expect.any(AbortSignal), + ) + expect(downloadFile).toHaveBeenCalledWith( + BLOB, expect.stringMatching(new RegExp(`^copyrit-findings-synthetic-evaluation-${FILENAME_TIME}\\.md$`)), + ) + await waitFor(() => expect(screen.queryByRole('dialog')).not.toBeInTheDocument()) + await waitFor(() => expect(screen.getByRole('button', { name: 'Export findings' })).toHaveFocus()) + }) + + it('lists and exports once a finding that a later page repeats', async () => { + const user = userEvent.setup() + // A finding created while pages load shifts the next page, so it can start with the previous page's last row. + jest.mocked(operationsApi.listFindings) + .mockResolvedValueOnce({ items: [ALPHA, BETA], has_more: true, next_offset: 2 }) + .mockResolvedValueOnce(onePage([BETA, GAMMA])) + renderDialog() + const dialog = await openDialog(user) + await within(dialog).findByRole('checkbox', { name: GAMMA.title }) + expect(within(dialog).getAllByRole('checkbox', { name: BETA.title })).toHaveLength(1) + expect(within(dialog).getAllByRole('checkbox')).toHaveLength(4) + await user.click(within(dialog).getByRole('button', { name: 'Download Word' })) + + await waitFor(() => expect(operationsApi.exportFindings).toHaveBeenCalledTimes(1)) + expect(jest.mocked(operationsApi.exportFindings).mock.calls[0][1].finding_ids) + .toEqual([ALPHA.id, BETA.id, GAMMA.id]) + }) + + it('moves between all, none and subset selections', async () => { + const user = userEvent.setup() + renderDialog() + const dialog = await openDialog(user) + const selectAll = await within(dialog).findByRole('checkbox', { name: 'Select all' }) + const [alpha, beta, gamma] = [ALPHA, BETA, GAMMA].map(finding => ( + within(dialog).getByRole('checkbox', { name: finding.title }))) + const download = within(dialog).getByRole('button', { name: 'Download Word' }) + expect(selectAll).toBeChecked() + expect(selectAll).not.toBePartiallyChecked() + + await user.click(beta) + expect(selectAll).toBePartiallyChecked() + expect(download).toBeEnabled() + await user.click(selectAll) + for (const row of [alpha, beta, gamma]) expect(row).toBeChecked() + expect(selectAll).toBeChecked() + + await user.click(selectAll) + for (const row of [alpha, beta, gamma]) expect(row).not.toBeChecked() + expect(selectAll).not.toBeChecked() + expect(selectAll).not.toBePartiallyChecked() + expect(download).toBeDisabled() + + await user.click(alpha) + expect(selectAll).toBePartiallyChecked() + await user.click(beta) + await user.click(gamma) + expect(selectAll).toBeChecked() + expect(selectAll).not.toBePartiallyChecked() + }) + + it('defaults to Word and keeps the selection when the format changes', async () => { + const user = userEvent.setup() + renderDialog() + const dialog = await openDialog(user) + const beta = await within(dialog).findByRole('checkbox', { name: BETA.title }) + const format = within(dialog).getByRole('combobox', { name: 'Format' }) + expect(format).toHaveDisplayValue('Word (.docx)') + expect(within(format).getAllByRole('option').map(option => option.textContent)) + .toEqual(['Word (.docx)', 'Markdown (.md)']) + + await user.click(beta) + await user.selectOptions(format, 'Markdown (.md)') + expect(within(dialog).getByRole('button', { name: 'Download Markdown' })).toBeEnabled() + await user.selectOptions(format, 'Word (.docx)') + expect(beta).not.toBeChecked() + expect(within(dialog).getByRole('checkbox', { name: ALPHA.title })).toBeChecked() + await user.click(within(dialog).getByRole('button', { name: 'Download Word' })) + + await waitFor(() => expect(downloadFile).toHaveBeenCalledTimes(1)) + expect(operationsApi.exportFindings).toHaveBeenCalledWith(OPERATION.id, { + finding_ids: [ALPHA.id, GAMMA.id], format: 'docx', dashboard_base_url: 'http://localhost/', + }, expect.any(AbortSignal)) + expect(downloadFile).toHaveBeenCalledWith( + BLOB, expect.stringMatching(new RegExp(`^copyrit-findings-synthetic-evaluation-${FILENAME_TIME}\\.docx$`)), + ) + }) + + it('shows an empty Operation without an enabled download', async () => { + const user = userEvent.setup() + jest.mocked(operationsApi.listFindings).mockResolvedValue(onePage([])) + renderDialog() + const dialog = await openDialog(user) + expect(await within(dialog).findByText('This operation has no findings to export.')).toBeInTheDocument() + expect(within(dialog).queryByRole('checkbox')).not.toBeInTheDocument() + expect(within(dialog).getByRole('button', { name: 'Download Word' })).toBeDisabled() + }) + + it('offers no partial selection when a page fails, and retries the whole list', async () => { + const user = userEvent.setup() + jest.mocked(operationsApi.listFindings) + .mockResolvedValueOnce(FIRST_PAGE) + .mockRejectedValueOnce(new Error('Page unavailable')) + .mockResolvedValueOnce(FIRST_PAGE) + .mockResolvedValueOnce(LAST_PAGE) + renderDialog() + const dialog = await openDialog(user) + expect(await within(dialog).findByText(/Could not load findings: Page unavailable/)).toBeInTheDocument() + expect(within(dialog).queryByRole('checkbox')).not.toBeInTheDocument() + expect(within(dialog).getByRole('button', { name: 'Download Word' })).toBeDisabled() + + await user.click(within(dialog).getByRole('button', { name: 'Retry' })) + expect(await within(dialog).findByRole('checkbox', { name: LAST_PAGE_FINDING.title })).toBeChecked() + expect(within(dialog).getByRole('checkbox', { name: 'Select all' })).toBeChecked() + expect(within(dialog).queryByText(/Could not load findings/)).not.toBeInTheDocument() + expect(jest.mocked(operationsApi.listFindings).mock.calls.map(([, params]) => params.offset)) + .toEqual([0, 100, 0, 100]) + }) + + it('keeps the selection after an export error so the download can be retried', async () => { + const user = userEvent.setup() + jest.mocked(operationsApi.exportFindings).mockRejectedValueOnce({ + isAxiosError: true, response: { status: 500, data: { detail: 'Renderer unavailable' } }, + }) + renderDialog() + const dialog = await openDialog(user) + await user.click(await within(dialog).findByRole('checkbox', { name: BETA.title })) + await user.click(within(dialog).getByRole('button', { name: 'Download Word' })) + + expect(await within(dialog).findByText('Could not export findings: Renderer unavailable')).toBeInTheDocument() + expect(downloadFile).not.toHaveBeenCalled() + expect(within(dialog).getByRole('checkbox', { name: BETA.title })).not.toBeChecked() + expect(within(dialog).getByRole('checkbox', { name: ALPHA.title })).toBeChecked() + + await user.click(within(dialog).getByRole('button', { name: 'Download Word' })) + await waitFor(() => expect(downloadFile).toHaveBeenCalledTimes(1)) + for (const [, request] of jest.mocked(operationsApi.exportFindings).mock.calls) { + expect(request.finding_ids).toEqual([ALPHA.id, GAMMA.id]) + } + }) + + it('keeps the dialog and selection when the browser download cannot start', async () => { + const user = userEvent.setup() + jest.mocked(downloadFile).mockImplementationOnce(() => { throw new Error('Download blocked') }) + renderDialog() + const dialog = await openDialog(user) + await user.click(await within(dialog).findByRole('checkbox', { name: BETA.title })) + await user.click(within(dialog).getByRole('button', { name: 'Download Word' })) + + expect(await within(dialog).findByText('Could not export findings: Download blocked')).toBeInTheDocument() + expect(screen.getByRole('dialog', { name: 'Export findings' })).toBeInTheDocument() + expect(within(dialog).getByRole('checkbox', { name: BETA.title })).not.toBeChecked() + expect(within(dialog).getByRole('button', { name: 'Download Word' })).toBeEnabled() + }) + + it('prevents duplicate export requests while the document is prepared', async () => { + const user = userEvent.setup() + const exported = deferred() + jest.mocked(operationsApi.exportFindings).mockReturnValueOnce(exported.promise) + renderDialog() + const dialog = await openDialog(user) + await within(dialog).findByRole('checkbox', { name: 'Select all' }) + await user.click(within(dialog).getByRole('button', { name: 'Download Word' })) + + const pending = within(dialog).getByRole('button', { name: 'Preparing download…' }) + expect(pending).toHaveAttribute('aria-disabled', 'true') + expect(within(dialog).getByRole('checkbox', { name: 'Select all' })).toBeDisabled() + expect(within(dialog).getByRole('combobox', { name: 'Format' })).toBeDisabled() + await user.click(pending) + expect(operationsApi.exportFindings).toHaveBeenCalledTimes(1) + + await act(async () => { exported.resolve(BLOB) }) + expect(downloadFile).toHaveBeenCalledTimes(1) + }) + + it.each(['Cancel', 'Escape'])('restores focus to the trigger after closing with %s', async dismiss => { + const user = userEvent.setup() + renderDialog() + const dialog = await openDialog(user) + await within(dialog).findByRole('checkbox', { name: 'Select all' }) + if (dismiss === 'Escape') await user.keyboard('{Escape}') + else await user.click(within(dialog).getByRole('button', { name: 'Cancel' })) + + await waitFor(() => expect(screen.queryByRole('dialog')).not.toBeInTheDocument()) + await waitFor(() => expect(screen.getByRole('button', { name: 'Export findings' })).toHaveFocus()) + expect(operationsApi.exportFindings).not.toHaveBeenCalled() + }) + + it('warns that exports can contain unredacted sensitive content', async () => { + const user = userEvent.setup() + renderDialog() + const dialog = await openDialog(user) + expect(within(dialog).getByText( + 'Findings, attack objectives and score rationales can contain sensitive content. ' + + 'Nothing is redacted automatically.', + )).toBeInTheDocument() + }) + + it('sends the dashboard address with its application path prefix', async () => { + const user = userEvent.setup() + process.env.BASE_URL = '/team/pyrit/' + renderDialog() + const dialog = await openDialog(user) + await within(dialog).findByRole('checkbox', { name: 'Select all' }) + await user.click(within(dialog).getByRole('button', { name: 'Download Word' })) + + await waitFor(() => expect(operationsApi.exportFindings).toHaveBeenCalledTimes(1)) + expect(jest.mocked(operationsApi.exportFindings).mock.calls[0][1].dashboard_base_url) + .toBe('http://localhost/team/pyrit/') + }) + + it('ignores a list that finishes after the dialog was closed and reopened', async () => { + const user = userEvent.setup() + const stale = deferred() + const fresh = syntheticFinding(7, { title: 'Fresh finding' }) + jest.mocked(operationsApi.listFindings) + .mockReturnValueOnce(stale.promise) + .mockResolvedValueOnce(onePage([fresh])) + renderDialog() + let dialog = await openDialog(user) + await user.click(within(dialog).getByRole('button', { name: 'Cancel' })) + await waitFor(() => expect(screen.queryByRole('dialog')).not.toBeInTheDocument()) + dialog = await openDialog(user) + expect(await within(dialog).findByRole('checkbox', { name: fresh.title })).toBeChecked() + + await act(async () => { stale.resolve(onePage([syntheticFinding(8, { title: 'Stale finding' })])) }) + expect(within(dialog).queryByRole('checkbox', { name: 'Stale finding' })).not.toBeInTheDocument() + expect(within(dialog).getByRole('checkbox', { name: fresh.title })).toBeChecked() + }) + + it('does not download a file that finishes after the dialog was closed, and reopens with everything selected', async () => { + const user = userEvent.setup() + const exported = deferred() + jest.mocked(operationsApi.exportFindings).mockReturnValueOnce(exported.promise) + renderDialog() + let dialog = await openDialog(user) + await user.click(await within(dialog).findByRole('checkbox', { name: BETA.title })) + await user.selectOptions(within(dialog).getByRole('combobox', { name: 'Format' }), 'Markdown (.md)') + await user.click(within(dialog).getByRole('button', { name: 'Download Markdown' })) + await user.click(within(dialog).getByRole('button', { name: 'Cancel' })) + await waitFor(() => expect(screen.queryByRole('dialog')).not.toBeInTheDocument()) + expect(jest.mocked(operationsApi.exportFindings).mock.calls[0][2]?.aborted).toBe(true) + + await act(async () => { exported.resolve(BLOB) }) + expect(downloadFile).not.toHaveBeenCalled() + + dialog = await openDialog(user) + expect(await within(dialog).findByRole('checkbox', { name: BETA.title })).toBeChecked() + expect(within(dialog).getByRole('checkbox', { name: 'Select all' })).toBeChecked() + expect(within(dialog).getByRole('combobox', { name: 'Format' })).toHaveDisplayValue('Word (.docx)') + expect(within(dialog).getByRole('button', { name: 'Download Word' })).toBeEnabled() + expect(within(dialog).queryByText(/Could not export findings/)).not.toBeInTheDocument() + }) + + it('does not download the previous Operation\'s file after the Operation changes', async () => { + const user = userEvent.setup() + const exported = deferred() + jest.mocked(operationsApi.exportFindings).mockReturnValueOnce(exported.promise) + const view = renderDialog(OPERATION) + const dialog = await openDialog(user) + await within(dialog).findByRole('checkbox', { name: 'Select all' }) + await user.click(within(dialog).getByRole('button', { name: 'Download Word' })) + + view.rerender( + + + , + ) + expect(jest.mocked(operationsApi.exportFindings).mock.calls[0][2]?.aborted).toBe(true) + await act(async () => { exported.resolve(BLOB) }) + expect(downloadFile).not.toHaveBeenCalled() + expect(screen.queryByRole('dialog')).not.toBeInTheDocument() + }) + + it('ignores a list from the previous Operation that finishes after the Operation changes', async () => { + const user = userEvent.setup() + const staleList = deferred() + const other = syntheticFinding(9, { operation_id: OTHER_OPERATION.id, title: 'Other operation finding' }) + jest.mocked(operationsApi.listFindings) + .mockReturnValueOnce(staleList.promise) + .mockResolvedValueOnce(onePage([other])) + const view = renderDialog(OPERATION) + await openDialog(user) + + view.rerender( + + + , + ) + const dialog = await openDialog(user) + expect(await within(dialog).findByRole('checkbox', { name: other.title })).toBeChecked() + expect(operationsApi.listFindings).toHaveBeenLastCalledWith(OTHER_OPERATION.id, { limit: 100, offset: 0 }) + + await act(async () => { staleList.resolve(onePage([ALPHA])) }) + expect(within(dialog).queryByRole('checkbox', { name: ALPHA.title })).not.toBeInTheDocument() + expect(within(dialog).getByRole('checkbox', { name: other.title })).toBeChecked() + }) +}) diff --git a/frontend/src/components/Operations/FindingExportDialog.tsx b/frontend/src/components/Operations/FindingExportDialog.tsx new file mode 100644 index 0000000000..7417d84606 --- /dev/null +++ b/frontend/src/components/Operations/FindingExportDialog.tsx @@ -0,0 +1,209 @@ +import { useEffect, useRef, useState } from 'react' + +import { + Button, Checkbox, Dialog, DialogActions, DialogBody, DialogContent, DialogSurface, DialogTitle, DialogTrigger, + Field, MessageBar, MessageBarBody, Select, Spinner, Text, +} from '@fluentui/react-components' +import { ArrowDownloadRegular } from '@fluentui/react-icons' + +import { operationsApi } from '@/services/api' +import { toApiError } from '@/services/errors' +import type { FindingExportFormat, FindingListItem, Operation } from '@/types' +import { downloadFile } from '@/utils/downloadFile' +import { findingExportFilename } from '@/utils/findingExportFilename' + +import { useFindingExportDialogStyles } from './FindingExportDialog.styles' +import FindingSeverityBadge from './FindingSeverityBadge' + +const PAGE_SIZE = 100 +const FORMATS: Record = { + docx: { option: 'Word (.docx)', name: 'Word', extension: 'docx' }, + markdown: { option: 'Markdown (.md)', name: 'Markdown', extension: 'md' }, +} +const NO_SELECTION: ReadonlySet = new Set() + +type FindingsLoad = + | { status: 'loading' } + | { status: 'error'; error: string } + | { status: 'loaded'; findings: FindingListItem[] } + + +interface FindingExportDialogProps { + operation: Operation +} + +export default function FindingExportDialog({ operation }: FindingExportDialogProps) { + const styles = useFindingExportDialogStyles() + const [open, setOpen] = useState(false) + const [loadRevision, setLoadRevision] = useState(0) + const [load, setLoad] = useState({ status: 'loading' }) + const [selected, setSelected] = useState>(NO_SELECTION) + const [format, setFormat] = useState('docx') + const [exportError, setExportError] = useState('') + const [exporting, setExporting] = useState(false) + const submitting = useRef(false) + const triggerRef = useRef(null) + // Aborted when the dialog closes or unmounts, so a late export never downloads into a later view. + const session = useRef(null) + + const findings = load.status === 'loaded' ? load.findings : [] + const selectedIds = findings.filter(finding => selected.has(finding.id)).map(finding => finding.id) + const allSelected = selectedIds.length === findings.length + const selectAllChecked = allSelected ? true : selectedIds.length === 0 ? false : 'mixed' + + useEffect(() => { + if (!open) return + const controller = new AbortController() + session.current = controller + const trigger = triggerRef.current + return () => { + controller.abort() + // Tabster's restore depends on the browser reporting focus loss from the removed dialog; don't rely on it. + trigger?.focus() + } + }, [open]) + + useEffect(() => { + if (!open) return + let ignore = false + const loadAllPages = async (): Promise => { + const loaded = new Map() + let offset: number | null = 0 + while (offset !== null) { + const page = await operationsApi.listFindings(operation.id, { limit: PAGE_SIZE, offset }) + if (ignore) return + // A finding created while pages load shifts later pages, so a page can repeat an earlier row. + for (const finding of page.items) { + if (!loaded.has(finding.id)) loaded.set(finding.id, finding) + } + offset = page.has_more ? page.next_offset : null + } + setLoad({ status: 'loaded', findings: [...loaded.values()] }) + setSelected(new Set(loaded.keys())) + } + loadAllPages().catch((cause: unknown) => { + if (!ignore) setLoad({ status: 'error', error: toApiError(cause).detail }) + }) + return () => { ignore = true } + }, [open, loadRevision, operation.id]) + + const start = (): void => { + submitting.current = false + setExporting(false) + setExportError('') + setFormat('docx') + setSelected(NO_SELECTION) + setLoad({ status: 'loading' }) + setOpen(true) + } + + const toggle = (findingId: string, checked: boolean): void => { + setSelected(current => { + const next = new Set(current) + if (checked) next.add(findingId) + else next.delete(findingId) + return next + }) + } + + const download = async (): Promise => { + const controller = session.current + if (controller === null || controller.signal.aborted || submitting.current || selectedIds.length === 0) return + submitting.current = true + setExporting(true) + setExportError('') + try { + const file = await operationsApi.exportFindings(operation.id, { + finding_ids: selectedIds, + format, + dashboard_base_url: new URL(import.meta.env.BASE_URL || '/', window.location.origin).href, + }, controller.signal) + if (controller.signal.aborted) return + // The user's local time; a cross-origin deployment may not expose the server's attachment name. + downloadFile(file, findingExportFilename(operation, FORMATS[format].extension, new Date())) + setOpen(false) + } catch (cause: unknown) { + if (!controller.signal.aborted) setExportError(toApiError(cause).detail) + } finally { + if (!controller.signal.aborted) { + submitting.current = false + setExporting(false) + } + } + } + + const renderFindings = (): React.ReactNode => { + if (load.status === 'loading') return + if (load.status === 'error') { + return ( + + Could not load findings: {load.error}{' '} + + + ) + } + if (findings.length === 0) return This operation has no findings to export. + return ( + <> + { setSelected(allSelected ? NO_SELECTION : new Set(findings.map(finding => finding.id))) }} /> +
    + {findings.map(finding => ( +
  • + { toggle(finding.id, data.checked === true) }} /> + +
    • + ))} +
+ + ) + } + + return ( + { + if (data.open) start() + else setOpen(false) + }}> + + + + + + Export findings + + + Findings, attack objectives and score rationales can contain sensitive content. Nothing is redacted + automatically. + + {exportError && + Could not export findings: {exportError} + } + + + + {renderFindings()} + + + + + + + + + + + ) +} diff --git a/frontend/src/components/Operations/OperationDeleteDialog.tsx b/frontend/src/components/Operations/OperationDeleteDialog.tsx index 76dfe56d41..013dfeb7bf 100644 --- a/frontend/src/components/Operations/OperationDeleteDialog.tsx +++ b/frontend/src/components/Operations/OperationDeleteDialog.tsx @@ -3,6 +3,7 @@ import { Button, Dialog, DialogBody, DialogContent, DialogSurface, DialogTitle, DialogTrigger, MessageBar, MessageBarBody, Text, Tooltip, } from '@fluentui/react-components' +import { DeleteRegular } from '@fluentui/react-icons' import { operationsApi } from '@/services/api' import { toApiError } from '@/services/errors' @@ -64,7 +65,7 @@ export default function OperationDeleteDialog({ operation, onInUse, onDeleted }: }}> - + diff --git a/frontend/src/components/Operations/OperationDetailPage.test.tsx b/frontend/src/components/Operations/OperationDetailPage.test.tsx index 5d2ec2cb30..908bde9acc 100644 --- a/frontend/src/components/Operations/OperationDetailPage.test.tsx +++ b/frontend/src/components/Operations/OperationDetailPage.test.tsx @@ -12,7 +12,7 @@ jest.mock('@/services/api', () => ({ get: jest.fn(), update: jest.fn(), delete: jest.fn(), listFindings: jest.fn(), createFinding: jest.fn(), updateFinding: jest.fn(), deleteFinding: jest.fn(), listFindingEvidence: jest.fn(), detachFindingEvidence: jest.fn(), - getFindingOptions: jest.fn(), + getFindingOptions: jest.fn(), exportFindings: jest.fn(), }, attacksApi: { listAttacks: jest.fn() }, scenariosApi: { listRuns: jest.fn() }, @@ -477,6 +477,42 @@ it('keeps deletion errors visible and prevents duplicate confirmations', async ( expect(operationsApi.deleteFinding).toHaveBeenCalledTimes(1) }) +it('opens findings export from the header without changing findings or the dashboard page', async () => { + const user = userEvent.setup() + jest.mocked(operationsApi.listFindings).mockResolvedValue({ ...EMPTY, items: [FINDING] }) + renderPage('/operations/op-1?offset=20') + await screen.findByRole('heading', { level: 2, name: FINDING.title }) + const trigger = screen.getByRole('button', { name: 'Export findings' }) + await user.click(trigger) + const dialog = screen.getByRole('dialog', { name: 'Export findings' }) + expect(await within(dialog).findByRole('checkbox', { name: FINDING.title })).toBeChecked() + expect(operationsApi.listFindings).toHaveBeenCalledWith('op-1', { limit: 100, offset: 0 }) + + await user.click(within(dialog).getByRole('button', { name: 'Cancel' })) + await waitFor(() => expect(screen.queryByRole('dialog')).not.toBeInTheDocument()) + await waitFor(() => expect(trigger).toHaveFocus()) + expect(screen.getByText('Page 2')).toBeInTheDocument() + expect(jest.mocked(operationsApi.listFindings).mock.calls.filter(([, params]) => params.limit === 20)) + .toEqual([['op-1', { limit: 20, offset: 20 }]]) + expect(operationsApi.exportFindings).not.toHaveBeenCalled() + expect(operationsApi.createFinding).not.toHaveBeenCalled() + expect(operationsApi.updateFinding).not.toHaveBeenCalled() + expect(operationsApi.deleteFinding).not.toHaveBeenCalled() +}) + +it('shows decorative icons on the header actions but not on the finding actions', async () => { + jest.mocked(operationsApi.listFindings).mockResolvedValue({ ...EMPTY, items: [FINDING] }) + renderPage() + const heading = await screen.findByRole('heading', { level: 1, name: OPERATION.name }) + await screen.findByRole('heading', { level: 2, name: FINDING.title }) + const header = within(heading.closest('header') as HTMLElement) + const withoutIcon = ['Edit', 'Delete', 'Export findings', 'New finding'] + .filter(name => header.getByRole('button', { name }).querySelector('svg[aria-hidden="true"]') === null) + expect(withoutIcon).toEqual([]) + expect(screen.getByRole('button', { name: `Edit finding: ${FINDING.title}` }).querySelector('svg')).toBeNull() + expect(screen.getByRole('button', { name: `Delete finding: ${FINDING.title}` }).querySelector('svg')).toBeNull() +}) + it('pages through findings', async () => { const user = userEvent.setup() jest.mocked(operationsApi.listFindings).mockResolvedValueOnce({ items: [FINDING], has_more: true, next_offset: 20 }) diff --git a/frontend/src/components/Operations/OperationDetailPage.tsx b/frontend/src/components/Operations/OperationDetailPage.tsx index 9f777d6b66..d327beabb8 100644 --- a/frontend/src/components/Operations/OperationDetailPage.tsx +++ b/frontend/src/components/Operations/OperationDetailPage.tsx @@ -4,7 +4,7 @@ import { Button, Dialog, DialogBody, DialogContent, DialogSurface, DialogTitle, MessageBar, MessageBarBody, Spinner, Text, } from '@fluentui/react-components' -import { ArrowLeftRegular } from '@fluentui/react-icons' +import { AddRegular, ArrowLeftRegular } from '@fluentui/react-icons' import { DEFAULT_HISTORY_FILTERS, filtersToSearchParams } from '@/components/History/historyFilters' import { operationsApi } from '@/services/api' @@ -12,6 +12,7 @@ import { toApiError } from '@/services/errors' import type { Finding, FindingCreate, FindingListItem, FindingListResponse, OperationDetail } from '@/types' import FindingDialog, { type FindingSaveAttempt } from './FindingDialog' import FindingEvidenceList from './FindingEvidenceList' +import FindingExportDialog from './FindingExportDialog' import FindingSeverityBadge from './FindingSeverityBadge' import OperationDeleteDialog from './OperationDeleteDialog' import OperationEditDialog from './OperationEditDialog' @@ -202,7 +203,8 @@ function OperationDetailContent({ operationId, onOperationChanged }: { operation onOperationChanged?.() void navigate('/operations') }} /> - + diff --git a/frontend/src/services/api.test.ts b/frontend/src/services/api.test.ts index e12d0c483b..9aba3acf74 100644 --- a/frontend/src/services/api.test.ts +++ b/frontend/src/services/api.test.ts @@ -1,6 +1,7 @@ -// Mock axios before importing the module +// Mock axios before importing the module. The instance is callable, like a +// real Axios instance, so the 401 handler can replay a request through it. jest.mock("axios", () => ({ - create: jest.fn(() => ({ + create: jest.fn(() => Object.assign(jest.fn(), { get: jest.fn(), post: jest.fn(), put: jest.fn(), @@ -36,9 +37,12 @@ import { scenarioSuitesApi, datasetsApi, analyticsApi, + setMsalInstance, } from "./api"; import { compatibility, COMPATIBILITY_HEADER } from "./compatibility"; +import { toApiError } from "./errors"; import { makeAnalyticsFacets, makeAnalyticsReport, makeAnalyticsResults } from "@/test-utils/analyticsFixtures"; +import type { FindingExportRequest } from "@/types"; import { DEFAULT_ANALYTICS_VIEW } from "@/utils/attackAnalytics"; describe("api service", () => { @@ -88,6 +92,22 @@ describe("api service", () => { expect(await operationsApi.deleteFinding("a/b", "c/d")).toBeUndefined(); expect(apiClient.delete).toHaveBeenCalledWith("/operations/a%2Fb/findings/c%2Fd"); }); + it("downloads the explicitly selected findings as a cancellable file", async () => { + const request: FindingExportRequest = { + finding_ids: ["11111111-1111-4111-8111-111111111111", "22222222-2222-4222-8222-222222222222"], + format: "markdown", + dashboard_base_url: "https://pyrit.example.test/red-team/", + }; + const blob = new Blob(["# Findings"], { type: "text/markdown" }); + const controller = new AbortController(); + (apiClient.post as jest.Mock).mockResolvedValueOnce({ data: blob }); + expect(await operationsApi.exportFindings("a/b", request, controller.signal)).toBe(blob); + expect(apiClient.post).toHaveBeenCalledWith( + "/operations/a%2Fb/findings/export", + request, + { responseType: "blob", signal: controller.signal }, + ); + }); }); // Interceptor functions are registered at module-load time. // Capture them before beforeEach's clearAllMocks wipes the call records. @@ -268,6 +288,117 @@ describe("api service", () => { consoleSpy.mockRestore(); }); + + describe("file download errors", () => { + const backendId = `1.2.0.dev0+g${"b".repeat(40)}`; + + function jsonBlob(body: unknown): Blob { + return new Blob([JSON.stringify(body)], { type: "application/json" }); + } + + function fileRequestError(status: number, data: unknown) { + const headers: Record = { "X-Request-ID": "export-request" }; + return { + isAxiosError: true, + config: { + method: "post", + url: "/operations/op/findings/export", + responseType: "blob", + headers: Object.assign(headers, { set: (name: string, value: string) => { headers[name] = value; } }), + }, + response: { status, data }, + }; + } + + function replaceCompatibilityStore() { + const compatibilityModule = jest.requireMock("./compatibility"); + const store = new compatibilityModule.CompatibilityStore(compatibility.bundledId); + jest.replaceProperty(compatibilityModule, "compatibility", store); + return store; + } + + let consoleError: jest.SpyInstance; + + beforeEach(() => { + consoleError = jest.spyOn(console, "error").mockImplementation(() => {}); + }); + + afterEach(() => { + setMsalInstance(null); + jest.restoreAllMocks(); + }); + + it.each([ + ["JSON", 409, "urn:pyrit:compatibility:mismatch", false], + ["Blob", 409, "urn:pyrit:compatibility:mismatch", true], + ["Blob", 400, "urn:pyrit:compatibility:invalid", true], + ])("blocks an incompatible frontend on a %s %i compatibility problem", async (_body, status, type, asBlob) => { + const store = replaceCompatibilityStore(); + const problem = { type, expected: backendId, actual: compatibility.bundledId }; + const error = fileRequestError(status, asBlob ? jsonBlob(problem) : problem); + + await expect(responseOnError(error)).rejects.toBe(error); + + expect(store.getSnapshot()).toEqual({ + status: "blocked", + reason: "The backend rejected this frontend build identity.", + expected: backendId, + actual: compatibility.bundledId, + }); + }); + + it.each([["JSON", false], ["Blob", true]])( + "refreshes the token and replays the file request once after a %s 401", + async (_body, asBlob) => { + const acquireTokenSilent = jest.fn().mockResolvedValue({ accessToken: "fresh-token" }); + setMsalInstance({ + getActiveAccount: () => ({ username: "user@example.com" }), + acquireTokenSilent, + } as unknown as Parameters[0]); + const replayed = { status: 200, data: new Blob(["document"]) }; + (apiClient as unknown as jest.Mock).mockResolvedValueOnce(replayed); + const unauthorized = { detail: "Not authenticated" }; + const error = fileRequestError(401, asBlob ? jsonBlob(unauthorized) : unauthorized); + + await expect(responseOnError(error)).resolves.toBe(replayed); + + expect(acquireTokenSilent).toHaveBeenCalledWith(expect.objectContaining({ forceRefresh: true })); + expect(apiClient).toHaveBeenCalledTimes(1); + expect(apiClient).toHaveBeenCalledWith(error.config); + expect(error.config).toMatchObject({ _retried: true, responseType: "blob" }); + expect(error.config.headers.Authorization).toBe("Bearer fresh-token"); + expect(consoleError).not.toHaveBeenCalled(); + }, + ); + + it("logs and rejects with a Blob error's decoded detail and original status", async () => { + const error = fileRequestError(404, jsonBlob({ detail: "Finding not found" })); + + await expect(responseOnError(error)).rejects.toBe(error); + + expect(toApiError(error)).toMatchObject({ status: 404, detail: "Finding not found" }); + expect(consoleError).toHaveBeenCalledWith( + "[API] POST /operations/op/findings/export failed | status=404 | " + + "requestId=export-request | Finding not found", + ); + }); + + it("keeps an unreadable Blob error a visible failure with its status", async () => { + const read = jest.spyOn(FileReader.prototype, "readAsText").mockImplementation(() => undefined); + const body = jsonBlob({ detail: "Unreadable" }); + const error = fileRequestError(500, body); + + const rejected = expect(responseOnError(error)).rejects.toBe(error); + const reader = read.mock.contexts[0]; + if (!(reader instanceof FileReader)) throw new Error("Blob read did not start"); + reader.dispatchEvent(new ProgressEvent("error")); + await rejected; + + expect(error.response.data).toBe(body); + expect(toApiError(error)).toMatchObject({ status: 500, detail: "Server error (500)" }); + expect(consoleError).toHaveBeenCalledWith(expect.stringContaining("status=500")); + }); + }); }); describe("analyticsApi", () => { diff --git a/frontend/src/services/api.ts b/frontend/src/services/api.ts index 1037dde32f..60220a1da6 100644 --- a/frontend/src/services/api.ts +++ b/frontend/src/services/api.ts @@ -1,7 +1,7 @@ import axios, { type InternalAxiosRequestConfig } from 'axios' import { InteractionRequiredAuthError, type PublicClientApplication } from '@azure/msal-browser' import { generateClientId } from '@/utils/clientId' -import { toApiError } from './errors' +import { decodeBlobErrorAsync, toApiError } from './errors' import { compatibility, COMPATIBILITY_HEADER } from './compatibility' import { getGraphScopes } from '../auth/msalConfig' import type { @@ -83,6 +83,7 @@ import type { FindingEvidenceCreateRequest, FindingEvidenceAttachResponse, FindingEvidenceListResponse, + FindingExportRequest, Operation, OperationCreate, OperationDetail, @@ -152,6 +153,13 @@ export const operationsApi = { `${operationPath(operationId)}/findings/${encodeURIComponent(findingId)}/evidence/${encodeURIComponent(evidenceId)}`, ) }, + exportFindings: async ( + operationId: string, request: FindingExportRequest, signal?: AbortSignal, + ): Promise => { + return (await apiClient.post( + `${operationPath(operationId)}/findings/export`, request, { responseType: 'blob', signal }, + )).data + }, } const apiClient = axios.create({ @@ -228,6 +236,8 @@ apiClient.interceptors.request.use(async (config) => { apiClient.interceptors.response.use( (response) => response, async (error) => { + // File downloads receive error bodies as Blobs; read them before any check below. + await decodeBlobErrorAsync(error) const problem = error?.response?.data if ( (error?.response?.status === 400 && problem?.type === 'urn:pyrit:compatibility:invalid') || diff --git a/frontend/src/services/errors.test.ts b/frontend/src/services/errors.test.ts index e917f6c35d..a926cf5451 100644 --- a/frontend/src/services/errors.test.ts +++ b/frontend/src/services/errors.test.ts @@ -1,4 +1,4 @@ -import { retryAfterMilliseconds, toApiError } from './errors' +import { decodeBlobErrorAsync, retryAfterMilliseconds, toApiError } from './errors' import type { ApiError } from './errors' // --------------------------------------------------------------------------- @@ -208,3 +208,71 @@ describe('toApiError', () => { expect(result.type).toBe('validation_error') }) }) + +describe('decodeBlobErrorAsync', () => { + it('replaces a JSON Blob body with the parsed problem and keeps the status', async () => { + const problem = { detail: 'Operation not found' } + const err = makeAxiosError({ + status: 404, + data: new Blob([JSON.stringify(problem)], { type: 'application/json' }), + }) + + await decodeBlobErrorAsync(err) + + expect(err).toMatchObject({ response: { status: 404, data: problem } }) + expect(toApiError(err)).toMatchObject({ status: 404, detail: 'Operation not found' }) + }) + + it('keeps a non-JSON Blob body, such as a proxy page, as readable text', async () => { + const page = 'Bad Gateway' + const err = makeAxiosError({ status: 502, data: new Blob([page], { type: 'text/html' }) }) + + await decodeBlobErrorAsync(err) + + expect(err).toMatchObject({ response: { status: 502, data: page } }) + expect(toApiError(err)).toMatchObject({ status: 502, detail: page }) + }) + + it('leaves the error failing with its status when the Blob cannot be read', async () => { + const read = jest.spyOn(FileReader.prototype, 'readAsText').mockImplementation(() => undefined) + try { + const body = new Blob(['{"detail":"Unreadable"}'], { type: 'application/json' }) + const err = makeAxiosError({ status: 500, data: body }) + + const decoding = decodeBlobErrorAsync(err) + const reader = read.mock.contexts[0] + if (!(reader instanceof FileReader)) throw new Error('Blob read did not start') + reader.dispatchEvent(new ProgressEvent('error')) + + await expect(decoding).resolves.toBeUndefined() + expect(err).toMatchObject({ response: { status: 500 } }) + expect((err as { response: { data: unknown } }).response.data).toBe(body) + expect(toApiError(err)).toMatchObject({ status: 500, detail: 'Server error (500)' }) + } finally { + read.mockRestore() + } + }) + + it.each([ + ['a JSON body', { detail: 'Finding not found' }], + ['a text body', 'Bad Gateway'], + ['no body', null], + ])('leaves an Axios error with %s unchanged', async (_name, data) => { + const err = makeAxiosError({ status: 404, data }) + + await decodeBlobErrorAsync(err) + + expect((err as { response: { data: unknown } }).response.data).toBe(data) + }) + + it('ignores a Blob that is not an Axios response body', async () => { + const body = new Blob(['{"detail":"Not from Axios"}'], { type: 'application/json' }) + const notAxios = { response: { status: 404, data: body } } + + await decodeBlobErrorAsync(notAxios) + await decodeBlobErrorAsync(new Error('Something broke')) + await decodeBlobErrorAsync(makeAxiosError({ hasResponse: false })) + + expect(notAxios.response.data).toBe(body) + }) +}) diff --git a/frontend/src/services/errors.ts b/frontend/src/services/errors.ts index 4a197944bf..492d34a5dd 100644 --- a/frontend/src/services/errors.ts +++ b/frontend/src/services/errors.ts @@ -132,6 +132,28 @@ export function retryAfterMilliseconds(error: unknown): number { return Number.isFinite(date) ? Math.max(0, date - Date.now()) : 0 } +/** + * Replace a file download's Blob error body with its JSON, or its text when it + * isn't JSON, so compatibility checks, token refresh and {@link toApiError} can + * read it. An unreadable body is left as is; other errors are unchanged. + */ +export async function decodeBlobErrorAsync(error: unknown): Promise { + const response = isAxiosError(error) ? error.response : undefined + const body: unknown = response?.data + if (!response || !(body instanceof Blob)) return + let text: string + try { + text = await readBlobText(body) + } catch { + return + } + try { + response.data = JSON.parse(text) + } catch { + response.data = text + } +} + // --------------------------------------------------------------------------- // Internal helpers // --------------------------------------------------------------------------- @@ -141,6 +163,15 @@ function isAxiosError(err: unknown): err is AxiosError { return typeof err === 'object' && err !== null && (err as AxiosError).isAxiosError === true } +function readBlobText(blob: Blob): Promise { + return new Promise((resolve, reject) => { + const reader = new FileReader() + reader.onload = () => resolve(reader.result as string) + reader.onerror = () => reject(reader.error) + reader.readAsText(blob) + }) +} + /** * Extract `detail` and `type` from an Axios response body. * diff --git a/frontend/src/types/index.ts b/frontend/src/types/index.ts index bf65c1844d..f735d181b3 100644 --- a/frontend/src/types/index.ts +++ b/frontend/src/types/index.ts @@ -2019,3 +2019,13 @@ export interface FindingEvidenceListResponse { has_more: boolean next_offset: number | null } + +export type FindingExportFormat = 'docx' | 'markdown' + +/** Selected findings of one Operation to download as a single document. */ +export interface FindingExportRequest { + finding_ids: string[] + format: FindingExportFormat + /** The dashboard address, including any path prefix, that evidence links open. */ + dashboard_base_url: string +} diff --git a/frontend/src/utils/conversationExport.ts b/frontend/src/utils/conversationExport.ts index 154b62b1c6..11df6cf861 100644 --- a/frontend/src/utils/conversationExport.ts +++ b/frontend/src/utils/conversationExport.ts @@ -1,4 +1,5 @@ import type { Message, MessageAttachment, MessageError } from '../types' +import { downloadFile } from './downloadFile' import { fileToBase64 } from './messageMapper' export type ExportFormat = 'markdown' | 'json' | 'html' @@ -275,26 +276,11 @@ export function buildExportFilename( } /** - * Trigger a browser download of `content` as `filename`. Uses the Blob → object - * URL → anchor-click idiom and always revokes the object URL, even if the click - * throws. + * Trigger a browser download of `content` as `filename`, encoded as a Blob of + * `mimeType`. See {@link downloadFile} for the link and object URL cleanup. */ export function downloadTextFile(content: string, filename: string, mimeType: string): void { - const blob = new Blob([content], { type: mimeType }) - const objectUrl = URL.createObjectURL(blob) - try { - const link = document.createElement('a') - link.href = objectUrl - link.download = filename - document.body.appendChild(link) - try { - link.click() - } finally { - link.remove() - } - } finally { - URL.revokeObjectURL(objectUrl) - } + downloadFile(new Blob([content], { type: mimeType }), filename) } function withoutLoadingPlaceholders(messages: Message[]): Message[] { diff --git a/frontend/src/utils/downloadFile.test.ts b/frontend/src/utils/downloadFile.test.ts new file mode 100644 index 0000000000..998acffbb9 --- /dev/null +++ b/frontend/src/utils/downloadFile.test.ts @@ -0,0 +1,66 @@ +import { downloadFile } from "./downloadFile"; + +const WORD_MIME_TYPE = "application/vnd.openxmlformats-officedocument.wordprocessingml.document"; + +/** Observe the download link the helper creates, replacing its click with `onClick`. */ +function spyOnDownloadLink(onClick: (link: HTMLAnchorElement) => void): () => HTMLAnchorElement | undefined { + let link: HTMLAnchorElement | undefined; + const createElement = document.createElement.bind(document); + jest.spyOn(document, "createElement").mockImplementation((tagName: string) => { + const element = createElement(tagName); + if (element instanceof HTMLAnchorElement) { + link = element; + jest.spyOn(element, "click").mockImplementation(() => onClick(element)); + } + return element; + }); + return () => link; +} + +describe("downloadFile", () => { + beforeEach(() => { + jest.clearAllMocks(); + }); + + afterEach(() => { + jest.restoreAllMocks(); + }); + + it("should click a link to the Blob under the requested filename, then remove it and revoke the URL", () => { + const blob = new Blob(["report"], { type: WORD_MIME_TYPE }); + const clicks: Array<{ href: string; download: string; attached: boolean; revoked: boolean }> = []; + const getLink = spyOnDownloadLink((link) => { + clicks.push({ + href: link.href, + download: link.download, + attached: link.isConnected, + revoked: jest.mocked(URL.revokeObjectURL).mock.calls.length > 0, + }); + }); + + downloadFile(blob, "copyrit-findings-operation-20261021T120000Z.docx"); + + expect(URL.createObjectURL).toHaveBeenCalledWith(blob); + expect(clicks).toEqual([{ + href: "blob:mock-url", + download: "copyrit-findings-operation-20261021T120000Z.docx", + attached: true, + revoked: false, + }]); + expect(getLink()?.isConnected).toBe(false); + expect(URL.revokeObjectURL).toHaveBeenCalledTimes(1); + expect(URL.revokeObjectURL).toHaveBeenCalledWith("blob:mock-url"); + }); + + it("should remove the link and revoke the URL when starting the download throws", () => { + const getLink = spyOnDownloadLink(() => { + throw new Error("download blocked"); + }); + + expect(() => downloadFile(new Blob(["# Findings"], { type: "text/markdown" }), "findings.md")) + .toThrow("download blocked"); + + expect(getLink()?.isConnected).toBe(false); + expect(URL.revokeObjectURL).toHaveBeenCalledWith("blob:mock-url"); + }); +}); diff --git a/frontend/src/utils/downloadFile.ts b/frontend/src/utils/downloadFile.ts new file mode 100644 index 0000000000..859a088f61 --- /dev/null +++ b/frontend/src/utils/downloadFile.ts @@ -0,0 +1,21 @@ +/** + * Start a browser download of `blob` as `filename` by clicking a temporary link + * to an object URL. The link is removed and the URL revoked even if the click + * throws. + */ +export function downloadFile(blob: Blob, filename: string): void { + const objectUrl = URL.createObjectURL(blob) + try { + const link = document.createElement('a') + link.href = objectUrl + link.download = filename + document.body.appendChild(link) + try { + link.click() + } finally { + link.remove() + } + } finally { + URL.revokeObjectURL(objectUrl) + } +} diff --git a/frontend/src/utils/findingExportFilename.test.ts b/frontend/src/utils/findingExportFilename.test.ts new file mode 100644 index 0000000000..f50b800158 --- /dev/null +++ b/frontend/src/utils/findingExportFilename.test.ts @@ -0,0 +1,31 @@ +import { findingExportFilename } from './findingExportFilename' + +const OPERATION_ID = 'c1e5502b-91f6-4d65-b182-4b028c7a75f6' +// Built from local parts, so the expected time does not depend on the test machine's time zone. +const EXPORT_TIME = new Date(2026, 9, 22, 12, 58, 33) + +describe('findingExportFilename', () => { + it.each([ + ['Synthetic evaluation', 'synthetic-evaluation'], + ['Q3 Red Team / Copilot (v2)', 'q3-red-team-copilot-v2'], + ['Évaluation: modèle 2', 'évaluation-modèle-2'], + ['E\u0301valuation', 'évaluation'], + // Devanagari writes vowel signs and viramas as combining marks, which belong to their word. + ['हिन्दी परीक्षण', 'हिन्दी-परीक्षण'], + ['snake_case name', 'snake-case-name'], + [`${'a'.repeat(58)} bcdef`, 'a'.repeat(58)], + ['x'.repeat(80), 'x'.repeat(60)], + ['𠀀'.repeat(61), '𠀀'.repeat(60)], + // The 61st code point is a vowel sign, so the cut moves back before its letter. + [`क${'कि'.repeat(30)}`, `क${'कि'.repeat(29)}`], + ['???', OPERATION_ID], + ])('names %j after the Operation, in local time', (name, expected) => { + expect(findingExportFilename({ id: OPERATION_ID, name }, 'docx', EXPORT_TIME)) + .toBe(`copyrit-findings-${expected}-2026-10-22-1258.docx`) + }) + + it('pads single-digit months, days, hours and minutes', () => { + expect(findingExportFilename({ id: OPERATION_ID, name: 'Synthetic evaluation' }, 'md', new Date(2026, 0, 5, 7, 4))) + .toBe('copyrit-findings-synthetic-evaluation-2026-01-05-0704.md') + }) +}) diff --git a/frontend/src/utils/findingExportFilename.ts b/frontend/src/utils/findingExportFilename.ts new file mode 100644 index 0000000000..3d8d784353 --- /dev/null +++ b/frontend/src/utils/findingExportFilename.ts @@ -0,0 +1,26 @@ +import type { Operation } from '@/types' + +const NAME_LIMIT = 60 +const MARK = /^\p{M}$/u + +/** + * Name an exported findings document after its Operation and the user's local time, e.g. + * `copyrit-findings-synthetic-evaluation-2026-10-22-1258.docx`. The backend applies the same + * rule in labelled UTC for clients that use its suggested name. + */ +export function findingExportFilename(operation: Pick, extension: string, now: Date): string { + const name = Array.from( + operation.name.normalize('NFC').toLowerCase().replace(/[^\p{L}\p{M}\p{N}]+/gu, '-').replace(/^-+|-+$/g, ''), + ) + let end = name.length + if (name.length > NAME_LIMIT) { + end = NAME_LIMIT + if (name[end] !== '-' && name.slice(0, end).includes('-')) end = name.lastIndexOf('-', end - 1) + // A word longer than the limit is cut between letters, never between a letter and its marks. + while (end > 0 && MARK.test(name[end])) end -= 1 + } + const slug = name.slice(0, end).join('').replace(/-+$/, '') + const pad = (value: number) => String(value).padStart(2, '0') + const date = `${now.getFullYear()}-${pad(now.getMonth() + 1)}-${pad(now.getDate())}` + return `copyrit-findings-${slug || operation.id}-${date}-${pad(now.getHours())}${pad(now.getMinutes())}.${extension}` +} diff --git a/frontend/src/utils/routeParams.ts b/frontend/src/utils/routeParams.ts index d0ca938e3e..c679c32760 100644 --- a/frontend/src/utils/routeParams.ts +++ b/frontend/src/utils/routeParams.ts @@ -41,7 +41,12 @@ export function attackRoutePath( ) } -/** Builds an attack-conversation route with optional bounded scenario-run provenance. */ +/** + * Builds an attack-conversation route with optional bounded scenario-run provenance. + * + * Exported findings documents link to this route (`_conversation_link` in + * `pyrit/backend/services/finding_export.py`), so keep its old shape working if it changes. + */ export function attackConversationRoutePath( attackResultId: string, conversationId: string, diff --git a/pyrit/backend/models/operations.py b/pyrit/backend/models/operations.py index 1b62d10bc0..2b8ace2283 100644 --- a/pyrit/backend/models/operations.py +++ b/pyrit/backend/models/operations.py @@ -4,7 +4,7 @@ from enum import Enum from uuid import UUID -from pydantic import BaseModel, ConfigDict, Field +from pydantic import AnyHttpUrl, BaseModel, ConfigDict, Field, field_validator from pyrit.backend.models.attacks import ScoreView from pyrit.models import ( @@ -134,3 +134,54 @@ class FindingEvidenceListResponse(BaseModel): items: list[FindingEvidenceItem] has_more: bool next_offset: int | None + + +class FindingExportFormat(str, Enum): + """A file format for exported findings.""" + + DOCX = "docx" + MARKDOWN = "markdown" + + +class FindingExportRequest(BaseModel): + """The findings to download, the file format, and the dashboard address their evidence links open.""" + + model_config = ConfigDict(extra="forbid") + + finding_ids: list[UUID] = Field(min_length=1, description="Distinct findings of the operation to include") + format: FindingExportFormat = FindingExportFormat.DOCX + dashboard_base_url: AnyHttpUrl = Field( + description="The dashboard's address, including any path prefix, that evidence links open" + ) + + @field_validator("finding_ids") + @classmethod + def _reject_duplicates(cls, finding_ids: list[UUID]) -> list[UUID]: + """ + Require each finding to be selected once. + + Returns: + list[UUID]: The selected finding IDs. + + Raises: + ValueError: If a finding is selected more than once. + """ + if len(set(finding_ids)) != len(finding_ids): + raise ValueError("Each finding can be selected only once.") + return finding_ids + + @field_validator("dashboard_base_url") + @classmethod + def _reject_credentials_query_and_fragment(cls, url: AnyHttpUrl) -> AnyHttpUrl: + """ + Allow an application path prefix, but not credentials, a query, or a fragment. + + Returns: + AnyHttpUrl: The dashboard address. + + Raises: + ValueError: If the address has credentials, a query, or a fragment. + """ + if url.username or url.password or url.query is not None or url.fragment is not None: + raise ValueError("Must not include credentials, a query, or a fragment.") + return url diff --git a/pyrit/backend/routes/operations.py b/pyrit/backend/routes/operations.py index b7719df90d..1cfa13f7bc 100644 --- a/pyrit/backend/routes/operations.py +++ b/pyrit/backend/routes/operations.py @@ -3,6 +3,7 @@ from collections.abc import Awaitable from typing import Annotated, TypeVar +from urllib.parse import quote from uuid import UUID from fastapi import APIRouter, HTTPException, Query, Response, status @@ -12,12 +13,14 @@ FindingEvidenceAttachResponse, FindingEvidenceCreateRequest, FindingEvidenceListResponse, + FindingExportRequest, FindingListResponse, FindingOptionsResponse, OperationDetail, OperationListResponse, OperationUpdateRequest, ) +from pyrit.backend.services.finding_export import export_findings_async from pyrit.backend.services.operation_service import ( DuplicateOperationError, FindingConflictError, @@ -164,6 +167,29 @@ async def list_operation_findings_async( ) +@router.post("/{operation_id}/findings/export", response_class=Response) +async def export_operation_findings_async(*, operation_id: UUID, request: FindingExportRequest) -> Response: + """ + Download selected findings of an operation as one Word or Markdown document, in the dashboard's order. + + Evidence appears as saved metadata and dashboard links, without transcripts. + + Returns: + Response: The complete document as an attachment. + """ + exported = await _require_operation_async(export_findings_async(operation_id=operation_id, request=request)) + # Headers can't carry raw non-ASCII text, so such names use RFC 5987 encoding, as Starlette's FileResponse does. + quoted = quote(exported.filename) + disposition = ( + f'attachment; filename="{exported.filename}"' + if quoted == exported.filename + else f"attachment; filename*=utf-8''{quoted}" + ) + return Response( + content=exported.content, media_type=exported.media_type, headers={"Content-Disposition": disposition} + ) + + @router.put("/{operation_id}/findings/{finding_id}", response_model=Finding) async def update_operation_finding_async(*, operation_id: UUID, finding_id: UUID, request: FindingCreate) -> Finding: """ diff --git a/pyrit/backend/services/finding_export.py b/pyrit/backend/services/finding_export.py new file mode 100644 index 0000000000..5a8d0c40fb --- /dev/null +++ b/pyrit/backend/services/finding_export.py @@ -0,0 +1,515 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +"""Selected findings of an Operation as a Word or Markdown section to copy into a report.""" + +import asyncio +import re +import unicodedata +from dataclasses import dataclass +from datetime import UTC, datetime +from io import BytesIO +from typing import Literal +from urllib.parse import quote, urlencode +from uuid import UUID + +import regex +from docx import Document +from docx.document import Document as DocxDocument +from docx.opc.constants import RELATIONSHIP_TYPE +from docx.oxml import OxmlElement +from docx.oxml.ns import qn +from docx.shared import Pt, RGBColor +from docx.table import _Cell +from docx.text.paragraph import Paragraph +from pydantic import AnyHttpUrl + +from pyrit.backend.models.attacks import ScoreView +from pyrit.backend.models.operations import ( + EvidenceAvailability, + FindingEvidenceItem, + FindingExportFormat, + FindingExportRequest, +) +from pyrit.backend.services.operation_service import FindingNotFoundError, OperationService +from pyrit.models import Finding, FindingSeverity, ScoreStatus +from pyrit.models.harm_category import HarmCategory + +_PAGE_SIZE = 100 +_MARKDOWN_MEDIA_TYPE = "text/markdown; charset=utf-8" +_DOCX_MEDIA_TYPE = "application/vnd.openxmlformats-officedocument.wordprocessingml.document" +_EVIDENCE_NOTE = "Evidence is listed as references with links to the PyRIT dashboard; transcripts are not included." +_UNAVAILABLE = "Evidence unavailable" +_LINK_COLOR = RGBColor(0x05, 0x63, 0xC1) +_HUMAN_SCORER = "ManualScorer" +_BOOLEAN_SCORES = {"true": "True", "false": "False"} +# Word 2013 and later; the template's older mode opens documents in Compatibility Mode. +_WORD_COMPATIBILITY_MODE = "15" +_WORD_BODY_FONT = "Calibri" +_WORD_HEADING_FONT = "Calibri Light" +_WORD_TITLE_COLOR = RGBColor(0x2F, 0x54, 0x96) +_WORD_TITLE_SIZE = Pt(16) +_WORD_TABLE_STYLE = "Table Grid" +_WORD_COLUMNS = ("Finding", "Assessment") +_DETAIL_SEPARATOR = " · " +_WORD_BULLET_STYLE = "List Bullet" +# The template's List Bullet text indent, so an entry's lines align under its bullet text. +_WORD_BULLET_TEXT_INDENT = Pt(18) +_THEME_COLOR_ATTRIBUTES = ("w:themeColor", "w:themeShade", "w:themeTint") +_LINE_BREAK = re.compile("\r\n|[\r\n\v\f\x1c-\x1e\x85\u2028\u2029]") +_UNPRINTABLE = re.compile(r"[\x00-\x08\x0b-\x1f\x7f-\x9f\ud800-\udfff\ufffe\uffff]") +_MARKDOWN_SPECIAL = re.compile(r"[\\`*_\[\]<>&#|~$]") +_MARKDOWN_DESTINATION_UNSAFE = re.compile(r"[<>\s]") +_FILENAME_SEPARATORS = regex.compile(r"[^\p{L}\p{M}\p{N}]+") +_COMBINING_MARK = regex.compile(r"\p{M}") +_FILENAME_NAME_LIMIT = 60 + + +@dataclass(frozen=True) +class FindingExportFile: + """A generated findings document and the name to download it as.""" + + content: bytes + media_type: str + filename: str + + +@dataclass(frozen=True) +class _ReportField: + label: str + value: str + column: Literal["finding", "assessment"] = "finding" + + +_NO_EVIDENCE = _ReportField(label="Evidence", value="none attached") + + +@dataclass(frozen=True) +class _ReportEvidence: + title: str + summary: tuple[_ReportField, ...] + details: tuple[_ReportField, ...] + link_url: str | None + + @property + def unavailable(self) -> bool: + return self.link_url is None + + +@dataclass(frozen=True) +class _ReportFinding: + title: str + severity: _ReportField + fields: tuple[_ReportField, ...] + evidence: tuple[_ReportEvidence, ...] + + +@dataclass(frozen=True) +class _FindingReport: + operation_name: str + findings: tuple[_ReportFinding, ...] + + +async def export_findings_async(*, operation_id: UUID, request: FindingExportRequest) -> FindingExportFile: + """ + Build one document of an Operation's selected findings, in the dashboard's listing order, with evidence references. + + Evidence appears as saved metadata and dashboard links, without transcripts or media. + + Returns: + FindingExportFile: The complete document. + + Raises: + OperationNotFoundError: If the Operation does not exist. + FindingNotFoundError: If a selected finding is not in the Operation. + """ + report = await _build_report_async(operation_id=operation_id, request=request) + now = datetime.now(UTC) + if request.format == FindingExportFormat.MARKDOWN: + extension, media_type = "md", _MARKDOWN_MEDIA_TYPE + content = _render_markdown(report).encode("utf-8") + else: + extension, media_type = "docx", _DOCX_MEDIA_TYPE + content = await asyncio.to_thread(_render_docx, report, now=now) + filename = _export_filename( + operation_id=operation_id, operation_name=report.operation_name, extension=extension, now=now + ) + return FindingExportFile(content=content, media_type=media_type, filename=filename) + + +def _export_filename(*, operation_id: UUID, operation_name: str, extension: str, now: datetime) -> str: + """ + Name a download after its Operation, with the time in labelled UTC. + + Browsers save under the dashboard's own name, which uses the user's local time instead. + + Returns: + str: For example ``copyrit-findings-synthetic-evaluation-2026-10-22-1658-utc.docx``. + """ + name = _FILENAME_SEPARATORS.sub("-", unicodedata.normalize("NFC", operation_name).lower()).strip("-") + if len(name) > _FILENAME_NAME_LIMIT: + end = _FILENAME_NAME_LIMIT + if name[end] != "-" and "-" in name[:end]: + end = name.rindex("-", 0, end) + # A word longer than the limit is cut between letters, never between a letter and its marks. + while end > 0 and _COMBINING_MARK.match(name[end]): + end -= 1 + name = name[:end].rstrip("-") + return f"copyrit-findings-{name or operation_id}-{now.astimezone(UTC):%Y-%m-%d-%H%M}-utc.{extension}" + + +async def _build_report_async(*, operation_id: UUID, request: FindingExportRequest) -> _FindingReport: + service = OperationService() + operation = await service.get_async(operation_id) + findings = await _load_selected_findings_async( + service=service, operation_id=operation_id, finding_ids=request.finding_ids + ) + report_findings = [] + for finding in findings: + evidence = await _load_evidence_async(service=service, finding=finding) + severity, fields = _project_finding_fields(finding) + report_findings.append( + _ReportFinding( + title=finding.title, + severity=severity, + fields=fields, + evidence=tuple( + _project_evidence( + item=item, operation_id=operation_id, dashboard_base_url=request.dashboard_base_url + ) + for item in evidence + ), + ) + ) + return _FindingReport(operation_name=operation.name, findings=tuple(report_findings)) + + +async def _load_selected_findings_async( + *, service: OperationService, operation_id: UUID, finding_ids: list[UUID] +) -> list[Finding]: + """ + Read the selected findings in the order ``OperationService.list_findings_async`` lists them for the dashboard. + + Returns: + list[Finding]: Every selected finding. + + Raises: + OperationNotFoundError: If the Operation does not exist. + FindingNotFoundError: If a selected finding is not in the Operation. + """ + wanted = set(finding_ids) + selected: dict[UUID, Finding] = {} + offset: int | None = 0 + while offset is not None and len(selected) < len(wanted): + page = await service.list_findings_async(operation_id=operation_id, limit=_PAGE_SIZE, offset=offset) + for finding in page.items: + # A concurrent change can shift a row onto the next page, so keep its first read. + if finding.id in wanted: + selected.setdefault(finding.id, finding) + offset = page.next_offset + missing = next((finding_id for finding_id in finding_ids if finding_id not in selected), None) + if missing is not None: + raise FindingNotFoundError(f"Finding '{missing}' not found in operation '{operation_id}'.") + return list(selected.values()) + + +async def _load_evidence_async(*, service: OperationService, finding: Finding) -> list[FindingEvidenceItem]: + items: dict[UUID, FindingEvidenceItem] = {} + offset: int | None = 0 + while offset is not None: + page = await service.list_finding_evidence_async( + operation_id=finding.operation_id, finding_id=finding.id, limit=_PAGE_SIZE, offset=offset + ) + for item in page.items: + items.setdefault(item.item.id, item) + offset = page.next_offset + return list(items.values()) + + +def _project_finding_fields(finding: Finding) -> tuple[_ReportField, tuple[_ReportField, ...]]: + """ + Label a finding's severity and its populated fields, in report order, and place each in a Word column. + + Narrative text goes in the Finding column and classifications in the Assessment column. + + Returns: + tuple[_ReportField, tuple[_ReportField, ...]]: The severity, then the populated fields. + """ + severity = ( + finding.severity_other or "Other" + if finding.severity == FindingSeverity.OTHER + else finding.severity.value.capitalize() + ) + harm_type = finding.harm_type_other if finding.harm_type == HarmCategory.OTHER else finding.harm_type + fields = [ + _ReportField(label="Harm-type", value=str(harm_type), column="assessment") if harm_type else None, + _ReportField(label="Description", value=finding.description) if finding.description else None, + ] + return _ReportField(label="Severity", value=severity), tuple(field for field in fields if field) + + +def _project_evidence( + *, item: FindingEvidenceItem, operation_id: UUID, dashboard_base_url: AnyHttpUrl +) -> _ReportEvidence: + if item.outcome is None: + # The attack result is gone, so its saved ID stands in for its objective. + return _ReportEvidence(title=str(item.item.attack_result_id), summary=(), details=(), link_url=None) + objective = item.objective if item.objective and item.objective.strip() else "No objective" + summary = ( + _ReportField(label="Outcome", value=item.outcome.value.capitalize()), + _ReportField(label="Score", value=_score_text(item.score)), + ) + details = [] + if item.score is not None and item.score.score_rationale: + details.append(_ReportField(label="Rationale", value=item.score.score_rationale)) + # As on the dashboard, an attack's Operation is named only when it isn't the document's own. + if item.source_operation_id != operation_id: + details.append(_ReportField(label="From operation", value=item.source_operation_name or "Not in an operation")) + link_url = None + if item.availability == EvidenceAvailability.AVAILABLE and item.conversation_id is not None: + link_url = _conversation_link( + dashboard_base_url=dashboard_base_url, item=item, conversation_id=item.conversation_id + ) + return _ReportEvidence(title=objective, summary=summary, details=tuple(details), link_url=link_url) + + +def _score_text(score: ScoreView | None) -> str: + if score is None: + return "Not scored" + if score.status == ScoreStatus.UNDETERMINED: + value = "Undetermined" + else: + value = _BOOLEAN_SCORES.get(str(score.score_value).lower(), str(score.score_value)) + scorer = "human review" if score.scorer_type == _HUMAN_SCORER else score.scorer_type + return f"{value} ({scorer})" + + +def _conversation_link(*, dashboard_base_url: AnyHttpUrl, item: FindingEvidenceItem, conversation_id: str) -> str: + """ + Build the dashboard route that opens evidence's conversation and returns to the evidence. + + Returns: + str: The absolute link, matching the dashboard's ``attackConversationRoutePath``. + """ + path = ( + f"/attacks/{_encode_segment(str(item.item.attack_result_id))}/conversations/{_encode_segment(conversation_id)}" + ) + query = {"scenarioResultId": str(item.scenario_result_id)} if item.scenario_result_id else {} + query["findingEvidenceId"] = str(item.item.id) + return f"{str(dashboard_base_url).rstrip('/')}{path}?{urlencode(query)}" + + +def _encode_segment(value: str) -> str: + # The same characters JavaScript's encodeURIComponent leaves unescaped. + return quote(value, safe="!*'()") + + +def _render_docx(report: _FindingReport, *, now: datetime) -> bytes: + document = Document() + title = _plain_text(f"Findings - {report.operation_name}") + _replace_template_metadata(document=document, title=title, now=now) + _replace_theme_formatting(document) + _use_current_word_spacing(document) + document.add_heading(title, level=1) + document.add_paragraph(_EVIDENCE_NOTE) + table = document.add_table(rows=1, cols=len(_WORD_COLUMNS)) + table.style = document.styles[_WORD_TABLE_STYLE] + header = table.rows[0] + for cell, column in zip(header.cells, _WORD_COLUMNS, strict=True): + cell.paragraphs[0].add_run(column).bold = True + for finding in report.findings: + finding_cell, assessment_cell = table.add_row().cells + finding_cell.paragraphs[0].add_run(_plain_text(finding.title)).bold = True + for field in finding.fields: + if field.column == "finding": + _write_word_field(paragraph=finding_cell.add_paragraph(), field=field) + _add_word_evidence(cell=finding_cell, evidence=finding.evidence) + _write_word_field(paragraph=assessment_cell.paragraphs[0], field=finding.severity) + for field in finding.fields: + if field.column == "assessment": + _write_word_field(paragraph=assessment_cell.add_paragraph(), field=field) + buffer = BytesIO() + document.save(buffer) + return buffer.getvalue() + + +def _replace_template_metadata(*, document: DocxDocument, title: str, now: datetime) -> None: + """Describe the export rather than python-docx's template, and open it outside Word's Compatibility Mode.""" + for setting in document.settings.element.iter(qn("w:compatSetting")): + if setting.get(qn("w:name")) == "compatibilityMode": + setting.set(qn("w:val"), _WORD_COMPATIBILITY_MODE) + properties = document.core_properties + properties.title = title + properties.author = properties.comments = properties.last_modified_by = "" + properties.created = properties.modified = now.astimezone(UTC) + properties.revision = 1 + # The template's sample preview image and "Microsoft Macintosh Word" application details don't describe this file. + package_relationships = document.part.package.rels + for relationship_id, relationship in list(package_relationships.items()): + if relationship.reltype in (RELATIONSHIP_TYPE.EXTENDED_PROPERTIES, RELATIONSHIP_TYPE.THUMBNAIL): + del package_relationships[relationship_id] + + +def _replace_theme_formatting(document: DocxDocument) -> None: + """Name Latin fonts and text colors explicitly rather than through the template's 2007-era theme.""" + styles = document.styles.element + for fonts in styles.iter(qn("w:rFonts")): + theme = fonts.get(qn("w:asciiTheme")) + if theme is None: + continue + family = _WORD_HEADING_FONT if theme.startswith("major") else _WORD_BODY_FONT + for attribute in ("w:asciiTheme", "w:hAnsiTheme"): + fonts.attrib.pop(qn(attribute), None) + fonts.set(qn("w:ascii"), family) + fonts.set(qn("w:hAnsi"), family) + for color in styles.iter(qn("w:color")): + for attribute in _THEME_COLOR_ATTRIBUTES: + color.attrib.pop(qn(attribute), None) + + +def _use_current_word_spacing(document: DocxDocument) -> None: + """Give body text and the title the spacing and look of a new Word document.""" + normal = document.styles["Normal"].paragraph_format + normal.space_after = Pt(8) + normal.line_spacing = 1.08 + title = document.styles["Heading 1"] + title.font.color.rgb = _WORD_TITLE_COLOR + title.font.size = _WORD_TITLE_SIZE + title.font.bold = False + title.paragraph_format.space_before = Pt(0) + title.paragraph_format.space_after = Pt(4) + + +def _add_word_evidence(*, cell: _Cell, evidence: tuple[_ReportEvidence, ...]) -> None: + if not evidence: + _write_word_field(paragraph=cell.add_paragraph(), field=_NO_EVIDENCE) + return + cell.add_paragraph().add_run(f"{_NO_EVIDENCE.label}:").bold = True + for entry in evidence: + lead = cell.add_paragraph(style=_WORD_BULLET_STYLE) + lead.paragraph_format.space_before = Pt(6) + lead.paragraph_format.space_after = Pt(2) + if entry.link_url is not None: + _add_word_hyperlink(paragraph=lead, url=entry.link_url, text=_plain_text(entry.title)) + else: + lead.add_run(_plain_text(entry.title)).bold = True + if entry.summary: + line = _indented_paragraph(cell) + for index, field in enumerate(entry.summary): + if index: + line.add_run(_DETAIL_SEPARATOR) + _write_word_field(paragraph=line, field=field) + for field in entry.details: + _write_word_field(paragraph=_indented_paragraph(cell), field=field) + if entry.unavailable: + _indented_paragraph(cell).add_run(_UNAVAILABLE) + + +def _indented_paragraph(cell: _Cell) -> Paragraph: + paragraph: Paragraph = cell.add_paragraph() + paragraph.paragraph_format.left_indent = _WORD_BULLET_TEXT_INDENT + paragraph.paragraph_format.space_after = Pt(2) + return paragraph + + +def _write_word_field(*, paragraph: Paragraph, field: _ReportField) -> None: + paragraph.add_run(f"{_plain_text(field.label)}: ").bold = True + paragraph.add_run(_plain_text(field.value)) + + +def _add_word_hyperlink(*, paragraph: Paragraph, url: str, text: str) -> None: + hyperlink = OxmlElement("w:hyperlink") + hyperlink.set(qn("r:id"), paragraph.part.relate_to(url, RELATIONSHIP_TYPE.HYPERLINK, is_external=True)) + run = paragraph.add_run(text) + run.font.underline = True + run.font.color.rgb = _LINK_COLOR + hyperlink.append(run._r) + paragraph._p.append(hyperlink) + + +def _plain_text(value: str) -> str: + # Both formats break lines at the same characters, and mark each control character visibly rather than drop it. + return _UNPRINTABLE.sub("\ufffd", _LINE_BREAK.sub("\n", value)) + + +def _render_markdown(report: _FindingReport) -> str: + """ + Lay the report out like the Word table, read top to bottom. + + Each finding lists its Assessment fields, then its Finding fields, then its evidence as bullets. + + Returns: + str: The Markdown document. + """ + blocks = [f"# Findings - {_markdown_inline(report.operation_name)}", _EVIDENCE_NOTE] + for finding in report.findings: + blocks.append(f"## {_markdown_inline(finding.title)}") + assessment = [field for field in finding.fields if field.column == "assessment"] + account = [field for field in finding.fields if field.column == "finding"] + blocks.extend(_markdown_field(field) for field in (finding.severity, *assessment, *account)) + blocks.append(_markdown_evidence(finding.evidence)) + return "\n\n".join(blocks) + "\n" + + +def _markdown_evidence(evidence: tuple[_ReportEvidence, ...]) -> str: + if not evidence: + return _markdown_field(_NO_EVIDENCE) + lines = [f"**{_NO_EVIDENCE.label}:**", ""] + for entry in evidence: + title = _markdown_inline(entry.title) + if entry.link_url is not None: + lines.append(f"- [{title}](<{_MARKDOWN_DESTINATION_UNSAFE.sub(_percent_encode, entry.link_url)}>)") + else: + lines.append(f"- **{title}**") + items = [] + if entry.summary and all(_is_short(field) for field in entry.summary): + items.append(_DETAIL_SEPARATOR.join(_markdown_field(field) for field in entry.summary)) + else: + items.extend(_markdown_field(field) for field in entry.summary) + items.extend(_markdown_field(field) for field in entry.details) + if entry.unavailable: + items.append(_UNAVAILABLE) + lines.extend(_markdown_sub_item(item) for item in items) + return "\n".join(lines) + + +def _markdown_sub_item(text: str) -> str: + # Continuation lines, such as a fenced block's, are indented to the sub-item's content so they stay inside it. + first, *rest = text.split("\n") + return "\n".join([f" - {first}", *(f" {line}" if line else "" for line in rest)]) + + +def _is_short(field: _ReportField) -> bool: + return "\n" not in _plain_text(field.value) + + +def _markdown_field(field: _ReportField) -> str: + if _is_short(field): + return f"**{_markdown_inline(field.label)}:** {_markdown_inline(field.value)}" + return f"**{_markdown_inline(field.label)}**\n\n{_markdown_literal(field.value)}" + + +def _markdown_inline(value: str) -> str: + """ + Make text safe to place after other text on a Markdown line. + + Returns: + str: The text on one line, with Markdown and HTML syntax characters escaped. + """ + return _MARKDOWN_SPECIAL.sub(r"\\\g<0>", re.sub(r"\n+", " ", _plain_text(value))) + + +def _markdown_literal(value: str) -> str: + """ + Fence text as a literal block that no line of the text can close. + + Returns: + str: The fenced block. + """ + text = _plain_text(value) + fence = "`" * max(3, max((len(run) for run in re.findall(r"`+", text)), default=0) + 1) + return f"{fence}text\n{text}\n{fence}" + + +def _percent_encode(match: re.Match[str]) -> str: + return quote(match.group(0), safe="") diff --git a/tests/unit/backend/test_finding_export.py b/tests/unit/backend/test_finding_export.py new file mode 100644 index 0000000000..00fa661d72 --- /dev/null +++ b/tests/unit/backend/test_finding_export.py @@ -0,0 +1,1152 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +import re +import xml.etree.ElementTree as ET +import zipfile +from collections.abc import Mapping, Sequence +from datetime import datetime, timedelta, timezone +from io import BytesIO +from unittest.mock import patch +from uuid import UUID, uuid4 + +import pytest +from docx import Document +from docx.oxml.ns import qn +from docx.table import Table +from fastapi.testclient import TestClient + +from pyrit.backend.main import app +from pyrit.backend.models.attacks import ScoreView +from pyrit.backend.models.operations import ( + EvidenceAvailability, + FindingEvidenceItem, + FindingExportRequest, + FindingListResponse, +) +from pyrit.backend.services import finding_export +from pyrit.backend.services.finding_export import ( + _FindingReport, + _project_evidence, + _render_docx, + _render_markdown, + _ReportEvidence, + _ReportField, + _ReportFinding, +) +from pyrit.backend.services.operation_service import OperationService +from pyrit.memory import SQLiteMemory +from pyrit.models import ( + AttackOutcome, + AttackResult, + ComponentIdentifier, + ConversationReference, + ConversationType, + Finding, + FindingEvidence, + Message, + MessagePiece, + Operation, + Score, + ScoreStatus, +) +from pyrit.models.harm_category import HarmCategory + +pytestmark = pytest.mark.usefixtures("patch_central_database") + +FORMATS = ["docx", "markdown"] +DASHBOARD_BASE = "https://pyrit.example.test/team(a)/" +TRANSCRIPT_MARKER = "SYNTHETIC_TRANSCRIPT_NOT_FOR_EXPORT" +MISSING_ID = UUID(int=404) +DOCX_MEDIA_TYPE = "application/vnd.openxmlformats-officedocument.wordprocessingml.document" +MEDIA_TYPES = {"docx": DOCX_MEDIA_TYPE, "markdown": "text/markdown; charset=utf-8"} +EXTENSIONS = {"docx": "docx", "markdown": "md"} +UTC_STAMP = r"\d{4}-\d{2}-\d{2}-\d{4}-utc" +FILENAME_OPERATION_ID = UUID("c1e5502b-91f6-4d65-b182-4b028c7a75f6") +# 12:58 in UTC-4 is 16:58 UTC. +FILENAME_TIME = datetime(2026, 10, 22, 12, 58, 33, tzinfo=timezone(timedelta(hours=-4))) +HYPERLINK_RELATIONSHIP = "http://schemas.openxmlformats.org/officeDocument/2006/relationships/hyperlink" +WORD_NAMESPACE = "{http://schemas.openxmlformats.org/wordprocessingml/2006/main}" +RELATIONSHIP_ID = "{http://schemas.openxmlformats.org/officeDocument/2006/relationships}id" +EVIDENCE_NOTE = "Evidence is listed as references with links to the PyRIT dashboard; transcripts are not included." +# A saved conversation ID with path, query, and fragment characters, and its encodeURIComponent form. +ODD_CONVERSATION_ID = "branch/α (1)?#%" +ENCODED_CONVERSATION_ID = "branch%2F%CE%B1%20(1)%3F%23%25" +HOSTILE_BODY = ( + "Literal ```` run\n````\n# Forged finding\n\n" + "![image](https://evil.example/i.png)\n[link](https://evil.example)\nÜnïcødé 😀 中文\n" +) +NOT_SCORED = _ReportField(label="Score", value="Not scored") +NOT_IN_AN_OPERATION = _ReportField(label="From operation", value="Not in an operation") + + +def _client() -> TestClient: + return TestClient(app, raise_server_exceptions=False) + + +def _export( + *, + headers: dict[str, str], + operation_id: UUID, + finding_ids: Sequence[UUID | str], + export_format: str | None = "docx", +): + body: dict[str, object] = { + "finding_ids": [str(finding_id) for finding_id in finding_ids], + "dashboard_base_url": DASHBOARD_BASE, + } + if export_format is not None: + body["format"] = export_format + return _client().post(f"/api/operations/{operation_id}/findings/export", json=body, headers=headers) + + +async def _report_async(*, operation_id: UUID, finding_ids: Sequence[UUID]) -> _FindingReport: + request = FindingExportRequest.model_validate( + {"finding_ids": list(finding_ids), "dashboard_base_url": DASHBOARD_BASE} + ) + return await finding_export._build_report_async(operation_id=operation_id, request=request) + + +def _dashboard_link(*, evidence: FindingEvidence, conversation_id: str) -> str: + return ( + f"https://pyrit.example.test/team(a)/attacks/{evidence.attack_result_id}/conversations/{conversation_id}" + f"?findingEvidenceId={evidence.id}" + ) + + +def _paragraphs(content: bytes) -> list[str]: + """Every paragraph's text in document order, including table cells (each row's left cell, then its right).""" + texts: list[str] = [] + for block in Document(BytesIO(content)).iter_inner_content(): + if isinstance(block, Table): + texts.extend(paragraph.text for row in block.rows for cell in row.cells for paragraph in cell.paragraphs) + else: + texts.append(block.text) + return texts + + +def _finding_cells(content: bytes) -> list[tuple[list[str], list[str]]]: + """Each finding's Finding and Assessment cell paragraphs, below the table's header row.""" + table = Document(BytesIO(content)).tables[0] + rows = [[[paragraph.text for paragraph in cell.paragraphs] for cell in row.cells] for row in table.rows[1:]] + return [(finding, assessment) for finding, assessment in rows] + + +def _finding_titles(content: bytes, export_format: str) -> list[str]: + """Finding titles in document order: Markdown headings, or the first line of each Word row.""" + if export_format == "markdown": + return [line[3:] for line in content.decode("utf-8").split("\n") if line.startswith("## ")] + return [finding[0] for finding, _ in _finding_cells(content)] + + +def _links(content: bytes, export_format: str) -> list[tuple[str, str]]: + """Read each hyperlink's text and target from the document's own structure.""" + if export_format == "markdown": + pattern = r"^- \[(.+)\]\(<([^>\n]+)>\)$" + return [(match[1], match[2]) for match in re.finditer(pattern, content.decode("utf-8"), flags=re.MULTILINE)] + with zipfile.ZipFile(BytesIO(content)) as package: + relationships = ET.fromstring(package.read("word/_rels/document.xml.rels")) + document = ET.fromstring(package.read("word/document.xml")) + targets = { + relationship.attrib["Id"]: relationship.attrib["Target"] + for relationship in relationships + if relationship.get("Type") == HYPERLINK_RELATIONSHIP and relationship.get("TargetMode") == "External" + } + return [ + ("".join(text.text or "" for text in link.iter(f"{WORD_NAMESPACE}t")), targets[link.attrib[RELATIONSHIP_ID]]) + for link in document.iter(f"{WORD_NAMESPACE}hyperlink") + ] + + +def _bold_paragraph(content: bytes, text: str) -> bool: + """Whether the Word paragraph with exactly ``text``, in the body or a table cell, is bold throughout.""" + document = Document(BytesIO(content)) + cells = [cell for table in document.tables for row in table.rows for cell in row.cells] + paragraphs = [*document.paragraphs, *(paragraph for cell in cells for paragraph in cell.paragraphs)] + paragraph = next(paragraph for paragraph in paragraphs if paragraph.text == text) + return all(run.bold for run in paragraph.runs) + + +def _assert_attachment(*, status_code: int, headers: Mapping[str, str], filename_slug: str, export_format: str) -> None: + assert status_code == 200 + assert headers["content-type"] == MEDIA_TYPES[export_format] + assert re.fullmatch( + rf'attachment; filename="copyrit-findings-{filename_slug}-{UTC_STAMP}\.{EXTENSIONS[export_format]}"', + headers["content-disposition"], + ) + + +async def _add_finding_async(memory: SQLiteMemory, *, operation: Operation, title: str, **fields: object) -> Finding: + fields.setdefault("severity", "moderate") + return await memory.add_finding_async(finding=Finding(operation_id=operation.id, title=title, **fields)) + + +async def _add_attack_async(memory: SQLiteMemory, *, conversation_id: str, **fields: object) -> AttackResult: + attack = AttackResult(conversation_id=conversation_id, **fields) + await memory.add_attack_results_to_memory_async(attack_results=[attack]) + await memory.add_message_to_memory_async( + request=Message( + message_pieces=[ + MessagePiece(role="user", original_value=TRANSCRIPT_MARKER, conversation_id=conversation_id) + ] + ) + ) + return attack + + +async def _link_async(memory: SQLiteMemory, *, finding: Finding, attack_result_id: str | UUID) -> FindingEvidence: + return await memory.add_finding_evidence_async( + evidence=FindingEvidence(finding_id=finding.id, attack_result_id=UUID(str(attack_result_id))) + ) + + +async def _saved_state_async(memory: SQLiteMemory, *, operation: Operation) -> tuple[object, ...]: + findings = await memory.get_findings_async(operation_id=operation.id, limit=1000) + evidence = [await memory.get_finding_evidence_async(finding_id=f.id, limit=1000, offset=0) for f in findings] + attack_ids = sorted({str(item.attack_result_id) for page in evidence for item in page}) + attacks = await memory.get_attack_results_async(attack_result_ids=attack_ids) if attack_ids else [] + return findings, evidence, sorted(attacks, key=lambda attack: attack.attack_result_id) + + +async def _save_named_findings_async( + memory: SQLiteMemory, *, filler_count: int +) -> tuple[Operation, dict[str, Finding]]: + operation = await memory.add_operation_async(Operation(name="Synthetic evaluation")) + named = { + "Critical finding": await _add_finding_async( + memory, + operation=operation, + title="Critical finding", + severity="critical", + harm_type=HarmCategory.PROFANITY, + description="Synthetic description\nsecond line", + ), + "Included finding": await _add_finding_async(memory, operation=operation, title="Included finding"), + "Excluded finding": await _add_finding_async(memory, operation=operation, title="Excluded finding"), + } + for index in range(filler_count): + await _add_finding_async(memory, operation=operation, title=f"Filler {index:02d}") + named["Low finding"] = await _add_finding_async(memory, operation=operation, title="Low finding", severity="low") + attack = await _add_attack_async( + memory, conversation_id="main", objective="Restricted-marker test", operation_id=operation.id + ) + await _link_async(memory, finding=named["Critical finding"], attack_result_id=attack.attack_result_id) + await _link_async(memory, finding=named["Excluded finding"], attack_result_id=attack.attack_result_id) + return operation, named + + +async def _critical_evidence_async(memory: SQLiteMemory, *, named: dict[str, Finding]) -> FindingEvidence: + (evidence,) = await memory.get_finding_evidence_async(finding_id=named["Critical finding"].id, limit=10, offset=0) + return evidence + + +def _single_finding_report( + *, fields: tuple[_ReportField, ...], evidence: tuple[_ReportEvidence, ...] +) -> _FindingReport: + finding = _ReportFinding( + title="Synthetic finding", + severity=_ReportField(label="Severity", value="Low"), + fields=fields, + evidence=evidence, + ) + return _FindingReport(operation_name="Synthetic evaluation", findings=(finding,)) + + +def _split_fenced_blocks(markdown: str) -> tuple[list[str], list[str]]: + """ + Separate lines outside fenced blocks from each block's literal body, following CommonMark's fence rules. + + A fence nested in a list item is indented with the item's content, so that indentation is removed from its body. + """ + outside: list[str] = [] + bodies: list[str] = [] + fence: str | None = None + indent = 0 + body: list[str] = [] + for line in markdown.split("\n"): + if fence is None: + opening = re.match(r"( *)(`{3,}|~{3,})", line) + if opening: + indent, fence, body = len(opening.group(1)), opening.group(2), [] + else: + outside.append(line) + elif re.fullmatch(rf" *{re.escape(fence[0])}{{{len(fence)},}}[ \t]*", line): + bodies.append("\n".join(body)) + fence = None + else: + body.append(line[indent:] if line.startswith(" " * indent) else line.lstrip(" ")) + assert fence is None, "A fenced block was left open" + return outside, bodies + + +@pytest.fixture +async def named_findings_async(sqlite_instance: SQLiteMemory) -> tuple[Operation, dict[str, Finding]]: + return await _save_named_findings_async(sqlite_instance, filler_count=0) + + +@pytest.fixture +async def paged_findings_async(sqlite_instance: SQLiteMemory) -> tuple[Operation, dict[str, Finding]]: + """Save 101 findings, so the lowest severity one is past the first 100-row page.""" + return await _save_named_findings_async(sqlite_instance, filler_count=97) + + +@pytest.mark.parametrize("export_format", FORMATS) +async def test_export_downloads_selected_findings_in_dashboard_order( + sqlite_instance: SQLiteMemory, + compatibility_headers: dict[str, str], + named_findings_async: tuple[Operation, dict[str, Finding]], + export_format: str, +) -> None: + operation, named = named_findings_async + evidence = await _critical_evidence_async(sqlite_instance, named=named) + before = await _saved_state_async(sqlite_instance, operation=operation) + selected = [named[title].id for title in ["Low finding", "Included finding", "Critical finding"]] + + response = _export( + headers=compatibility_headers, operation_id=operation.id, finding_ids=selected, export_format=export_format + ) + + _assert_attachment( + status_code=response.status_code, + headers=response.headers, + filename_slug="synthetic-evaluation", + export_format=export_format, + ) + assert _finding_titles(response.content, export_format) == ["Critical finding", "Included finding", "Low finding"] + assert _links(response.content, export_format) == [ + ("Restricted-marker test", _dashboard_link(evidence=evidence, conversation_id="main")) + ] + assert await _saved_state_async(sqlite_instance, operation=operation) == before + + +async def test_export_defaults_to_docx( + compatibility_headers: dict[str, str], named_findings_async: tuple[Operation, dict[str, Finding]] +) -> None: + operation, named = named_findings_async + + response = _export( + headers=compatibility_headers, + operation_id=operation.id, + finding_ids=[named["Included finding"].id], + export_format=None, + ) + + _assert_attachment( + status_code=response.status_code, + headers=response.headers, + filename_slug="synthetic-evaluation", + export_format="docx", + ) + assert _paragraphs(response.content)[:5] == [ + "Findings - Synthetic evaluation", + EVIDENCE_NOTE, + "Finding", + "Assessment", + "Included finding", + ] + + +@pytest.mark.parametrize( + ("operation_name", "expected_name"), + [ + ("Synthetic evaluation", "synthetic-evaluation"), + ("Q3 Red Team / Copilot (v2)", "q3-red-team-copilot-v2"), + ("Évaluation: modèle 2", "évaluation-modèle-2"), + ("E\u0301valuation", "évaluation"), + # Devanagari writes vowel signs and viramas as combining marks, which belong to their word. + ("हिन्दी परीक्षण", "हिन्दी-परीक्षण"), + ("snake_case name", "snake-case-name"), + ("a" * 58 + " bcdef", "a" * 58), + ("x" * 80, "x" * 60), + # The 61st code point is a vowel sign, so the cut moves back before its letter. + ("क" + "कि" * 30, "क" + "कि" * 29), + ("???", str(FILENAME_OPERATION_ID)), + ], +) +def test_export_filename_names_the_operation_in_labelled_utc(operation_name: str, expected_name: str) -> None: + filename = finding_export._export_filename( + operation_id=FILENAME_OPERATION_ID, operation_name=operation_name, extension="docx", now=FILENAME_TIME + ) + + assert filename == f"copyrit-findings-{expected_name}-2026-10-22-1658-utc.docx" + + +async def test_export_encodes_a_non_ascii_filename_in_the_header( + sqlite_instance: SQLiteMemory, compatibility_headers: dict[str, str] +) -> None: + operation = await sqlite_instance.add_operation_async(Operation(name="Évaluation 中文")) + finding = await _add_finding_async(sqlite_instance, operation=operation, title="Unicode finding") + + response = _export( + headers=compatibility_headers, operation_id=operation.id, finding_ids=[finding.id], export_format="markdown" + ) + + assert response.status_code == 200 + assert re.fullmatch( + rf"attachment; filename\*=utf-8''copyrit-findings-%C3%A9valuation-%E4%B8%AD%E6%96%87-{UTC_STAMP}\.md", + response.headers["content-disposition"], + ) + + +@pytest.mark.parametrize( + ("case", "status_code"), + [ + ("empty", 422), + ("duplicate", 422), + ("malformed", 422), + ("missing_field", 422), + ("unknown_field", 422), + ("unknown_format", 422), + ("foreign", 404), + ("missing_finding", 404), + ("missing_operation", 404), + ], +) +async def test_export_rejects_invalid_selection( + sqlite_instance: SQLiteMemory, + compatibility_headers: dict[str, str], + named_findings_async: tuple[Operation, dict[str, Finding]], + case: str, + status_code: int, +) -> None: + operation, named = named_findings_async + other = await sqlite_instance.add_operation_async(Operation(name="Other operation")) + foreign = await _add_finding_async(sqlite_instance, operation=other, title="Foreign finding") + before = await _saved_state_async(sqlite_instance, operation=operation) + finding_id = str(named["Included finding"].id) + body: dict[str, object] = {"finding_ids": [finding_id], "dashboard_base_url": DASHBOARD_BASE} + path_operation_id: UUID = operation.id + if case == "empty": + body["finding_ids"] = [] + elif case == "duplicate": + body["finding_ids"] = [finding_id, finding_id.upper()] + elif case == "malformed": + body["finding_ids"] = [finding_id, "not-a-uuid"] + elif case == "missing_field": + del body["finding_ids"] + elif case == "unknown_field": + body["all"] = True + elif case == "unknown_format": + body["format"] = "pdf" + elif case == "foreign": + body["finding_ids"] = [finding_id, str(foreign.id)] + elif case == "missing_finding": + body["finding_ids"] = [finding_id, str(MISSING_ID)] + else: + path_operation_id = MISSING_ID + + response = _client().post( + f"/api/operations/{path_operation_id}/findings/export", json=body, headers=compatibility_headers + ) + + assert response.status_code == status_code + assert "content-disposition" not in response.headers + assert await _saved_state_async(sqlite_instance, operation=operation) == before + + +@pytest.mark.parametrize( + "dashboard_base_url", + [ + None, + "", + "/team/", + "pyrit.example.test", + "ftp://pyrit.example.test/", + "javascript:alert(1)", + "******pyrit.example.test/", + "https://pyrit.example.test/?next=/", + "https://pyrit.example.test/?", + "https://pyrit.example.test/#top", + "https://pyrit.example.test/#", + ], +) +def test_export_rejects_invalid_dashboard_base( + compatibility_headers: dict[str, str], dashboard_base_url: str | None +) -> None: + body: dict[str, object] = {"finding_ids": [str(uuid4())]} + if dashboard_base_url is not None: + body["dashboard_base_url"] = dashboard_base_url + + response = _client().post(f"/api/operations/{uuid4()}/findings/export", json=body, headers=compatibility_headers) + + assert response.status_code == 422 + assert "content-disposition" not in response.headers + + +@pytest.mark.parametrize( + ("export_format", "failing"), + [ + ("docx", "get_findings_async"), + ("markdown", "get_attack_results_async"), + ("docx", "_render_docx"), + ("markdown", "_render_markdown"), + ], +) +async def test_export_fails_instead_of_returning_partial_content( + sqlite_instance: SQLiteMemory, + compatibility_headers: dict[str, str], + named_findings_async: tuple[Operation, dict[str, Finding]], + export_format: str, + failing: str, +) -> None: + operation, named = named_findings_async + owner = finding_export if failing.startswith("_render") else sqlite_instance + + with patch.object(owner, failing, side_effect=RuntimeError("synthetic failure")): + response = _export( + headers=compatibility_headers, + operation_id=operation.id, + finding_ids=[named["Critical finding"].id], + export_format=export_format, + ) + + assert response.status_code == 500 + assert "content-disposition" not in response.headers + assert response.json()["type"] == "/errors/internal-error" + + +async def test_report_selects_findings_across_pages_in_dashboard_order( + sqlite_instance: SQLiteMemory, paged_findings_async: tuple[Operation, dict[str, Finding]] +) -> None: + operation, named = paged_findings_async + evidence = await _critical_evidence_async(sqlite_instance, named=named) + # The request order does not override the dashboard's severity-first order. + selected = [named[title].id for title in ["Low finding", "Included finding", "Critical finding"]] + + report = await _report_async(operation_id=operation.id, finding_ids=selected) + + # The attack belongs to the report's own Operation, so its evidence does not name it again. + assert report == _FindingReport( + operation_name="Synthetic evaluation", + findings=( + _ReportFinding( + title="Critical finding", + severity=_ReportField(label="Severity", value="Critical"), + fields=( + _ReportField(label="Harm-type", value="Profanity", column="assessment"), + _ReportField(label="Description", value="Synthetic description\nsecond line"), + ), + evidence=( + _ReportEvidence( + title="Restricted-marker test", + summary=(_ReportField(label="Outcome", value="Undetermined"), NOT_SCORED), + details=(), + link_url=_dashboard_link(evidence=evidence, conversation_id="main"), + ), + ), + ), + _ReportFinding( + title="Included finding", + severity=_ReportField(label="Severity", value="Moderate"), + fields=(), + evidence=(), + ), + _ReportFinding( + title="Low finding", severity=_ReportField(label="Severity", value="Low"), fields=(), evidence=() + ), + ), + ) + + +async def test_report_follows_dashboard_listing_order(sqlite_instance: SQLiteMemory) -> None: + operation = await sqlite_instance.add_operation_async(Operation(name="Listing order")) + findings = [ + await _add_finding_async( + sqlite_instance, operation=operation, title=f"{severity.capitalize()} finding", severity=severity + ) + for severity in ["critical", "moderate", "low"] + ] + list_findings_async = OperationService.list_findings_async + + async def _reversed_listing_async( + service: OperationService, *, operation_id: UUID, limit: int, offset: int + ) -> FindingListResponse: + page = await list_findings_async(service, operation_id=operation_id, limit=limit, offset=offset) + return page.model_copy(update={"items": page.items[::-1]}) + + with patch.object(OperationService, "list_findings_async", autospec=True, side_effect=_reversed_listing_async): + report = await _report_async(operation_id=operation.id, finding_ids=[finding.id for finding in findings]) + + assert [finding.title for finding in report.findings] == ["Low finding", "Moderate finding", "Critical finding"] + + +async def test_report_includes_all_evidence_pages(sqlite_instance: SQLiteMemory) -> None: + operation = await sqlite_instance.add_operation_async(Operation(name="Evidence pages")) + finding = await _add_finding_async(sqlite_instance, operation=operation, title="Many references") + attacks = [ + AttackResult(conversation_id=f"conversation-{index:03d}", objective=f"Objective {index:03d}") + for index in range(101) + ] + await sqlite_instance.add_attack_results_to_memory_async(attack_results=attacks) + await sqlite_instance.add_message_pieces_to_memory_async( + message_pieces=[ + MessagePiece(role="user", original_value=TRANSCRIPT_MARKER, conversation_id=attack.conversation_id) + for attack in attacks + ] + ) + evidence = [ + await _link_async(sqlite_instance, finding=finding, attack_result_id=a.attack_result_id) for a in attacks + ] + + report = await _report_async(operation_id=operation.id, finding_ids=[finding.id]) + + (exported,) = report.findings + assert len(exported.evidence) == 101 + assert set(exported.evidence) == { + _ReportEvidence( + title=f"Objective {index:03d}", + summary=(_ReportField(label="Outcome", value="Undetermined"), NOT_SCORED), + details=(NOT_IN_AN_OPERATION,), + link_url=_dashboard_link(evidence=item, conversation_id=f"conversation-{index:03d}"), + ) + for index, item in enumerate(evidence) + } + + +async def test_report_uses_effective_score_and_conversation(sqlite_instance: SQLiteMemory) -> None: + operation = await sqlite_instance.add_operation_async(Operation(name="Scores")) + source = await sqlite_instance.add_operation_async(Operation(name="Source operation")) + finding = await _add_finding_async(sqlite_instance, operation=operation, title="Scored finding") + pieces = { + conversation_id: MessagePiece( + role="assistant", original_value=TRANSCRIPT_MARKER, conversation_id=conversation_id + ) + for conversation_id in ["main", ODD_CONVERSATION_ID] + } + await sqlite_instance.add_message_pieces_to_memory_async(message_pieces=list(pieces.values())) + + def make_score(conversation_id: str, scorer: str, value: str) -> Score: + return Score( + score_value=value, + score_type="true_false", + score_rationale=f"{scorer} rationale", + scorer_class_identifier=ComponentIdentifier(class_name=scorer, class_module="pyrit.score"), + message_piece_id=pieces[conversation_id].id, + ) + + automated = make_score("main", "SelfAskTrueFalseScorer", "true") + human = make_score(ODD_CONVERSATION_ID, "ManualScorer", "false") + await sqlite_instance.add_scores_to_memory_async(scores=[automated, human]) + attack = AttackResult( + conversation_id="main", + objective="Scored objective", + operation_id=source.id, + automated_score=automated, + human_score=human, + outcome=AttackOutcome.FAILURE, + related_conversations={ + ConversationReference(conversation_id=ODD_CONVERSATION_ID, conversation_type=ConversationType.PRUNED) + }, + ) + await sqlite_instance.add_attack_results_to_memory_async(attack_results=[attack]) + evidence = await _link_async(sqlite_instance, finding=finding, attack_result_id=attack.attack_result_id) + + report = await _report_async(operation_id=operation.id, finding_ids=[finding.id]) + + # The human score is the attack's current score, so the branch it was given in is the one to open. + assert report.findings == ( + _ReportFinding( + title="Scored finding", + severity=_ReportField(label="Severity", value="Moderate"), + fields=(), + evidence=( + _ReportEvidence( + title="Scored objective", + summary=( + _ReportField(label="Outcome", value="Failure"), + _ReportField(label="Score", value="False (human review)"), + ), + details=( + _ReportField(label="Rationale", value="ManualScorer rationale"), + _ReportField(label="From operation", value="Source operation"), + ), + link_url=_dashboard_link(evidence=evidence, conversation_id=ENCODED_CONVERSATION_ID), + ), + ), + ), + ) + + +async def test_report_retains_undetermined_score(sqlite_instance: SQLiteMemory) -> None: + operation = await sqlite_instance.add_operation_async(Operation(name="Undetermined")) + finding = await _add_finding_async(sqlite_instance, operation=operation, title="Undecided finding") + attack = await _add_attack_async(sqlite_instance, conversation_id="main", objective="Undecided objective") + piece = (await sqlite_instance.get_message_pieces_async(conversation_id="main"))[0] + score = Score( + score_value=None, + status=ScoreStatus.UNDETERMINED, + score_type="true_false", + score_rationale="The scorer could not decide.", + scorer_class_identifier=ComponentIdentifier(class_name="SelfAskTrueFalseScorer", class_module="pyrit.score"), + message_piece_id=piece.id, + ) + await sqlite_instance.add_scores_to_memory_async(scores=[score]) + await sqlite_instance.update_attack_result_by_id_async( + attack_result_id=attack.attack_result_id, update_fields={"automated_score_id": str(score.id)} + ) + evidence = await _link_async(sqlite_instance, finding=finding, attack_result_id=attack.attack_result_id) + + report = await _report_async(operation_id=operation.id, finding_ids=[finding.id]) + + (exported,) = report.findings + assert exported.evidence == ( + _ReportEvidence( + title="Undecided objective", + summary=( + _ReportField(label="Outcome", value="Undetermined"), + _ReportField(label="Score", value="Undetermined (SelfAskTrueFalseScorer)"), + ), + details=(_ReportField(label="Rationale", value="The scorer could not decide."), NOT_IN_AN_OPERATION), + link_url=_dashboard_link(evidence=evidence, conversation_id="main"), + ), + ) + + +async def test_report_retains_unavailable_evidence(sqlite_instance: SQLiteMemory) -> None: + operation = await sqlite_instance.add_operation_async(Operation(name="Unavailable")) + finding = await _add_finding_async(sqlite_instance, operation=operation, title="Unavailable finding") + missing_attack_id = uuid4() + await _link_async(sqlite_instance, finding=finding, attack_result_id=missing_attack_id) + messageless = AttackResult(conversation_id="nothing-saved", objective="Messageless objective") + await sqlite_instance.add_attack_results_to_memory_async(attack_results=[messageless]) + await _link_async(sqlite_instance, finding=finding, attack_result_id=messageless.attack_result_id) + + report = await _report_async(operation_id=operation.id, finding_ids=[finding.id]) + + (exported,) = report.findings + assert len(exported.evidence) == 2 + # Without an objective, the saved attack ID leads the entry, as on the dashboard. + assert set(exported.evidence) == { + _ReportEvidence(title=str(missing_attack_id), summary=(), details=(), link_url=None), + _ReportEvidence( + title="Messageless objective", + summary=(_ReportField(label="Outcome", value="Undetermined"), NOT_SCORED), + details=(NOT_IN_AN_OPERATION,), + link_url=None, + ), + } + + +async def test_report_projects_optional_and_custom_fields(sqlite_instance: SQLiteMemory) -> None: + operation = await sqlite_instance.add_operation_async(Operation(name="Classifications")) + custom = await _add_finding_async( + sqlite_instance, + operation=operation, + title="Custom finding", + severity="other", + severity_other="Team severity", + harm_type=HarmCategory.OTHER, + harm_type_other="Team harm", + ) + plain = await _add_finding_async( + sqlite_instance, operation=operation, title="Plain finding", severity="informational", description="Only text" + ) + attack = await _add_attack_async(sqlite_instance, conversation_id="main", objective=" ") + evidence = await _link_async(sqlite_instance, finding=custom, attack_result_id=attack.attack_result_id) + + report = await _report_async(operation_id=operation.id, finding_ids=[custom.id, plain.id]) + + # Other sorts after every named severity, as on the dashboard. + assert report.findings == ( + _ReportFinding( + title="Plain finding", + severity=_ReportField(label="Severity", value="Informational"), + fields=(_ReportField(label="Description", value="Only text"),), + evidence=(), + ), + _ReportFinding( + title="Custom finding", + severity=_ReportField(label="Severity", value="Team severity"), + fields=(_ReportField(label="Harm-type", value="Team harm", column="assessment"),), + evidence=( + _ReportEvidence( + title="No objective", + summary=(_ReportField(label="Outcome", value="Undetermined"), NOT_SCORED), + details=(NOT_IN_AN_OPERATION,), + link_url=_dashboard_link(evidence=evidence, conversation_id="main"), + ), + ), + ), + ) + + +@pytest.mark.parametrize( + ("score_value", "scorer", "expected"), + [ + ("true", "SubStringScorer", "True (SubStringScorer)"), + ("false", "ManualScorer", "False (human review)"), + ("0.75", "SelfAskLikertScorer", "0.75 (SelfAskLikertScorer)"), + ], +) +def test_score_text_capitalizes_booleans_and_names_human_review(score_value: str, scorer: str, expected: str) -> None: + score = Score( + score_value=score_value, + score_type="float_scale" if score_value == "0.75" else "true_false", + score_rationale="Synthetic rationale", + scorer_class_identifier=ComponentIdentifier(class_name=scorer, class_module="pyrit.score"), + message_piece_id=uuid4(), + ) + + assert finding_export._score_text(ScoreView.from_domain(score)) == expected + + +LINK = "https://pyrit.example.test/attacks/a/conversations/b?findingEvidenceId=c" +TABLE_REPORT = _FindingReport( + operation_name="Synthetic evaluation", + findings=( + _ReportFinding( + title="First finding", + severity=_ReportField(label="Severity", value="Critical"), + fields=( + _ReportField(label="Harm-type", value="Synthetic harm", column="assessment"), + _ReportField(label="Description", value="Synthetic description"), + ), + evidence=( + _ReportEvidence( + title="Linked objective", + summary=( + _ReportField(label="Outcome", value="Success"), + _ReportField(label="Score", value="True (SubStringScorer)"), + ), + details=(_ReportField(label="Rationale", value="Synthetic rationale"),), + link_url=LINK, + ), + _ReportEvidence( + title="Unscored objective", + summary=(_ReportField(label="Outcome", value="Failure"), NOT_SCORED), + details=(_ReportField(label="From operation", value="Synthetic source"),), + link_url=LINK, + ), + _ReportEvidence(title="attack-404", summary=(), details=(), link_url=None), + ), + ), + _ReportFinding( + title="Second finding", severity=_ReportField(label="Severity", value="Low"), fields=(), evidence=() + ), + ), +) + + +def test_docx_opens_as_a_current_word_document() -> None: + content = _render_docx(_single_finding_report(fields=(), evidence=()), now=FILENAME_TIME) + + document = Document(BytesIO(content)) + with zipfile.ZipFile(BytesIO(content)) as package: + names = package.namelist() + settings = package.read("word/settings.xml").decode("utf-8") + styles = package.read("word/styles.xml").decode("utf-8") + # Word opens mode 14 files in Compatibility Mode. + assert ( + re.search(r'w:name="compatibilityMode" w:uri="http://schemas.microsoft.com/office/word" w:val="15"', settings) + is not None + ) + properties = document.core_properties + assert properties.title == "Findings - Synthetic evaluation" + assert (properties.author, properties.comments, properties.last_modified_by) == ("", "", "") + # The moment the server's filename names, stored in UTC. + assert properties.created is not None and f"{properties.created:%Y-%m-%d %H:%M:%S}" == "2026-10-22 16:58:33" + assert properties.modified == properties.created + assert not [name for name in names if name in {"docProps/app.xml", "docProps/thumbnail.jpeg"}] + # Latin fonts and text colors are explicit rather than taken from the template's 2007-era theme. + assert re.search(r"w:(ascii|hAnsi)Theme=", styles) is None + assert re.search(r'w:ascii="Calibri"', styles) is not None + title_style = document.styles["Heading 1"].font + assert (str(title_style.color.rgb), title_style.size.pt if title_style.size else None, title_style.bold) == ( + "2F5496", + 16, + False, + ) + assert re.search(r"]*w:themeColor=", styles) is None + + +def test_docx_lists_every_finding_in_one_finding_and_assessment_table() -> None: + content = _render_docx(TABLE_REPORT, now=FILENAME_TIME) + + document = Document(BytesIO(content)) + assert [paragraph.text for paragraph in document.paragraphs] == ["Findings - Synthetic evaluation", EVIDENCE_NOTE] + assert len(document.tables) == 1 + table = document.tables[0] + assert table.style is not None and table.style.name == "Table Grid" + header = table.rows[0] + assert [cell.text for cell in header.cells] == ["Finding", "Assessment"] + assert all(run.bold for cell in header.cells for run in cell.paragraphs[0].runs) + # The header appears once, at the top of the table, rather than repeating on later pages. + assert header._tr.trPr is None or header._tr.trPr.find(qn("w:tblHeader")) is None + assert _finding_cells(content) == [ + ( + [ + "First finding", + "Description: Synthetic description", + "Evidence:", + "Linked objective", + "Outcome: Success · Score: True (SubStringScorer)", + "Rationale: Synthetic rationale", + "Unscored objective", + "Outcome: Failure · Score: Not scored", + # Each entry keeps the same shape: the summary line, then one line per detail. + "From operation: Synthetic source", + "attack-404", + "Evidence unavailable", + ], + ["Severity: Critical", "Harm-type: Synthetic harm"], + ), + (["Second finding", "Evidence: none attached"], ["Severity: Low"]), + ] + assert _bold_paragraph(content, "First finding") + assert _bold_paragraph(content, "attack-404") + assert _links(content, "docx") == [("Linked objective", LINK), ("Unscored objective", LINK)] + # Each entry starts with a bullet, and its lines sit indented under the bullet's text. + evidence_paragraphs = table.rows[1].cells[0].paragraphs[3:] + styles = [paragraph.style.name if paragraph.style else None for paragraph in evidence_paragraphs] + assert styles == ["List Bullet", "Normal", "Normal", "List Bullet", "Normal", "Normal", "List Bullet", "Normal"] + indents = [paragraph.paragraph_format.left_indent for paragraph in evidence_paragraphs] + assert [indent.pt if indent is not None else None for indent in indents] == [ + None, + 18, + 18, + None, + 18, + 18, + None, + 18, + ] + + +def test_markdown_reads_like_the_word_table_top_to_bottom() -> None: + markdown = _render_markdown(TABLE_REPORT) + + # Assessment fields lead under the title, as the Assessment column sits beside it in Word. + assert markdown.split("\n") == [ + "# Findings - Synthetic evaluation", + "", + EVIDENCE_NOTE, + "", + "## First finding", + "", + "**Severity:** Critical", + "", + "**Harm-type:** Synthetic harm", + "", + "**Description:** Synthetic description", + "", + "**Evidence:**", + "", + f"- [Linked objective](<{LINK}>)", + " - **Outcome:** Success · **Score:** True (SubStringScorer)", + " - **Rationale:** Synthetic rationale", + f"- [Unscored objective](<{LINK}>)", + " - **Outcome:** Failure · **Score:** Not scored", + " - **From operation:** Synthetic source", + "- **attack-404**", + " - Evidence unavailable", + "", + "## Second finding", + "", + "**Severity:** Low", + "", + "**Evidence:** none attached", + "", + ] + + +def test_renderers_keep_arbitrary_projected_fields() -> None: + evidence = _ReportEvidence( + title="Source text", + summary=(_ReportField(label="Synthetic evidence field", value="evidence value"),), + details=(_ReportField(label="Synthetic evidence notes", value="note one\nnote two"),), + link_url=None, + ) + report = _FindingReport( + operation_name="Synthetic evaluation", + findings=( + _ReportFinding( + title="Configured finding", + severity=_ReportField(label="Synthetic severity", value="Level 3"), + fields=( + _ReportField(label="Synthetic custom field", value="line one\nline two"), + _ReportField(label="Synthetic narrative", value="part one\npart two"), + _ReportField(label="Synthetic class", value="Class B", column="assessment"), + ), + evidence=(evidence,), + ), + ), + ) + + content = _render_docx(report, now=FILENAME_TIME) + markdown = _render_markdown(report) + + # The projection, not the renderer, decides each field's column; severity always leads the Assessment. + assert _finding_cells(content) == [ + ( + [ + "Configured finding", + "Synthetic custom field: line one\nline two", + "Synthetic narrative: part one\npart two", + "Evidence:", + "Source text", + "Synthetic evidence field: evidence value", + "Synthetic evidence notes: note one\nnote two", + "Evidence unavailable", + ], + ["Synthetic severity: Level 3", "Synthetic class: Class B"], + ) + ] + assert _bold_paragraph(content, "Source text") + # A value that spans lines is fenced rather than flattened, inside an evidence bullet too. + assert markdown.split("\n") == [ + "# Findings - Synthetic evaluation", + "", + EVIDENCE_NOTE, + "", + "## Configured finding", + "", + "**Synthetic severity:** Level 3", + "", + "**Synthetic class:** Class B", + "", + "**Synthetic custom field**", + "", + "```text", + "line one", + "line two", + "```", + "", + "**Synthetic narrative**", + "", + "```text", + "part one", + "part two", + "```", + "", + "**Evidence:**", + "", + "- **Source text**", + " - **Synthetic evidence field:** evidence value", + " - **Synthetic evidence notes**", + "", + " ```text", + " note one", + " note two", + " ```", + " - Evidence unavailable", + "", + ] + + +def test_markdown_fences_literal_body_and_escapes_inline_values() -> None: + link = "https://pyrit.example.test/team(a)/attacks/a/conversations/b?findingEvidenceId=c" + evidence = _ReportEvidence( + title=HOSTILE_BODY, + summary=( + _ReportField(label="Outcome", value="Success\n"), + _ReportField(label="Score", value="True [l](x)"), + ), + details=(_ReportField(label="Rationale", value="`````\n"),), + link_url=link, + ) + report = _FindingReport( + operation_name="Ops bold\n# Forged operation", + findings=( + _ReportFinding( + title="First `code`\n## 9. Forged finding ![i](https://evil.example/i.png) [l](x)", + severity=_ReportField(label="Severity", value="Custom & ~~struck~~ | $x$ # Forged severity"), + fields=(_ReportField(label="Description", value=HOSTILE_BODY),), + evidence=(evidence,), + ), + _ReportFinding( + title="Second Ünïcødé 😀 中文 finding", + severity=_ReportField(label="Severity", value="Low"), + fields=(), + evidence=(), + ), + ), + ) + + markdown = _render_markdown(report) + + outside, bodies = _split_fenced_blocks(markdown) + assert markdown.endswith("\n") + # Values that span lines are fenced, even inside an evidence bullet. + assert bodies == [HOSTILE_BODY, "Success\n", "`````\n"] + headings = [line for line in outside if line.startswith("#")] + assert headings == [ + r"# Findings - Ops \bold\ \# Forged operation", + r"## First \`code\` \#\# 9. Forged finding \ !\[i\](https://evil.example/i.png) \[l\](x)", + "## Second Ünïcødé 😀 中文 finding", + ] + assert r"**Severity:** Custom \& \~\~struck\~\~ \| \$x\$ \# Forged severity" in outside + assert r" - **Score:** True \ \[l\](x)" in outside + assert "**Evidence:** none attached" in outside + # The objective is the link's text, escaped onto one line so it cannot close the link or open another. + link_lines = [line for line in outside if line.startswith("- [") and line.endswith(f"](<{link}>)")] + assert len(link_lines) == 1 + link_text = link_lines[0][len("- [") : -len(f"](<{link}>)")] + assert "\n" not in link_text and re.search(r"(?\[\]`]", link_text) is None + # Authored text never opens HTML, a link, an image, or code outside a literal block. + assert [line for line in outside if line not in link_lines and re.search(r"(?\[\]`]", line)] == [] + + +@pytest.mark.parametrize( + ("dashboard_base_url", "expected_base"), + [ + (DASHBOARD_BASE, "https://pyrit.example.test/team(a)"), + ("https://pyrit.example.test/team(a)", "https://pyrit.example.test/team(a)"), + ("http://localhost:3073", "http://localhost:3073"), + ], +) +def test_renderers_preserve_prefixed_evidence_links(dashboard_base_url: str, expected_base: str) -> None: + request = FindingExportRequest.model_validate({"finding_ids": [uuid4()], "dashboard_base_url": dashboard_base_url}) + scenario_result_id = uuid4() + item = FindingEvidenceItem( + item=FindingEvidence(finding_id=uuid4(), attack_result_id=uuid4()), + availability=EvidenceAvailability.AVAILABLE, + scenario_result_id=scenario_result_id, + objective="Linked objective", + outcome=AttackOutcome.SUCCESS, + conversation_id=ODD_CONVERSATION_ID, + ) + expected = ( + f"{expected_base}/attacks/{item.item.attack_result_id}/conversations/{ENCODED_CONVERSATION_ID}" + f"?scenarioResultId={scenario_result_id}&findingEvidenceId={item.item.id}" + ) + + evidence = _project_evidence(item=item, operation_id=uuid4(), dashboard_base_url=request.dashboard_base_url) + report = _single_finding_report(fields=(), evidence=(evidence,)) + + assert evidence.link_url == expected + assert _links(_render_docx(report, now=FILENAME_TIME), "docx") == [("Linked objective", expected)] + assert _links(_render_markdown(report).encode("utf-8"), "markdown") == [("Linked objective", expected)] + + +def test_docx_keeps_literal_text_and_marks_characters_word_cannot_store() -> None: + report = _single_finding_report( + fields=( + _ReportField(label="Description", value=HOSTILE_BODY), + _ReportField(label="Synthetic control text", value="bell\x07 escape\x1b\r\nnext line"), + ), + evidence=(), + ) + + paragraphs = _paragraphs(_render_docx(report, now=FILENAME_TIME)) + + assert f"Description: {HOSTILE_BODY}" in paragraphs + assert "Synthetic control text: bell\ufffd escape\ufffd\nnext line" in paragraphs + assert "Evidence: none attached" in paragraphs + + +def test_renderers_break_lines_and_mark_control_characters_alike() -> None: + report = _single_finding_report( + fields=( + _ReportField(label="Breaks", value="crlf\r\nvt\vff\ffs\x1cnel\x85ls\u2028ps\u2029end"), + _ReportField(label="Controls", value="c1\x80 c1\x9f del\x7f nul\x00 tab\tend"), + ), + evidence=(), + ) + + paragraphs = _paragraphs(_render_docx(report, now=FILENAME_TIME)) + markdown = _render_markdown(report) + + assert "Breaks: crlf\nvt\nff\nfs\nnel\nls\nps\nend" in paragraphs + assert "Controls: c1\ufffd c1\ufffd del\ufffd nul\ufffd tab\tend" in paragraphs + assert "\n**Breaks**\n\n```text\ncrlf\nvt\nff\nfs\nnel\nls\nps\nend\n```\n" in markdown + assert "\n**Controls:** c1\ufffd c1\ufffd del\ufffd nul\ufffd tab\tend\n" in markdown