diff --git a/README.adoc b/README.adoc index c8d67ae48..ef3c85397 100644 --- a/README.adoc +++ b/README.adoc @@ -293,33 +293,37 @@ VulnScout lets you export and re-import the assessments you have manually create * Sharing assessment decisions across different VulnScout instances. * Restoring triage state in CI pipelines after a database reset. -==== Exporting Custom Assessments +==== VulnScout JSON -The `--export-custom-assessments` flag produces one OpenVEX JSON file per variant, written individually to the outputs directory (default: `.vulnscout/outputs/`). +The `--export-custom-vulnscout-data` flag exports custom assessments, CVSS scores, and time estimates as one VulnScout JSON file. It includes every variant in the selected project unless `--variant` is provided. [source,shell] ---- -# Export as individual files (one per variant) -./vulnscout --project demo --export-custom-assessments +# Export all project variants +./vulnscout --project demo --export-custom-vulnscout-data -# Export as a compressed .tar.gz archive -./vulnscout --project demo --export-custom-assessments --compress +# Export one variant +./vulnscout --project demo --variant x86 --export-custom-vulnscout-data ---- -==== Importing Custom Assessments - -The `--import-custom-assessments` flag reads a `.json` file, `.tar.gz` archive, or a directory of OpenVEX JSON files and replays the assessment statements into the database, matching them to existing vulnerabilities and packages. +The `--import-custom-vulnscout-data` command restores entries according to the variant metadata in the file. [source,shell] ---- -# Import from a single OpenVEX JSON file -./vulnscout --project demo --import-custom-assessments /path/to/assessments.json +./vulnscout --project demo --import-custom-vulnscout-data /path/to/custom_vulnscout_data_all.json +---- + +==== OpenVEX -# Import from a tar.gz archive -./vulnscout --project demo --import-custom-assessments /path/to/custom_assessments.tar.gz +The OpenVEX custom-assessment commands operate on one JSON document and require `--variant` for both import and export. + +[source,shell] +---- +# Export a variant +./vulnscout --project demo --variant x86 --export-custom-openvex-assessments -# Import from a directory of OpenVEX JSON files (one per variant) -./vulnscout --project demo --import-custom-assessments /path/to/assessments/ +# Import into a variant +./vulnscout --project demo --variant x86 --import-custom-openvex-assessments /path/to/custom_openvex_x86.json ---- Both commands can be combined with other flags: @@ -328,7 +332,7 @@ Both commands can be combined with other flags: ---- ./vulnscout --project demo \ --add-spdx /path/to/sbom.spdx.json \ - --import-custom-assessments /path/to/custom_assessments.tar.gz \ + --import-custom-vulnscout-data /path/to/custom_vulnscout_data_all.json \ --export-spdx --report summary.adoc ---- diff --git a/doc/source/api.md b/doc/source/api.md index e3747a90c..0c5b1c559 100644 --- a/doc/source/api.md +++ b/doc/source/api.md @@ -456,15 +456,16 @@ Returns only user-created assessments (not scan-imported). GET /api/assessments/review/export ``` -Downloads a `.tar.gz` archive containing one OpenVEX JSON file per variant. +Downloads one OpenVEX JSON document for a selected variant. **Query parameters:** | Parameter | Type | Description | |-----------|------|-------------| | `author` | string | Author name in the export (default: `"Savoir-faire Linux"`) | +| `variant_id` | UUID | Required. The single variant to export. | -**Response:** Binary `.tar.gz` file (`Content-Type: application/gzip`). +**Response:** OpenVEX JSON file (`Content-Type: application/json`). #### Import Review Assessments @@ -472,9 +473,14 @@ Downloads a `.tar.gz` archive containing one OpenVEX JSON file per variant. POST /api/assessments/review/import ``` -Upload an OpenVEX `.json` or `.tar.gz` file. For `.tar.gz` archives, filenames inside must match variant names. +Upload an OpenVEX `.json` file into a selected variant. The uploaded filename is not used to select the target. -**Request:** Multipart form-data with a `file` field. +**Request:** Multipart form-data with the following fields: + +| Field | Type | Description | +|-------|------|-------------| +| `file` | file | OpenVEX `.json` document (required) | +| `variant_id` | UUID | Required target variant. | **Response:** ```json @@ -494,14 +500,16 @@ Upload an OpenVEX `.json` or `.tar.gz` file. For `.tar.gz` archives, filenames i GET /api/assessments/review/export-custom-data ``` -Exports all handmade (review) assessments, custom CVSS scores, and time estimates as a single JSON file. +Exports handmade review assessments, pending AI assessments, custom CVSS scores, +and time estimates as a single JSON file. Pending AI assessments can be reviewed +from the **AI Assessments** tab after import. **Query parameters:** | Parameter | Type | Description | |-----------|------|-------------| -| `variant_id` | UUID | Restrict to a single variant | -| `project_id` | UUID | Restrict to all variants of a project | +| `variant_id` | UUID | Restrict to selected variants; may be repeated. | +| `project_id` | UUID | Restrict to all variants in a project. | **Response:** JSON file download (`Content-Disposition: attachment; filename=custom_data.json`) containing: @@ -509,6 +517,7 @@ Exports all handmade (review) assessments, custom CVSS scores, and time estimate { "version": "1.0", "assessments": [...], + "ai_assessments": [...], "cvss": [...], "time_estimates": [...] } @@ -522,24 +531,23 @@ Returns `404` if there is no custom data to export. POST /api/assessments/review/import-custom-data ``` -Imports assessments, custom CVSS scores, and time estimates from a custom-data JSON file. +Imports assessments, pending AI assessments, custom CVSS scores, and time +estimates from a custom-data JSON file. Imported AI assessments retain their +pending state and appear in the **AI Assessments** tab. Accepts either: - `multipart/form-data` with a `file` field containing a `.json` file. - `application/json` body with the custom-data payload directly. -**Query parameters:** - -| Parameter | Type | Description | -|-----------|------|-------------| -| `variant_id` | UUID | *(optional)* Force all imported records to a specific variant | +The records keep the variant identifiers embedded in the file. **Response:** ```json { "status": "success", "assessments_imported": 12, + "ai_assessments_imported": 4, "cvss_imported": 5, "time_estimates_imported": 3 } diff --git a/doc/source/container-entrypoint.md b/doc/source/container-entrypoint.md index caa5b56a9..e37e36b8f 100644 --- a/doc/source/container-entrypoint.md +++ b/doc/source/container-entrypoint.md @@ -60,8 +60,10 @@ docker exec vulnscout /scan/src/entrypoint.sh --serve | `--export-spdx` | Export project as SPDX 3.0 SBOM to `/scan/outputs/` | | `--export-cdx` | Export project as CycloneDX 1.6 SBOM to `/scan/outputs/` | | `--export-openvex` | Export project as OpenVEX document to `/scan/outputs/` | -| `--export-custom-assessments` | Export custom (review) assessments of the project as `.tar.gz` (or `.json` if `--variant` is specified) to `/scan/outputs/` | -| `--import-custom-assessments ` | Import custom assessments from `.json` or `.tar.gz` | +| `--export-custom-vulnscout-data` | Export custom VulnScout JSON data for all project variants, or the selected `--variant`, to `/scan/outputs/` | +| `--import-custom-vulnscout-data ` | Import custom VulnScout JSON data using the variant metadata in the file | +| `--export-custom-openvex-assessments` | Export custom assessments for the selected `--variant` as an OpenVEX `.json` file | +| `--import-custom-openvex-assessments ` | Import an OpenVEX `.json` file into the selected `--variant` | | `--match-condition ` | Exit with code 2 if expression matches any vulnerability. Incompatible with `--serve` | | `--delete-scan ` | Delete a past scan by its ID | diff --git a/doc/source/interactive-mode.md b/doc/source/interactive-mode.md index a4308bc65..0bd50983e 100644 --- a/doc/source/interactive-mode.md +++ b/doc/source/interactive-mode.md @@ -298,5 +298,5 @@ The toolbar mirrors the vulnerability table's search bar. Filters are available Two buttons in the toolbar handle review portability: -- **Import Review**: accepts either an OpenVEX file (JSON or `.tar.gz`) or a VulnScout custom-data JSON file and merges its contents into the current project. OpenVEX files import assessments only; custom-data files additionally restore custom CVSS scores and time estimates. This is useful for receiving triage decisions from another team or migrating data between VulnScout instances. -- **Export Review**: downloads all handmade assessments, custom CVSS scores, and time estimates as a single JSON file. The export captures the full set of user-created data so it can be shared, archived, or loaded into another VulnScout deployment. +- **Import Review**: choose either an OpenVEX JSON document or a VulnScout JSON file. OpenVEX imports assessments into one selected variant. VulnScout JSON restores all included assessments, custom CVSS scores, and time estimates using the variants recorded in the file. +- **Export Review**: choose VulnScout JSON to export assessments, custom CVSS scores, and time estimates for selected variants, or OpenVEX to export assessments for one selected variant as a JSON document. diff --git a/doc/source/vulnscout-script.md b/doc/source/vulnscout-script.md index 6dc510a96..c390bf139 100644 --- a/doc/source/vulnscout-script.md +++ b/doc/source/vulnscout-script.md @@ -312,42 +312,40 @@ VulnScout lets you export and re-import the assessments you have manually create - Sharing assessment decisions across different VulnScout instances. - Restoring triage state in CI pipelines after a database reset. -### Exporting Custom Assessments +### VulnScout JSON -The `--export-custom-assessments` flag produces one OpenVEX JSON file per variant, written directly to the outputs directory: +The `--export-custom-vulnscout-data` flag exports custom assessments, pending AI +assessments, CVSS scores, and time estimates as one VulnScout JSON file. It +includes every variant in the selected project unless `--variant` is provided. +After import, pending AI assessments are available from the Review page's **AI +Assessments** tab for approval or rejection. ```bash -./vulnscout --project demo --export-custom-assessments +./vulnscout --project demo --export-custom-vulnscout-data ``` -To bundle the exported files into a `.tar.gz` archive instead, add the `--compress` flag: +Export one variant: ```bash -./vulnscout --project demo --export-custom-assessments --compress +./vulnscout --project demo --variant x86 --export-custom-vulnscout-data ``` -You can also use the `--variant` flag to export only a single variant: +The `--import-custom-vulnscout-data` command restores entries according to the variant metadata stored in the file: ```bash -./vulnscout --project demo --variant x86 --export-custom-assessments +./vulnscout --project demo --import-custom-vulnscout-data /path/to/custom_vulnscout_data_all.json ``` -### Importing Custom Assessments +### OpenVEX -The `--import-custom-assessments` flag reads a `.json` file, `.tar.gz` archive, or a directory of OpenVEX JSON files and replays the assessment statements into the database. If `--variant` is not specified, the variant is inferred from the file name. +OpenVEX custom-assessment transfers operate on one JSON document and require `--variant` for both import and export. ```bash -# Import from a single OpenVEX JSON file -./vulnscout --project demo --variant x86 --import-custom-assessments /path/to/assessments.json - -# Import from a single OpenVEX JSON file without specifying the variant -./vulnscout --project demo --import-custom-assessments /path/to/assessments/x86.json - -# Import from a tar.gz archive -./vulnscout --project demo --import-custom-assessments /path/to/custom_assessments.tar.gz +# Export a variant +./vulnscout --project demo --variant x86 --export-custom-openvex-assessments -# Import from a directory of OpenVEX JSON files (one per variant) -./vulnscout --project demo --import-custom-assessments /path/to/assessments/ +# Import into a variant +./vulnscout --project demo --variant x86 --import-custom-openvex-assessments /path/to/custom_openvex_x86.json ``` --- diff --git a/frontend/src/components/ReviewTransferModal.tsx b/frontend/src/components/ReviewTransferModal.tsx new file mode 100644 index 000000000..5259e18e2 --- /dev/null +++ b/frontend/src/components/ReviewTransferModal.tsx @@ -0,0 +1,106 @@ +import { useEffect } from 'react'; +import { FontAwesomeIcon } from '@fortawesome/react-fontawesome'; +import { faXmark } from '@fortawesome/free-solid-svg-icons'; +import type { Variant } from '../handlers/variant'; + +type Props = { + mode: 'import' | 'export'; + variants: Variant[]; + selectedVariantIds: string[]; + transferFormat: 'custom' | 'openvex'; + onSelectedVariantIdsChange: (ids: string[]) => void; + onTransferFormatChange: (format: 'custom' | 'openvex') => void; + onConfirm: () => void; + onCancel: () => void; +}; + +function ReviewTransferModal({ + mode, + variants, + selectedVariantIds, + transferFormat, + onSelectedVariantIdsChange, + onTransferFormatChange, + onConfirm, + onCancel, +}: Readonly) { + const title = mode === 'export' ? 'Export review data' : 'Import review data'; + const isOpenVex = transferFormat === 'openvex'; + const needsVariantSelection = isOpenVex || mode === 'export'; + const supportsMultipleVariants = !isOpenVex; + + useEffect(() => { + const onKeyDown = (event: KeyboardEvent) => { + if (event.key === 'Escape') onCancel(); + }; + document.addEventListener('keydown', onKeyDown); + return () => document.removeEventListener('keydown', onKeyDown); + }, [onCancel]); + + const toggleVariant = (variantId: string) => { + onSelectedVariantIdsChange( + selectedVariantIds.includes(variantId) + ? selectedVariantIds.filter(id => id !== variantId) + : [...selectedVariantIds, variantId], + ); + }; + + return ( +
+
+
+

{title}

+ +
+ +
+
+ Format +
+ + +
+
+ + {needsVariantSelection && ( +
+ {supportsMultipleVariants ? 'Variants' : 'Variant'} + {supportsMultipleVariants && ( +
+ +
+ )} +
+ {variants.map(variant => ( + + ))} +
+
+ )} +
+ +
+ + +
+
+
+ ); +} + +export default ReviewTransferModal; diff --git a/frontend/src/helpers/exportJson.ts b/frontend/src/helpers/exportJson.ts index 4dfe26a60..f8e307a1f 100644 --- a/frontend/src/helpers/exportJson.ts +++ b/frontend/src/helpers/exportJson.ts @@ -24,13 +24,17 @@ export function formatTimestampForFilename(date?: Date | string): string { */ export function downloadJson(data: unknown, filename: string): void { const json = JSON.stringify(data, null, 2); - const blob = new Blob([json], { type: 'application/json' }); + downloadBlob(new Blob([json], { type: 'application/json' }), filename); +} + +/** Trigger a download for a response body that is already encoded. */ +export function downloadBlob(blob: Blob, filename: string): void { const url = URL.createObjectURL(blob); - const a = document.createElement('a'); - a.href = url; - a.download = filename; - document.body.appendChild(a); - a.click(); - document.body.removeChild(a); + const anchor = document.createElement('a'); + anchor.href = url; + anchor.download = filename; + document.body.appendChild(anchor); + anchor.click(); + document.body.removeChild(anchor); URL.revokeObjectURL(url); } diff --git a/frontend/src/pages/Review.tsx b/frontend/src/pages/Review.tsx index 3b7cb0a19..bdc5d39f8 100644 --- a/frontend/src/pages/Review.tsx +++ b/frontend/src/pages/Review.tsx @@ -19,9 +19,9 @@ import ConfirmationModal from '../components/ConfirmationModal'; import MessageBanner from '../components/MessageBanner'; import Variants from '../handlers/variant'; import Packages from '../handlers/packages'; -import Projects from '../handlers/project'; import { useDocUrl } from '../helpers/useDocUrl'; import { splitPkgId, extractSupplierName } from '../helpers/pkgId'; +import ReviewTransferModal from '../components/ReviewTransferModal'; type AssessmentMutation = | { type: 'delete'; vulnId: string; ids: string[] } @@ -133,7 +133,6 @@ function Review({ variantId, projectId, onAssessmentChanged }: Readonly) const [showSearchHelper, setShowSearchHelper] = useState(false); const [importStatus, setImportStatus] = useState(null); const [variantNames, setVariantNames] = useState>({}); - const [projectNames, setProjectNames] = useState>({}); const [allVariants, setAllVariants] = useState([]); const [editingRow, setEditingRow] = useState(null); const [editVariants, setEditVariants] = useState([]); @@ -143,6 +142,9 @@ function Review({ variantId, projectId, onAssessmentChanged }: Readonly) const [bannerMessage, setBannerMessage] = useState(""); const [bannerType, setBannerType] = useState<"error" | "success">("success"); const [showBanner, setShowBanner] = useState(false); + const [transferMode, setTransferMode] = useState<'import' | 'export' | null>(null); + const [transferVariantIds, setTransferVariantIds] = useState([]); + const [transferFormat, setTransferFormat] = useState<'custom' | 'openvex'>('custom'); const showMessage = useCallback((message: string, type: "error" | "success") => { setBannerMessage(message); @@ -182,11 +184,6 @@ function Review({ variantId, projectId, onAssessmentChanged }: Readonly) setVariantNames(map); setAllVariants(vs); }).catch(() => {}); - Projects.list().then(ps => { - const map: Record = {}; - for (const p of ps) map[p.id] = p.name; - setProjectNames(map); - }).catch(() => {}); }, []); // When editing a row, only offer variants that actually have a finding for @@ -434,38 +431,66 @@ function Review({ variantId, projectId, onAssessmentChanged }: Readonly) setSelectedVariants(variantList); }; + const transferVariants = useMemo(() => { + const scopedProjectId = projectId ?? allVariants.find(v => v.id === variantId)?.project_id; + return scopedProjectId ? allVariants.filter(v => v.project_id === scopedProjectId) : allVariants; + }, [allVariants, projectId, variantId]); + + const openTransfer = useCallback((mode: 'import' | 'export') => { + setTransferFormat('custom'); + setTransferVariantIds(mode === 'export' + ? variantId ? [variantId] : transferVariants.map(variant => variant.id) + : []); + setTransferMode(mode); + }, [transferVariants, variantId]); + + const changeTransferFormat = useCallback((format: 'custom' | 'openvex') => { + setTransferFormat(format); + if (format === 'custom') { + setTransferVariantIds(transferMode === 'export' + ? variantId ? [variantId] : transferVariants.map(variant => variant.id) + : []); + return; + } + const defaultVariantId = variantId && transferVariants.some(v => v.id === variantId) + ? variantId + : transferVariants[0]?.id; + setTransferVariantIds(defaultVariantId ? [defaultVariantId] : []); + }, [transferMode, transferVariants, variantId]); + const handleExportReview = useCallback(async () => { try { const params = new URLSearchParams(); - if (variantId) params.set('variant_id', variantId); - else if (projectId) params.set('project_id', projectId); + transferVariantIds.forEach(variantId => params.append('variant_id', variantId)); + const endpoint = transferFormat === 'openvex' ? 'export' : 'export-custom-data'; const url = new URL( - import.meta.env.VITE_API_URL + "/api/assessments/review/export-custom-data" + + import.meta.env.VITE_API_URL + `/api/assessments/review/${endpoint}` + (params.toString() ? `?${params.toString()}` : ''), window.location.href, ); const res = await fetch(url.toString(), { mode: 'cors' }); if (!res.ok) { const err = await res.json().catch(() => ({})); - showMessage(err.error || 'Failed to export custom data.', 'error'); + showMessage(err.error || 'Failed to export review data.', 'error'); return; } - const data = await res.json(); const ts = formatTimestampForFilename(); - const pName = projectId ? projectNames[projectId] - : variantId ? projectNames[allVariants.find(v => v.id === variantId)?.project_id ?? ''] ?? '' - : ''; - const vName = variantId ? variantNames[variantId] ?? '' : ''; - const label = [pName, vName].filter(Boolean).join('_') || 'all'; - const filename = `custom_data_${sanitizeFilename(label)}_${ts}.json`; - downloadJson(data, filename); + if (transferFormat === 'openvex') { + const label = variantNames[transferVariantIds[0]] ?? 'variant'; + downloadJson(await res.json(), `review_openvex_${sanitizeFilename(label)}_${ts}.json`); + } else { + const label = transferVariantIds.length === 1 ? variantNames[transferVariantIds[0]] ?? 'variant' : 'all'; + downloadJson(await res.json(), `custom_data_${sanitizeFilename(label)}_${ts}.json`); + } + setTransferMode(null); } catch (err) { console.error('Export error:', err); - showMessage('Failed to export custom data.', 'error'); + showMessage('Failed to export review data.', 'error'); } - }, [variantId, projectId, showMessage, projectNames, variantNames, allVariants]); + }, [showMessage, variantNames, transferVariantIds, transferFormat]); const handleImportReview = useCallback(() => { + setTransferMode(null); fileInputRef.current?.click(); }, []); @@ -473,61 +498,54 @@ function Review({ variantId, projectId, onAssessmentChanged }: Readonly) const file = event.target.files?.[0]; if (!file) return; - // Old OpenVEX format: .tar.gz or .tgz files — use legacy import path - if (file.name.endsWith('.tar.gz') || file.name.endsWith('.tgz')) { + type ImportResult = { + status?: string; + error?: string; + assessments_imported?: number; + assessments_skipped?: number; + cvss_imported?: number; + time_estimates_imported?: number; + errors?: { error?: string }[]; + }; + + if (transferFormat === 'openvex') { + setImportStatus("Importing..."); const formData = new FormData(); formData.append('file', file); - const url = new URL(import.meta.env.VITE_API_URL + "/api/assessments/review/import", window.location.href); - setImportStatus("Importing..."); - fetch(url.toString(), { method: 'POST', body: formData, mode: 'cors' }) - .then(res => res.json()) - .then(data => { - if (data.status === 'success') { - Assessments.listReview(variantId, projectId).then(data => setAssessments(groupAssessments(data))); + formData.append('variant_id', transferVariantIds[0]); + fetch(new URL(import.meta.env.VITE_API_URL + "/api/assessments/review/import", window.location.href).toString(), { + method: 'POST', + body: formData, + mode: 'cors', + }) + .then(response => response.json() as Promise) + .then(result => { + if (result.status === 'success') { + Assessments.listReview(variantId, projectId).then(d => setAssessments(groupAssessments(d))); showMessage('Assessments imported successfully!', 'success'); } else { - showMessage(`Import error: ${data.error || 'Unknown error'}`, 'error'); + showMessage(`Import error: ${result.error || 'Unknown error'}`, 'error'); } - setImportStatus(null); }) .catch(err => { console.error(err); showMessage('Import failed', 'error'); - setImportStatus(null); }) .finally(() => { + setImportStatus(null); if (fileInputRef.current) fileInputRef.current.value = ''; }); return; } - // JSON files: detect format and route to appropriate backend endpoint + // VulnScout JSON carries its own variant IDs, so no target selection + // or override is sent with the request. const reader = new FileReader(); reader.onload = async () => { try { const text = reader.result as string; const parsed = JSON.parse(text); - // Detect old OpenVEX format: has @context with "openvex" - if (parsed?.['@context'] && String(parsed['@context']).includes('openvex')) { - const formData = new FormData(); - formData.append('file', file); - const url = new URL(import.meta.env.VITE_API_URL + "/api/assessments/review/import", window.location.href); - setImportStatus("Importing..."); - const res = await fetch(url.toString(), { method: 'POST', body: formData, mode: 'cors' }); - const data = await res.json(); - if (data.status === 'success') { - Assessments.listReview(variantId, projectId).then(d => setAssessments(groupAssessments(d))); - showMessage('Assessments imported successfully!', 'success'); - } else { - showMessage(`Import error: ${data.error || 'Unknown error'}`, 'error'); - } - setImportStatus(null); - if (fileInputRef.current) fileInputRef.current.value = ''; - return; - } - - // New custom data format — send to backend import-custom-data endpoint if (!parsed?.version || !parsed?.assessments) { showMessage('Invalid file format. Expected a VulnScout custom data export.', 'error'); if (fileInputRef.current) fileInputRef.current.value = ''; @@ -535,35 +553,33 @@ function Review({ variantId, projectId, onAssessmentChanged }: Readonly) } setImportStatus("Importing..."); - const params = new URLSearchParams(); - if (variantId) params.set('variant_id', variantId); - const url = new URL( - import.meta.env.VITE_API_URL + "/api/assessments/review/import-custom-data" + - (params.toString() ? `?${params.toString()}` : ''), - window.location.href, - ); - const res = await fetch(url.toString(), { + const url = new URL(import.meta.env.VITE_API_URL + "/api/assessments/review/import-custom-data", window.location.href); + const result = await fetch(url.toString(), { method: 'POST', mode: 'cors', headers: { 'Content-Type': 'application/json' }, body: text, }); - const result = await res.json(); + const data = await result.json() as ImportResult; - if (result.status === 'success') { + if (data.status === 'success') { + Assessments.listReview(variantId, projectId).then(d => setAssessments(groupAssessments(d))); + const assessmentsImported = data.assessments_imported ?? 0; + const assessmentsSkipped = data.assessments_skipped ?? 0; + const cvssImported = data.cvss_imported ?? 0; + const timeEstimatesImported = data.time_estimates_imported ?? 0; const summary: string[] = []; - if (result.assessments_imported > 0) { - let msg = `${result.assessments_imported} assessment(s) imported`; - if (result.assessments_skipped) msg += `, ${result.assessments_skipped} skipped`; + if (assessmentsImported > 0) { + let msg = `${assessmentsImported} assessment(s) imported`; + if (assessmentsSkipped) msg += `, ${assessmentsSkipped} skipped`; summary.push(msg); } - if (result.cvss_imported > 0) summary.push(`${result.cvss_imported} CVSS score(s)`); - if (result.time_estimates_imported > 0) summary.push(`${result.time_estimates_imported} time estimate(s)`); - if (result.errors?.length) summary.push(`${result.errors.length} error(s)`); - Assessments.listReview(variantId, projectId).then(d => setAssessments(groupAssessments(d))); + if (cvssImported > 0) summary.push(`${cvssImported} CVSS score(s)`); + if (timeEstimatesImported > 0) summary.push(`${timeEstimatesImported} time estimate(s)`); + if (data.errors?.length) summary.push(`${data.errors.length} error(s)`); showMessage(summary.length > 0 ? `Imported: ${summary.join(', ')}` : 'Import complete (no data changed)', 'success'); } else { - showMessage(`Import error: ${result.errors?.[0]?.error || 'Unknown error'}`, 'error'); + showMessage(`Import error: ${data.errors?.[0]?.error || data.error || 'Unknown error'}`, 'error'); } setImportStatus(null); } catch (err) { @@ -575,7 +591,7 @@ function Review({ variantId, projectId, onAssessmentChanged }: Readonly) } }; reader.readAsText(file); - }, [variantId, projectId, showMessage]); + }, [variantId, projectId, showMessage, transferFormat, transferVariantIds]); const handleDeleteRow = useCallback(async () => { if (!rowToDelete) return; @@ -1375,32 +1391,45 @@ function Review({ variantId, projectId, onAssessmentChanged }: Readonly) + {transferMode && ( + setTransferMode(null)} + /> + )} + {showBanner && (
({ __esModule: true, ...jest.requireActual('../../src/helpers/exportJson'), downloadJson: jest.fn(), + downloadBlob: jest.fn(), })); import Review from '../../src/pages/Review'; @@ -219,6 +220,11 @@ function mockNetwork(reviewList: unknown[] = [], opts: NetworkOpts = {}): void { ? JSON.stringify({ version: 1, assessments: [] }) : { status: 500, body: JSON.stringify({}) }; } + if (url.includes('/api/assessments/review/export')) { + return exportOk + ? JSON.stringify({ '@context': 'https://openvex.dev/ns/v0.2.0', statements: [] }) + : { status: 500, body: JSON.stringify({}) }; + } if (url.includes('/api/assessments/review/ai')) return JSON.stringify(aiReviewList); if (url.includes('/api/assessments/review')) return JSON.stringify(reviewList); if (/\/api\/vulnerabilities\/[^/]+\/assessments/.test(url)) return JSON.stringify([]); @@ -1300,31 +1306,58 @@ describe('Review — deleting an assessment', () => { // =========================================================================== describe('Review — import and export', () => { - test('exporting custom data downloads a file', async () => { + test('exports selected variants as VulnScout JSON', async () => { mockNetwork([makeAssessment('a1', 'v1')]); render(); const user = userEvent.setup(); await screen.findByTitle('Edit assessment'); - await user.click(screen.getByText('Export Custom Data')); + await user.click(screen.getByText('Export')); + const dialog = screen.getByRole('dialog'); + await user.click(within(dialog).getByRole('checkbox', { name: 'Variant Beta' })); + await user.click(within(dialog).getByRole('button', { name: 'Export' })); await waitFor(() => { expect(mockedDownloadJson).toHaveBeenCalled(); }); + const exportCall = fetchMock.mock.calls.find(call => String(call[0]).includes('/api/assessments/review/export-custom-data')); + expect(String(exportCall?.[0])).toContain('variant_id=v1'); + expect(String(exportCall?.[0])).not.toContain('variant_id=v2'); + expect(mockedDownloadJson.mock.calls[0][1]).toContain('custom_data_Variant_Alpha_'); }); - test('exporting when scoped to a variant builds a variant-labelled filename', async () => { + test('VulnScout JSON export defaults to the current variant', async () => { mockNetwork([makeAssessment('a1', 'v1')]); render(); const user = userEvent.setup(); await screen.findByTitle('Edit assessment'); - await user.click(screen.getByText('Export Custom Data')); + await user.click(screen.getByText('Export')); + await user.click(within(screen.getByRole('dialog')).getByRole('button', { name: 'Export' })); await waitFor(() => { expect(mockedDownloadJson).toHaveBeenCalled(); }); - const filename = mockedDownloadJson.mock.calls[0][1] as string; - expect(filename).toContain('Project_One'); - expect(filename).toContain('Variant_Alpha'); + const exportCall = fetchMock.mock.calls.find(call => String(call[0]).includes('/api/assessments/review/export-custom-data')); + expect(String(exportCall?.[0])).toContain('variant_id=v1'); + }); + + test('exports one selected variant as an OpenVEX JSON document', async () => { + mockNetwork([makeAssessment('a1', 'v1')]); + render(); + const user = userEvent.setup(); + await screen.findByTitle('Edit assessment'); + + await user.click(screen.getByText('Export')); + const dialog = screen.getByRole('dialog'); + await user.click(within(dialog).getByRole('radio', { name: /OpenVEX/ })); + await user.click(within(dialog).getByRole('radio', { name: 'Variant Beta' })); + await user.click(within(dialog).getByRole('button', { name: 'Export' })); + + await waitFor(() => expect(mockedDownloadJson).toHaveBeenCalled()); + const exportCall = fetchMock.mock.calls.find(call => String(call[0]).includes('/api/assessments/review/export?')); + expect(String(exportCall?.[0])).toContain('variant_id=v2'); + expect(String(exportCall?.[0])).not.toContain('variant_id=v1'); + expect(mockedDownloadJson.mock.calls[0][1]).toContain('review_openvex_Variant_Beta_'); + expect(mockedDownloadJson.mock.calls[0][1]).toContain('.json'); }); test('reports an error when export fails', async () => { @@ -1333,8 +1366,9 @@ describe('Review — import and export', () => { const user = userEvent.setup(); await screen.findByTitle('Edit assessment'); - await user.click(screen.getByText('Export Custom Data')); - await screen.findByText('Failed to export custom data.'); + await user.click(screen.getByText('Export')); + await user.click(within(screen.getByRole('dialog')).getByRole('button', { name: 'Export' })); + await screen.findByText('Failed to export review data.'); expect(mockedDownloadJson).not.toHaveBeenCalled(); }); @@ -1344,20 +1378,27 @@ describe('Review — import and export', () => { const user = userEvent.setup(); await screen.findByTitle('Edit assessment'); - await user.click(screen.getByText('Export Custom Data')); - await screen.findByText('Failed to export custom data.'); + await user.click(screen.getByText('Export')); + await user.click(within(screen.getByRole('dialog')).getByRole('button', { name: 'Export' })); + await screen.findByText('Failed to export review data.'); await user.click(screen.getByRole('button', { name: 'Dismiss' })); await waitFor(() => { - expect(screen.queryByText('Failed to export custom data.')).not.toBeInTheDocument(); + expect(screen.queryByText('Failed to export review data.')).not.toBeInTheDocument(); }); }); - test('importing a custom-data JSON file reports a summary', async () => { + test('importing VulnScout JSON has no variant selector and reports a summary', async () => { mockNetwork([makeAssessment('a1', 'v1')]); render(); + const user = userEvent.setup(); await screen.findByTitle('Edit assessment'); + await user.click(screen.getByText('Import')); + const dialog = screen.getByRole('dialog'); + expect(within(dialog).queryByText('Variant')).not.toBeInTheDocument(); + await user.click(within(dialog).getByRole('button', { name: 'Choose file' })); + const file = new File( [JSON.stringify({ version: '1', assessments: [{ id: 'x' }] })], 'custom.json', @@ -1368,34 +1409,38 @@ describe('Review — import and export', () => { await screen.findByText(/Imported:/); }); - test('importing a legacy OpenVEX JSON file succeeds', async () => { + test('imports OpenVEX into one selected variant without using the filename', async () => { mockNetwork([makeAssessment('a1', 'v1')]); render(); + const user = userEvent.setup(); await screen.findByTitle('Edit assessment'); + await user.click(screen.getByText('Import')); + const dialog = screen.getByRole('dialog'); + await user.click(within(dialog).getByRole('radio', { name: /OpenVEX/ })); + await user.click(within(dialog).getByRole('radio', { name: 'Variant Beta' })); + await user.click(within(dialog).getByRole('button', { name: 'Choose file' })); + const file = new File( [JSON.stringify({ '@context': 'https://openvex.dev/ns/v0.2.0', statements: [] })], - 'openvex.json', + 'does-not-match-a-variant.json', { type: 'application/json' }, ); fireEvent.change(fileInput(), { target: { files: [file] } }); await screen.findByText('Assessments imported successfully!'); + const importCalls = postCalls().filter(call => String(call[0]).endsWith('/api/assessments/review/import')); + expect(importCalls).toHaveLength(1); + const targetedVariantIds = importCalls.map(call => ((call[1] as RequestInit).body as FormData).get('variant_id')); + expect(targetedVariantIds).toEqual(['v2']); }); - test('importing a legacy tar.gz file uses the legacy endpoint', async () => { + test('file selection accepts JSON only', async () => { mockNetwork([makeAssessment('a1', 'v1')]); render(); await screen.findByTitle('Edit assessment'); - const file = new File(['binary'], 'export.tar.gz', { type: 'application/gzip' }); - fireEvent.change(fileInput(), { target: { files: [file] } }); - - await screen.findByText('Assessments imported successfully!'); - expect(fetchMock).toHaveBeenCalledWith( - expect.stringContaining('/api/assessments/review/import'), - expect.objectContaining({ method: 'POST' }) - ); + expect(fileInput()).toHaveAttribute('accept', '.json,application/json'); }); test('rejects a JSON file with an unrecognised format', async () => { diff --git a/src/bin/cmd_assessments.py b/src/bin/cmd_assessments.py index 1219be449..f360236da 100644 --- a/src/bin/cmd_assessments.py +++ b/src/bin/cmd_assessments.py @@ -1,13 +1,9 @@ # Copyright (C) 2026 Savoir-faire Linux, Inc. # SPDX-License-Identifier: GPL-3.0-only -"""Custom assessment import/export commands: -``flask export-custom-assessments`` and ``flask import-custom-assessments``.""" +"""VulnScout JSON and OpenVEX custom-assessment import/export commands.""" -import io -import uuid import click import json as _json -import tarfile import os from flask.cli import with_appcontext from ..helpers.assessment_io import ( @@ -16,239 +12,140 @@ build_openvex_doc, sanitize_variant_name, import_statements, - import_archive_bytes, - import_directory, + build_custom_data_export, import_custom_data, ) from ..models.assessment import Assessment as DBAssessment from ..models.variant import Variant as DBVariant from datetime import datetime as _dt, timezone as _tz -from collections import defaultdict from ._common import get_default_author, resolve_project, resolve_project_variant +_JSON_SUFFIX = ".json" + + +def _load_json_file(file_path: str) -> dict: + if not os.path.isfile(file_path): + raise click.ClickException(f"File not found: {file_path}") + if not file_path.endswith(_JSON_SUFFIX): + raise click.ClickException("Unsupported file type. Please provide a .json file.") + try: + with open(file_path) as file: + data = _json.load(file) + except Exception as error: + raise click.ClickException("Invalid JSON file.") from error + if not isinstance(data, dict): + raise click.ClickException("Invalid JSON file.") + return data + + +def _print_custom_data_import_result(result: dict) -> None: + if result.get("status") != "success": + for error in result.get("errors", []): + click.echo(f" {error}", err=True) + raise click.ClickException("Failed to import VulnScout JSON data.") + for error in result.get("errors", []): + click.echo(f" Warning: {error}", err=True) + click.echo( + f"Imported {result['assessments_imported']} assessments" + f" ({result['assessments_skipped']} skipped as duplicates)," + f" {result['ai_assessments_imported']} AI assessments" + f" ({result['ai_assessments_skipped']} skipped as duplicates)," + f" {result['cvss_imported']} CVSS," + f" {result['time_estimates_imported']} time estimates" + ) -@click.command("export-custom-assessments") + +@click.command("export-custom-vulnscout-data") @click.option("--output-dir", default="/scan/outputs", show_default=True, help="Directory where the exported file is written.") @click.option("--project", "-p", required=True, help="Project name.") @click.option("--variant", "-v", default=None, - help="Variant name. If empty, all variants will be exported.") -@click.option("--compress", is_flag=True, default=False, - help="Write a .tar.gz archive instead of individual files.") + help="Variant name. If empty, all project variants are exported.") @with_appcontext -def export_custom_assessments_command(output_dir: str, project: str, variant: str | None, compress: bool) -> None: - """Export handmade (custom) assessments as OpenVEX file(s).""" - - author = get_default_author() - now_iso = _dt.now(_tz.utc).isoformat() - +def export_custom_vulnscout_data_command(output_dir: str, project: str, variant: str | None) -> None: + """Export custom VulnScout JSON data for one or all project variants.""" project_obj = resolve_project(project) - - variants: list[DBVariant] if variant: _, variant_obj = resolve_project_variant(project, variant, create=False) - variants = [variant_obj] + variant_ids = [variant_obj.id] + filename_label = sanitize_variant_name(variant_obj.name) else: - variants = DBVariant.get_by_project(project_obj.id) - - handmade = DBAssessment.get_by_origin([v.id for v in variants]) - if not handmade: - click.echo("No custom assessments to export.", err=True) - raise SystemExit(1) - + variant_ids = [project_variant.id for project_variant in DBVariant.get_by_project(project_obj.id)] + filename_label = "all" + + data = build_custom_data_export(variant_ids) + if ( + not data["assessments"] + and not data["ai_assessments"] + and not data["cvss"] + and not data["time_estimates"] + ): + raise click.ClickException("No custom VulnScout data to export.") os.makedirs(output_dir, exist_ok=True) - vuln_cache: dict = {} - - def _write_doc(assessments, filename): - doc = build_openvex_doc(assessments, author, now_iso, vuln_cache) - out_path = os.path.join(output_dir, filename) - with open(out_path, "w") as fh: - _json.dump(doc, fh, indent=2) - return out_path - - def _write_tar(docs: dict[str, list]): - buf = io.BytesIO() - with tarfile.open(fileobj=buf, mode='w:gz') as tar: - for filename, assessments in docs.items(): - doc = build_openvex_doc(assessments, author, now_iso, vuln_cache) - json_bytes = _json.dumps(doc, indent=2).encode("utf-8") - info = tarfile.TarInfo(name=filename) - info.size = len(json_bytes) - tar.addfile(info, io.BytesIO(json_bytes)) - out_path = os.path.join(output_dir, "custom_assessments.tar.gz") - with open(out_path, "wb") as fh: - fh.write(buf.getvalue()) - return out_path - - if variant is None: - by_variant: dict[uuid.UUID, list[DBAssessment]] = defaultdict(list) - for assess in handmade: - if assess.variant_id is not None: - by_variant[assess.variant_id].append(assess) - - named_groups: dict[str, list] = {} - for vid, assessments in by_variant.items(): - vobj = DBVariant.get_by_id(vid) - assert vobj - filename = sanitize_variant_name(vobj.name) + ".json" - named_groups[filename] = assessments - - if compress: - out_path = _write_tar(named_groups) - click.echo(f"Custom assessments exported: {out_path}") - else: - for filename, assessments in named_groups.items(): - out_path = _write_doc(assessments, filename) - click.echo(f"Custom assessments exported: {out_path}") - else: - assert variant_obj - filename = sanitize_variant_name(variant_obj.name) + ".json" - - if compress: - out_path = _write_tar({filename: handmade}) - else: - out_path = _write_doc(handmade, filename) - click.echo(f"Custom assessments exported: {out_path}") + output_path = os.path.join(output_dir, f"custom_vulnscout_data_{filename_label}{_JSON_SUFFIX}") + with open(output_path, "w") as file: + _json.dump(data, file, indent=2) + click.echo(f"Custom VulnScout data exported: {output_path}") -@click.command("import-custom-assessments") +@click.command("import-custom-vulnscout-data") @click.argument("file_path") @click.option("--project", "-p", required=True, help="Project name.") -@click.option("--variant", "-v", default=None, help="Variant name. Defaults to the file name.") @with_appcontext -def import_custom_assessments_command(file_path: str, project: str, variant: str | None) -> None: - """Import custom assessments from a .json, .tar.gz, or directory of OpenVEX files.""" - if not os.path.isfile(file_path) and not os.path.isdir(file_path): - click.echo(f"Error: file not found: {file_path}", err=True) - raise SystemExit(1) - +def import_custom_vulnscout_data_command(file_path: str, project: str) -> None: + """Import custom VulnScout JSON data using its embedded variant metadata.""" project_obj = resolve_project(project) - - variant_obj: DBVariant | None = None - if variant: - _, variant_obj = resolve_project_variant(project, variant, create=False) + data = _load_json_file(file_path) + if "version" not in data or is_openvex_doc(data): + raise click.ClickException("Not a valid VulnScout JSON data file.") variant_by_name = build_variant_by_name_map(project_obj.id) - basename = os.path.basename(file_path) - total_created: list[dict] - total_errors: list[dict] - total_skipped: int - variant_files_found: int = 0 - - if file_path.endswith(".tar.gz") or file_path.endswith(".tgz"): - if variant: - click.echo("Error: cannot use the --variant argument with an archive of custom assessments.") - raise SystemExit(1) - - try: - with open(file_path, "rb") as fh: - file_bytes = fh.read() - total_created, total_errors, total_skipped, variant_files_found = ( - import_archive_bytes(file_bytes, variant_by_name) - ) - except ValueError as exc: - click.echo(f"Error: {exc}", err=True) - raise SystemExit(1) - - if variant_files_found == 0 and not total_created: - click.echo( - "Error: no valid OpenVEX files matching known " - "variants found in archive.", err=True - ) - for err in total_errors: - click.echo(f" {err}", err=True) - raise SystemExit(1) - - elif os.path.isdir(file_path): - if variant: - click.echo("Error: cannot use the --variant argument with a directory of custom assessments.") - raise SystemExit(1) - - try: - total_created, total_errors, total_skipped, variant_files_found = ( - import_directory(file_path, variant_by_name) - ) - except ValueError as exc: - click.echo(f"Error: {exc}", err=True) - raise SystemExit(1) - - if variant_files_found == 0 and not total_created: - click.echo( - "Error: no valid OpenVEX files matching known " - "variants found in directory.", err=True - ) - for err in total_errors: - click.echo(f" {err}", err=True) - raise SystemExit(1) - - elif file_path.endswith(".json"): - try: - with open(file_path) as fh: - data = _json.load(fh) - except Exception: - click.echo("Error: invalid JSON file.", err=True) - raise SystemExit(1) - - # Custom-data export format (web "export custom data" button): - # {version, assessments, cvss, time_estimates}. Routed to the - # dedicated importer; the embedded per-item variant is used unless - # --variant forces a target. - if isinstance(data, dict) and "version" in data and not is_openvex_doc(data): - result = import_custom_data( - data, variant_by_name, - variant_obj.id if variant_obj is not None else None, - ) - if result.get("status") != "success": - click.echo("Error: failed to import custom-data file.", err=True) - for err in result.get("errors", []): - click.echo(f" {err}", err=True) - raise SystemExit(1) - for err in result.get("errors", []): - click.echo(f" Warning: {err}", err=True) - click.echo( - f"Imported {result['assessments_imported']} assessments" - f" ({result['assessments_skipped']} skipped as duplicates)," - f" {result['cvss_imported']} CVSS," - f" {result['time_estimates_imported']} time estimates" - ) - return - - # OpenVEX single-document format: requires a variant (from --variant - # or the filename matching an existing variant name). - if variant: - assert variant_obj - else: - variant_name = basename[:-len(".json")] - variant_obj = variant_by_name.get(variant_name) - if variant_obj is None: - click.echo( - f"Error: no variant found matching filename " - f"'{variant_name}'. The JSON filename must " - f"correspond to an existing variant name. " - "Hint: use --variant to specify another name.", - err=True, - ) - raise SystemExit(1) - - if not is_openvex_doc(data): - click.echo("Error: not a valid OpenVEX document.", err=True) - raise SystemExit(1) - - total_created, total_errors, total_skipped = import_statements( - data["statements"], variant_obj.id - ) - else: - click.echo( - "Error: unsupported file type. " - "Please provide a .json, .tar.gz file, or directory.", - err=True, - ) - raise SystemExit(1) + _print_custom_data_import_result(import_custom_data(data, variant_by_name)) + + +@click.command("export-custom-openvex-assessments") +@click.option("--output-dir", default="/scan/outputs", show_default=True, + help="Directory where the exported file is written.") +@click.option("--project", "-p", required=True, help="Project name.") +@click.option("--variant", "-v", required=True, help="Variant name to export.") +@with_appcontext +def export_custom_openvex_assessments_command(output_dir: str, project: str, variant: str) -> None: + """Export custom assessments for one variant as an OpenVEX JSON file.""" + _, variant_obj = resolve_project_variant(project, variant, create=False) + handmade = DBAssessment.get_by_origin([variant_obj.id]) + if not handmade: + raise click.ClickException("No custom assessments to export.") + + author = get_default_author() + document = build_openvex_doc(handmade, author, _dt.now(_tz.utc).isoformat()) + os.makedirs(output_dir, exist_ok=True) + output_path = os.path.join( + output_dir, + f"custom_openvex_{sanitize_variant_name(variant_obj.name)}{_JSON_SUFFIX}", + ) + with open(output_path, "w") as file: + _json.dump(document, file, indent=2) + click.echo(f"Custom OpenVEX assessments exported: {output_path}") - for err in total_errors: - click.echo(f" Warning: {err}", err=True) +@click.command("import-custom-openvex-assessments") +@click.argument("file_path") +@click.option("--project", "-p", required=True, help="Project name.") +@click.option("--variant", "-v", required=True, help="Variant name to import into.") +@with_appcontext +def import_custom_openvex_assessments_command(file_path: str, project: str, variant: str) -> None: + """Import one OpenVEX JSON document into the specified variant.""" + _, variant_obj = resolve_project_variant(project, variant, create=False) + data = _load_json_file(file_path) + if not is_openvex_doc(data): + raise click.ClickException("Not a valid OpenVEX document.") + + total_created, total_errors, total_skipped = import_statements( + data["statements"], variant_obj.id + ) + for error in total_errors: + click.echo(f" Warning: {error}", err=True) click.echo( - f"Imported {len(total_created)} assessments" + f"Imported {len(total_created)} OpenVEX assessments" f" ({total_skipped} skipped as duplicates)" ) diff --git a/src/bin/merger_ci.py b/src/bin/merger_ci.py index 0a563637d..1e886758a 100644 --- a/src/bin/merger_ci.py +++ b/src/bin/merger_ci.py @@ -11,7 +11,7 @@ # This module is the entry point / orchestrator. The actual logic lives in: # cmd_process.py — ``flask merge`` and ``flask process`` # cmd_export.py — ``flask export`` and ``flask report`` -# cmd_assessments.py — ``flask export-custom-assessments`` / ``flask import-custom-assessments`` +# cmd_assessments.py — custom VulnScout JSON and OpenVEX import/export commands # cmd_scans.py — ``flask list-projects``, ``flask list-scans``, ``flask delete-scan`` # cmd_vuln_scan.py — ``flask nvd-scan`` and ``flask osv-scan`` @@ -29,8 +29,10 @@ ) from .cmd_export import export_command, report_command from .cmd_assessments import ( - export_custom_assessments_command, - import_custom_assessments_command, + export_custom_openvex_assessments_command, + export_custom_vulnscout_data_command, + import_custom_openvex_assessments_command, + import_custom_vulnscout_data_command, ) from .cmd_scans import ( list_projects_command, @@ -46,8 +48,10 @@ def init_app(app) -> None: app.cli.add_command(process_command) app.cli.add_command(report_command) app.cli.add_command(export_command) - app.cli.add_command(export_custom_assessments_command) - app.cli.add_command(import_custom_assessments_command) + app.cli.add_command(export_custom_vulnscout_data_command) + app.cli.add_command(import_custom_vulnscout_data_command) + app.cli.add_command(export_custom_openvex_assessments_command) + app.cli.add_command(import_custom_openvex_assessments_command) app.cli.add_command(list_projects_command) app.cli.add_command(list_scans_command) app.cli.add_command(delete_scan_command) diff --git a/src/entrypoint.sh b/src/entrypoint.sh index f63f1a731..017f90c79 100755 --- a/src/entrypoint.sh +++ b/src/entrypoint.sh @@ -120,9 +120,10 @@ Scan & output commands: --export-spdx Export project as SPDX 3.0 SBOM to /scan/outputs/ --export-cdx Export project as CycloneDX 1.6 SBOM to /scan/outputs/ --export-openvex Export project as OpenVEX document to /scan/outputs/ - --export-custom-assessments Export custom (review) assessments as individual OpenVEX files to /scan/outputs/ - --compress Compress export output into a .tar.gz archive - --import-custom-assessments Import custom assessments from .json, .tar.gz, or directory + --export-custom-vulnscout-data Export custom VulnScout JSON data to /scan/outputs/ + --import-custom-vulnscout-data Import custom VulnScout JSON data + --export-custom-openvex-assessments Export custom OpenVEX assessments for --variant + --import-custom-openvex-assessments Import custom OpenVEX assessments into --variant --match-condition Exit code 2 if condition met (e.g. "cvss >= 9.0") --delete-scan Delete a past scan by its ID @@ -527,31 +528,25 @@ cmd_export() { setup_user } -cmd_export_custom_assessments() { +cmd_export_custom_vulnscout_data() { export_args=(--project "$PROJECT_NAME") if [[ -n "$VARIANT_NAME" ]]; then export_args+=(--variant "$VARIANT_NAME") fi - if [[ "${COMPRESS:-false}" == "true" ]]; then - export_args+=(--compress) - fi cd "$BASE_DIR" local output_dir="${OUTPUTS_DIR:-/scan/outputs}" export_args+=(--output-dir "$output_dir") flask --app src.bin.webapp db upgrade - flask --app src.bin.webapp export-custom-assessments "${export_args[@]}" + flask --app src.bin.webapp export-custom-vulnscout-data "${export_args[@]}" setup_user } -cmd_import_custom_assessments() { +cmd_import_custom_vulnscout_data() { local file="$1" import_args=(--project "$PROJECT_NAME") - if [[ -n "$VARIANT_NAME" ]]; then - import_args+=(--variant "$VARIANT_NAME") - fi cd "$BASE_DIR" local raw_basename dest_name dest_file @@ -566,7 +561,45 @@ cmd_import_custom_assessments() { import_args+=("$file") flask --app src.bin.webapp db upgrade - flask --app src.bin.webapp import-custom-assessments "${import_args[@]}" + flask --app src.bin.webapp import-custom-vulnscout-data "${import_args[@]}" + setup_user +} + +cmd_export_custom_openvex_assessments() { + if [[ -z "$VARIANT_NAME" ]]; then + echo "Error: --variant is required to export custom OpenVEX assessments." >&2 + exit 1 + fi + + cd "$BASE_DIR" + local output_dir="${OUTPUTS_DIR:-/scan/outputs}" + flask --app src.bin.webapp db upgrade + flask --app src.bin.webapp export-custom-openvex-assessments \ + --project "$PROJECT_NAME" --variant "$VARIANT_NAME" --output-dir "$output_dir" + setup_user +} + +cmd_import_custom_openvex_assessments() { + local file="$1" + if [[ -z "$VARIANT_NAME" ]]; then + echo "Error: --variant is required to import custom OpenVEX assessments." >&2 + exit 1 + fi + + cd "$BASE_DIR" + local raw_basename dest_name dest_file + raw_basename="$(basename "$file")" + dest_name="${raw_basename#vulnscout_stage_}" + if [[ "$dest_name" != "$raw_basename" ]]; then + # Strip the staging prefix added by the wrapper before importing. + dest_file="$(dirname "$file")/$dest_name" + mv "$file" "$dest_file" + file="$dest_file" + fi + + flask --app src.bin.webapp db upgrade + flask --app src.bin.webapp import-custom-openvex-assessments \ + --project "$PROJECT_NAME" --variant "$VARIANT_NAME" "$file" setup_user } @@ -751,12 +784,14 @@ while [[ $# -gt 0 ]]; do EXPORT_FORMATS+=("cdx16"); shift ;; --export-openvex) EXPORT_FORMATS+=("openvex"); shift ;; - --export-custom-assessments) - EXPORT_CUSTOM_ASSESSMENTS=true; shift ;; - --compress) - COMPRESS=true; shift ;; - --import-custom-assessments) - IMPORT_CUSTOM_ASSESSMENTS_FILE="$2"; shift 2 ;; + --export-custom-vulnscout-data) + EXPORT_CUSTOM_VULNSCOUT_DATA=true; shift ;; + --import-custom-vulnscout-data) + IMPORT_CUSTOM_VULNSCOUT_DATA_FILE="$2"; shift 2 ;; + --export-custom-openvex-assessments) + EXPORT_CUSTOM_OPENVEX_ASSESSMENTS=true; shift ;; + --import-custom-openvex-assessments) + IMPORT_CUSTOM_OPENVEX_ASSESSMENTS_FILE="$2"; shift 2 ;; --list-projects|--list-scans) cmd_get_data "$1"; shift ;; --config) @@ -793,12 +828,18 @@ for _fmt in "${EXPORT_FORMATS[@]:-}"; do [[ -n "$_fmt" ]] && cmd_export "$_fmt" done -# Step 4: Export/import custom assessments -if [[ "${EXPORT_CUSTOM_ASSESSMENTS:-false}" == "true" ]]; then - cmd_export_custom_assessments +# Step 4: Export/import custom assessment data +if [[ "${EXPORT_CUSTOM_VULNSCOUT_DATA:-false}" == "true" ]]; then + cmd_export_custom_vulnscout_data +fi +if [[ -n "${IMPORT_CUSTOM_VULNSCOUT_DATA_FILE:-}" ]]; then + cmd_import_custom_vulnscout_data "$IMPORT_CUSTOM_VULNSCOUT_DATA_FILE" +fi +if [[ "${EXPORT_CUSTOM_OPENVEX_ASSESSMENTS:-false}" == "true" ]]; then + cmd_export_custom_openvex_assessments fi -if [[ -n "${IMPORT_CUSTOM_ASSESSMENTS_FILE:-}" ]]; then - cmd_import_custom_assessments "$IMPORT_CUSTOM_ASSESSMENTS_FILE" +if [[ -n "${IMPORT_CUSTOM_OPENVEX_ASSESSMENTS_FILE:-}" ]]; then + cmd_import_custom_openvex_assessments "$IMPORT_CUSTOM_OPENVEX_ASSESSMENTS_FILE" fi # Step 5: Get data diff --git a/src/helpers/assessment_io.py b/src/helpers/assessment_io.py index debe2054a..d2abfe259 100644 --- a/src/helpers/assessment_io.py +++ b/src/helpers/assessment_io.py @@ -1,7 +1,7 @@ # Copyright (C) 2026 Savoir-faire Linux, Inc. # SPDX-License-Identifier: GPL-3.0-only -"""Shared helpers for importing and exporting assessments as OpenVEX archives. +"""Shared helpers for importing and exporting assessments as OpenVEX JSON. Both the CLI (``cmd_assessments.py``) and the web API (``routes/assessments.py``) perform the same build/parse logic. This module contains the common core so @@ -10,12 +10,7 @@ from __future__ import annotations -import io -import json -import os -import tarfile import uuid as _uuid -from collections import defaultdict from datetime import datetime as _dt, timezone as _tz from typing import TYPE_CHECKING, Any @@ -157,62 +152,6 @@ def build_openvex_doc( } -def build_openvex_archive( - handmade_assessments: "list[_Assessment]", - variant_names: dict[str, str], - author: str, - now_iso: str | None = None, -) -> bytes: - """Build an in-memory tar.gz archive of OpenVEX JSON files. - - One ``.json`` file is created per variant (named - ``.json``). Assessments without a variant go into - ``unassigned.json``. - - Parameters - ---------- - handmade_assessments: - List of DB ``Assessment`` objects (usually from - ``Assessment.get_by_origin()``). - variant_names: - Mapping ``str(variant_id) → variant_name`` used to name the files. - author: - Author string written into every OpenVEX document header. - now_iso: - ISO-8601 timestamp written into every document. Defaults to *now*. - - Returns - ------- - bytes - Raw tar.gz content. - """ - if now_iso is None: - now_iso = _dt.now(_tz.utc).isoformat() - - vuln_cache: "dict[str, _Vulnerability | None]" = {} - - by_variant: "dict[str | None, list[_Assessment]]" = defaultdict(list) - for assess in handmade_assessments: - vid = str(assess.variant_id) if assess.variant_id else None - by_variant[vid].append(assess) - - buf = io.BytesIO() - with tarfile.open(fileobj=buf, mode='w:gz') as tar: - for vid, assessments in by_variant.items(): - filename = sanitize_variant_name( - variant_names.get(vid, "unassigned") if vid else "unassigned" - ) + ".json" - - doc = build_openvex_doc(assessments, author, now_iso, vuln_cache) - - json_bytes = json.dumps(doc, indent=2).encode("utf-8") - info = tarfile.TarInfo(name=filename) - info.size = len(json_bytes) - tar.addfile(info, io.BytesIO(json_bytes)) - - return buf.getvalue() - - # --------------------------------------------------------------------------- # Import helpers # --------------------------------------------------------------------------- @@ -376,121 +315,6 @@ def build_variant_by_name_map(project_id: "_uuid.UUID | None" = None) -> "dict[s return variant_by_name -def import_archive_bytes( - file_bytes: bytes, - variant_by_name: "dict[str, _Variant]", -) -> tuple[list[dict[str, Any]], list[dict[str, Any]], int, int]: - """Import OpenVEX assessments from a tar.gz archive (as raw bytes). - - Returns - ------- - (created, errors, skipped, variant_files_found) - """ - total_created: list[dict[str, Any]] = [] - total_errors: list[dict[str, Any]] = [] - total_skipped = 0 - variant_files_found = 0 - - try: - tar = tarfile.open(fileobj=io.BytesIO(file_bytes), mode='r:gz') - except Exception: - raise ValueError("Unable to open tar.gz archive") - - for member in tar.getmembers(): - if not member.isfile() or not member.name.endswith(".json"): - continue - base = os.path.basename(member.name) - variant_name = base[: -len(".json")] - variant = variant_by_name.get(variant_name) - if variant is None: - total_errors.append({ - "file": member.name, - "error": f"No variant found matching name '{variant_name}'", - }) - continue - - f = tar.extractfile(member) - if f is None: - continue - try: - doc = json.load(f) - except Exception: - total_errors.append({"file": member.name, "error": "Invalid JSON"}) - continue - - if not is_openvex_doc(doc): - total_errors.append({ - "file": member.name, - "error": "Not a valid OpenVEX document", - }) - continue - - variant_files_found += 1 - c, e, s = import_statements(doc["statements"], variant.id) - total_created.extend(c) - total_errors.extend(e) - total_skipped += s - - tar.close() - return total_created, total_errors, total_skipped, variant_files_found - - -def import_directory( - dir_path: str, - variant_by_name: "dict[str, _Variant]", -) -> tuple[list[dict[str, Any]], list[dict[str, Any]], int, int]: - """Import OpenVEX assessments from a directory of JSON files. - - Each ``.json`` file is matched to a variant by its filename (sans - extension). - - Returns - ------- - (created, errors, skipped, variant_files_found) - """ - total_created: list[dict[str, Any]] = [] - total_errors: list[dict[str, Any]] = [] - total_skipped = 0 - variant_files_found = 0 - - json_files = sorted(f for f in os.listdir(dir_path) if f.endswith(".json")) - if not json_files: - raise ValueError("No .json files found in directory") - - for json_name in json_files: - variant_name = json_name[: -len(".json")] - variant = variant_by_name.get(variant_name) - if variant is None: - total_errors.append({ - "file": json_name, - "error": f"No variant found matching name '{variant_name}'", - }) - continue - - json_path = os.path.join(dir_path, json_name) - try: - with open(json_path) as fh: - doc = json.load(fh) - except Exception: - total_errors.append({"file": json_name, "error": "Invalid JSON"}) - continue - - if not is_openvex_doc(doc): - total_errors.append({ - "file": json_name, - "error": "Not a valid OpenVEX document", - }) - continue - - variant_files_found += 1 - c, e, s = import_statements(doc["statements"], variant.id) - total_created.extend(c) - total_errors.extend(e) - total_skipped += s - - return total_created, total_errors, total_skipped, variant_files_found - - # --------------------------------------------------------------------------- # Custom-data export (assessments + CVSS + time estimates) # --------------------------------------------------------------------------- @@ -509,8 +333,8 @@ def build_custom_data_export( Returns ------- - dict with keys ``version``, ``exported_at``, ``assessments``, ``cvss``, - ``time_estimates``. + dict with keys ``version``, ``exported_at``, ``assessments``, + ``ai_assessments``, ``cvss``, ``time_estimates``. """ from ..extensions import db from ..models.assessment import Assessment as DBAssessment @@ -519,27 +343,36 @@ def build_custom_data_export( from ..models.iso8601_duration import Iso8601Duration from ..models.variant import Variant as DBVariant - handmade = DBAssessment.get_by_origin(variant_ids) + handmade = DBAssessment.get_by_origin(variant_ids, origin="custom") + pending_ai = DBAssessment.get_by_origin(variant_ids, origin="ai") variant_name_by_id: dict[str, str] = {} variant_uuid_set: set[_uuid.UUID] = { - a.variant_id for a in handmade if a.variant_id is not None + a.variant_id for a in [*handmade, *pending_ai] + if a.variant_id is not None } - exported_assessments = [] - for a in handmade: - d = a.to_dict() - exported_assessments.append({ - "vuln_id": d["vuln_id"], - "status": d["status"], - "simplified_status": d.get("simplified_status", ""), - "justification": d.get("justification") or None, - "impact_statement": d.get("impact_statement") or None, - "status_notes": d.get("status_notes") or None, - "workaround": d.get("workaround") or None, - "packages": d["packages"], - "variant_id": d.get("variant_id"), - }) + def _export_assessments( + assessments: "list[DBAssessment]", + ) -> list[dict[str, Any]]: + exported = [] + for assessment in assessments: + assessment_dict = assessment.to_dict() + exported.append({ + "vuln_id": assessment_dict["vuln_id"], + "status": assessment_dict["status"], + "simplified_status": assessment_dict.get("simplified_status", ""), + "justification": assessment_dict.get("justification") or None, + "impact_statement": assessment_dict.get("impact_statement") or None, + "status_notes": assessment_dict.get("status_notes") or None, + "workaround": assessment_dict.get("workaround") or None, + "packages": assessment_dict["packages"], + "variant_id": assessment_dict.get("variant_id"), + }) + return exported + + exported_assessments = _export_assessments(handmade) + exported_ai_assessments = _export_assessments(pending_ai) # Gather ALL custom CVSS entries, not just those linked to handmade # assessments. @@ -628,6 +461,10 @@ def _hours_to_iso(h: int) -> str: vid = item.get("variant_id") item["variant"] = variant_name_by_id.get(vid) if vid else None + for item in exported_ai_assessments: + vid = item.get("variant_id") + item["variant"] = variant_name_by_id.get(vid) if vid else None + for item in cvss_entries: vid = item.get("variant_id") item["variant"] = variant_name_by_id.get(vid) if vid else None @@ -640,6 +477,7 @@ def _hours_to_iso(h: int) -> str: "version": 1, "exported_at": _dt.now(_tz.utc).isoformat(), "assessments": exported_assessments, + "ai_assessments": exported_ai_assessments, "cvss": cvss_entries, "time_estimates": time_estimates, } @@ -661,7 +499,7 @@ def import_custom_data( ---------- data: Parsed JSON matching the custom-data export format - (``{version, assessments, cvss, time_estimates}``). + (``{version, assessments, ai_assessments, cvss, time_estimates}``). variant_by_name: Mapping ``{name: Variant, sanitised_name: Variant}`` for variant resolution. @@ -671,7 +509,7 @@ def import_custom_data( Returns ------- - dict with ``status``, ``assessments_imported``, ``assessments_skipped``, + dict with ``status``, assessment import counts for custom and AI rows, ``cvss_imported``, ``time_estimates_imported``, ``errors``. """ from ..extensions import db @@ -691,6 +529,8 @@ def import_custom_data( "status": "success", "assessments_imported": 0, "assessments_skipped": 0, + "ai_assessments_imported": 0, + "ai_assessments_skipped": 0, "cvss_imported": 0, "time_estimates_imported": 0, "errors": [], @@ -725,10 +565,16 @@ def _resolve_variant(raw_item: dict) -> "_uuid.UUID | None": return None return mapped_variant.id - # -- Import assessments -- - assessments_list = data.get("assessments", []) - if isinstance(assessments_list, list): - for a in assessments_list: + def _import_assessments( + key: str, + origin: str, + imported_key: str, + skipped_key: str, + ) -> None: + assessment_list = data.get(key, []) + if not isinstance(assessment_list, list): + return + for a in assessment_list: if not isinstance(a, dict): continue vuln_name = a.get("vuln_id") @@ -774,11 +620,11 @@ def _resolve_variant(raw_item: dict) -> "_uuid.UUID | None": DBAssessment.finding_id == finding.id, DBAssessment.variant_id == target_variant_id, DBAssessment.status == status, - DBAssessment.origin == "custom", + DBAssessment.origin == origin, ) ).scalars().first() if existing is not None: - result["assessments_skipped"] += 1 + result[skipped_key] += 1 continue DBAssessment.create( @@ -788,7 +634,7 @@ def _resolve_variant(raw_item: dict) -> "_uuid.UUID | None": ), finding_id=finding.id, variant_id=target_variant_id, - origin="custom", + origin=origin, status_notes=status_notes, justification=justification, impact_statement=impact_statement, @@ -796,7 +642,7 @@ def _resolve_variant(raw_item: dict) -> "_uuid.UUID | None": responses=[], commit=True, ) - result["assessments_imported"] += 1 + result[imported_key] += 1 except Exception as e: result["errors"].append({ "vuln_id": vuln_name, @@ -804,6 +650,15 @@ def _resolve_variant(raw_item: dict) -> "_uuid.UUID | None": "error": str(e), }) + # Import pending AI assessments separately so the Review page continues to + # surface them in its AI Assessments tab for approval or rejection. + _import_assessments( + "assessments", "custom", "assessments_imported", "assessments_skipped" + ) + _import_assessments( + "ai_assessments", "ai", "ai_assessments_imported", "ai_assessments_skipped" + ) + # -- Import CVSS -- cvss_list = data.get("cvss", []) if isinstance(cvss_list, list): @@ -914,7 +769,12 @@ def _resolve_variant(raw_item: dict) -> "_uuid.UUID | None": apply_effort(record, target_variant_id, effort, log_prefix="import-custom-data") result["time_estimates_imported"] += 1 - if not result["assessments_imported"] and not result["cvss_imported"] and not result["time_estimates_imported"]: + if ( + not result["assessments_imported"] + and not result["ai_assessments_imported"] + and not result["cvss_imported"] + and not result["time_estimates_imported"] + ): if result["errors"]: result["status"] = "error" diff --git a/src/routes/assessments.py b/src/routes/assessments.py index b46e98e99..6004113da 100644 --- a/src/routes/assessments.py +++ b/src/routes/assessments.py @@ -14,11 +14,10 @@ from ._scan_queries import VulnerabilityText, fetch_vulnerabilities_texts from ._scan_diff import invalidate_scan_list_cache from ..helpers.assessment_io import ( - build_openvex_archive, + build_openvex_doc, is_openvex_doc, import_statements as _import_openvex_statements, build_variant_by_name_map, - import_archive_bytes, build_custom_data_export, import_custom_data, ) @@ -234,114 +233,72 @@ def review_ai_assessments() -> ResponseReturnValue: @app.route('/api/assessments/review/export') def export_review_openvex() -> ResponseReturnValue: - """Export handmade (review) assessments as a .tar.gz containing one - OpenVEX JSON file per variant (``.json``). - Assessments without a variant are placed in ``unassigned.json``. - """ + """Export review assessments for one variant as an OpenVEX JSON document.""" from ..models.variant import Variant as DBVariant - handmade = DBAssessment.get_by_origin() + raw_variant_ids = request.args.getlist('variant_id') + if len(raw_variant_ids) != 1: + return {"error": "Exactly one variant_id is required for OpenVEX export"}, 400 + variant_uuid, err = parse_uuid_or_400(raw_variant_ids[0], "variant_id") + if err: + return err + if variant_uuid is None: + return {"error": "Internal error"}, 500 + variant = DBVariant.get_by_id(variant_uuid) + if variant is None: + return {"error": "Variant not found"}, 404 + + handmade = DBAssessment.get_by_origin([variant_uuid], origin="custom") if not handmade: return {"error": "No review assessments to export"}, 404 author = request.args.get('author', 'Savoir-faire Linux') - variant_names = {str(v.id): v.name for v in DBVariant.get_all()} - - archive_bytes = build_openvex_archive(handmade, variant_names, author) - return archive_bytes, 200, { - "Content-Type": "application/gzip", - "Content-Disposition": "attachment; filename=review_openvex.tar.gz", + import json + json_data = json.dumps(build_openvex_doc(handmade, author), indent=2) + filename = re.sub(r"[^\w\-.]", "_", variant.name) + return json_data, 200, { + "Content-Type": "application/json", + "Content-Disposition": f'attachment; filename="review_openvex_{filename}.json"', } @app.route('/api/assessments/review/import', methods=['POST']) def import_review_openvex() -> ResponseReturnValue: - """Import OpenVEX review assessments from a ``.json`` or ``.tar.gz`` file. - - * **Single .json file** – the filename (without extension) must match - an existing variant name in the database. - * **.tar.gz archive** – each ``.json`` entry inside must be named after - an existing variant. Entries whose basename does not match a known - variant are reported as errors. - - Every file is validated as a well-formed OpenVEX document (must contain - ``@context`` with ``openvex`` and a ``statements`` array). + """Import a JSON OpenVEX document into exactly one selected variant.""" - Assessments are created with ``origin="custom"``. - """ - import os - - # ---- retrieve the uploaded file ---- if not (request.content_type and 'multipart/form-data' in request.content_type): return {"error": "Expected multipart/form-data with a file upload"}, 400 uploaded = request.files.get('file') if not uploaded or not uploaded.filename: return {"error": "No file uploaded"}, 400 + if not uploaded.filename.endswith(".json"): + return {"error": "Unsupported file type. Please upload a .json file."}, 400 - filename = uploaded.filename - - # ---- build variant-name → variant lookup ---- - all_variants = DBVariant.get_all() - # The export sanitises names (/ and \ replaced by _), so we store a - # sanitised-name → variant mapping for reliable round-trip matching. - variant_by_name: dict[str, "DBVariant"] = {} - for v in all_variants: - sanitised = v.name.replace("/", "_").replace("\\", "_") - variant_by_name[sanitised] = v - # Also keep the original name in case it differs - variant_by_name[v.name] = v - - variant_by_name = build_variant_by_name_map() - - # ---- .tar.gz handling ---- - if filename.endswith(".tar.gz") or filename.endswith(".tgz"): - try: - total_created, total_errors, total_skipped, found = import_archive_bytes( - uploaded.read(), variant_by_name - ) - except ValueError as exc: - return {"error": str(exc)}, 400 + raw_variant_id = request.form.get('variant_id') + if not raw_variant_id: + return {"error": "variant_id is required for OpenVEX import"}, 400 + target_variant_id, err = parse_uuid_or_400(raw_variant_id, "variant_id") + if err: + return err + if target_variant_id is None: + return {"error": "Internal error"}, 500 + if DBVariant.get_by_id(target_variant_id) is None: + return {"error": "Variant not found"}, 404 - if found == 0 and not total_created: - return { - "error": "No valid OpenVEX files matching known variants found in archive", - "errors": total_errors, - }, 400 + import json + try: + data = json.load(uploaded.stream) + except Exception: + return {"error": "Invalid JSON file"}, 400 + if not is_openvex_doc(data): return { - "status": "success", - "imported": len(total_created), - "skipped": total_skipped, - "errors": total_errors, - }, 200 - - # ---- single .json handling ---- - if filename.endswith(".json"): - import json - base = os.path.basename(filename) - variant_name = base[: -len(".json")] - variant = variant_by_name.get(variant_name) - if variant is None: - return { - "error": f"No variant found matching filename '{variant_name}'. " - f"The JSON filename must correspond to an existing variant name." - }, 400 - - try: - data = json.load(uploaded.stream) - except Exception: - return {"error": "Invalid JSON file"}, 400 - - if not is_openvex_doc(data): - return { - "error": "Not a valid OpenVEX document " - "(missing @context with 'openvex' " - "or 'statements' array)" - }, 400 - - created, errors, skipped = _import_openvex_statements(data["statements"], variant.id) - return {"status": "success", "imported": len(created), "skipped": skipped, "errors": errors}, 200 + "error": "Not a valid OpenVEX document " + "(missing @context with 'openvex' " + "or 'statements' array)" + }, 400 - return {"error": "Unsupported file type. Please upload a .json or .tar.gz file."}, 400 + created, errors, skipped = _import_openvex_statements(data["statements"], target_variant_id) + return {"status": "success", "imported": len(created), "skipped": skipped, "errors": errors}, 200 @app.route('/api/assessments/review/time-estimates') def review_time_estimates() -> ResponseReturnValue: @@ -501,62 +458,67 @@ def review_custom_cvss() -> ResponseReturnValue: @app.route('/api/assessments/review/export-custom-data') def export_review_custom_data() -> ResponseReturnValue: - """Export handmade (review) assessments, custom CVSS scores and time - estimates as a single JSON file. + """Export handmade and pending AI assessments, custom CVSS scores and + time estimates as a single JSON file. Query parameters: - * ``variant_id`` – restrict to a single variant. - * ``project_id`` – restrict to all variants in a project. + * ``variant_id`` - restrict to selected variants; may be repeated. + * ``project_id`` - restrict to all variants in a project. """ - variant_id = request.args.get('variant_id') + raw_variant_ids = request.args.getlist('variant_id') project_id = request.args.get('project_id') from ..models.project import Project as DBProject variant_ids: list[UUID] | None = None project_name = None - if variant_id: - vid, err = parse_uuid_or_400(variant_id, "variant_id") - if err: - return err - if vid is None: - return {"error": "Internal error"}, 500 - variant_ids = [vid] - variant = DBVariant.get_by_id(vid) + if raw_variant_ids: + variant_ids = [] + for raw_variant_id in raw_variant_ids: + variant_id, err = parse_uuid_or_400(raw_variant_id, "variant_id") + if err: + return err + if variant_id is None: + return {"error": "Internal error"}, 500 + variant_ids.append(variant_id) + variant = DBVariant.get_by_id(variant_ids[0]) if variant and variant.project: project_name = variant.project.name elif project_id: - pid, err = parse_uuid_or_400(project_id, "project_id") + project_uuid, err = parse_uuid_or_400(project_id, "project_id") if err: return err - if pid is None: + if project_uuid is None: return {"error": "Internal error"}, 500 - project = DBProject.get_by_id(pid) + project = DBProject.get_by_id(project_uuid) if project: project_name = project.name - variants = DBVariant.get_by_project(pid) - variant_ids = [v.id for v in variants] + variant_ids = [variant.id for variant in DBVariant.get_by_project(project_uuid)] data = build_custom_data_export(variant_ids) - if not data["assessments"] and not data["cvss"] and not data["time_estimates"]: + if ( + not data["assessments"] + and not data["ai_assessments"] + and not data["cvss"] + and not data["time_estimates"] + ): return {"error": "No custom data to export"}, 404 import json as _json - import re as _re json_bytes = _json.dumps(data, indent=2) - safe_name = _re.sub(r'[^\w\-.]', '_', project_name) if project_name else None - fname = f"custom_data_{safe_name}.json" if safe_name else "custom_data.json" + safe_name = re.sub(r'[^\w\-.]', '_', project_name) if project_name else None + filename = f"custom_data_{safe_name}.json" if safe_name else "custom_data.json" return json_bytes, 200, { "Content-Type": "application/json", - "Content-Disposition": f'attachment; filename="{fname}"', + "Content-Disposition": f'attachment; filename="{filename}"', } @app.route('/api/assessments/review/import-custom-data', methods=['POST']) def import_review_custom_data() -> ResponseReturnValue: - """Import assessments, CVSS scores and time estimates from a custom-data - JSON file. + """Import handmade and pending AI assessments, CVSS scores and time + estimates from a custom-data JSON file. Accepts either: @@ -564,17 +526,10 @@ def import_review_custom_data() -> ResponseReturnValue: file. * ``application/json`` body with the custom-data payload directly. - Optional query parameter ``variant_id`` to force all imported - assessments to a specific variant. """ import json as _json - - variant_id = None - raw_vid = request.args.get('variant_id') - if raw_vid: - variant_id, err = parse_uuid_or_400(raw_vid, "variant_id") - if err: - return err + if request.args.getlist('variant_id'): + return {"error": "VulnScout JSON import uses the variants in the file"}, 400 # Parse the incoming data data = None @@ -597,7 +552,7 @@ def import_review_custom_data() -> ResponseReturnValue: return {"error": "Invalid custom-data format. Expected {version, assessments, ...}"}, 400 variant_by_name = build_variant_by_name_map() - result = import_custom_data(data, variant_by_name, variant_id) + result = import_custom_data(data, variant_by_name) status_code = 200 if result["status"] == "success" else 400 return result, status_code diff --git a/tests/cli_tests/test_cmd_assessments.py b/tests/cli_tests/test_cmd_assessments.py index d138536e9..8ff3bf6e1 100644 --- a/tests/cli_tests/test_cmd_assessments.py +++ b/tests/cli_tests/test_cmd_assessments.py @@ -1,11 +1,8 @@ # Copyright (C) 2026 Savoir-faire Linux, Inc. # SPDX-License-Identifier: GPL-3.0-only -"""Coverage tests for src/bin/cmd_assessments.py. - -Targets uncovered branches: - cmd_assessments.py – line 107 (assert variant_obj in single-variant export) -""" +"""Coverage tests for src/bin/cmd_assessments.py.""" +import json import pytest from unittest.mock import patch @@ -25,11 +22,11 @@ def app(tmp_path, monkeypatch): _db.drop_all() -class TestExportCustomAssessmentsVariant: - """Cover the single-variant export path (line 107: assert variant_obj).""" +class TestExportCustomOpenVexAssessments: + """Cover the single-variant OpenVEX export path.""" def test_export_with_variant_covers_assert(self, app, tmp_path): - """Providing --variant hits the else branch and evaluates assert variant_obj.""" + """Providing --variant exports custom assessments as one OpenVEX document.""" from src.models.project import Project from src.models.variant import Variant from src.models.package import Package @@ -53,11 +50,51 @@ def test_export_with_variant_covers_assert(self, app, tmp_path): runner = app.test_cli_runner() result = runner.invoke(args=[ - "export-custom-assessments", + "export-custom-openvex-assessments", "--project", "AssessProj", "--variant", "v1", "--output-dir", str(tmp_path), ]) assert result.exit_code == 0, result.output - assert "Custom assessments exported" in result.output + assert "Custom OpenVEX assessments exported" in result.output + + +class TestExportCustomVulnScoutData: + """Cover the custom JSON export path.""" + + def test_export_includes_pending_ai_assessments(self, app, tmp_path): + """AI rows are preserved for the Review page AI Assessments tab.""" + from src.models.project import Project + from src.models.variant import Variant + from src.models.package import Package + from src.models.vulnerability import Vulnerability + from src.models.finding import Finding + from src.models.assessment import Assessment + + with app.app_context(): + project = Project.create("AiAssessProj") + variant = Variant.create("v1", project.id) + package = Package.create("testpkg", "1.0.0") + vulnerability = Vulnerability.create_record("CVE-2099-5678") + finding = Finding.create(package.id, vulnerability.id) + Assessment.create( + status="under_investigation", + finding_id=finding.id, + variant_id=variant.id, + origin="ai", + ) + + runner = app.test_cli_runner() + result = runner.invoke(args=[ + "export-custom-vulnscout-data", + "--project", "AiAssessProj", + "--output-dir", str(tmp_path), + ]) + + assert result.exit_code == 0, result.output + exported_path = tmp_path / "custom_vulnscout_data_all.json" + exported = json.loads(exported_path.read_text()) + assert exported["assessments"] == [] + assert len(exported["ai_assessments"]) == 1 + assert exported["ai_assessments"][0]["vuln_id"] == "CVE-2099-5678" diff --git a/tests/end_to_end_tests/test_merger_ci.py b/tests/end_to_end_tests/test_merger_ci.py index e3a14c842..192cd52b0 100644 --- a/tests/end_to_end_tests/test_merger_ci.py +++ b/tests/end_to_end_tests/test_merger_ci.py @@ -12,7 +12,6 @@ """ import pytest -import io import json import os @@ -482,7 +481,7 @@ def test_report_command_with_match_condition_cache(app, tmp_path, monkeypatch): # --------------------------------------------------------------------------- -# export-custom-assessments & import-custom-assessments CLI commands +# Custom VulnScout JSON and OpenVEX CLI commands # --------------------------------------------------------------------------- def _create_custom_assessment(app): @@ -516,31 +515,33 @@ def _create_custom_assessment(app): return db_a, variant -def test_export_custom_assessments_no_data(app, tmp_path): +def test_export_custom_openvex_assessments_no_data(app, tmp_path): """Export with no custom assessments exits with error code 1.""" with app.app_context(): runner = app.test_cli_runner() result = runner.invoke(args=[ - "export-custom-assessments", + "export-custom-openvex-assessments", "--project", _PROJECT_NAME, + "--variant", _VARIANT_NAME, "--output-dir", str(tmp_path), ]) assert result.exit_code == 1 assert "No custom assessments" in result.output -def test_export_custom_assessments_success(app, tmp_path): - """Export creates individual OpenVEX JSON files per variant.""" +def test_export_custom_openvex_assessments_success(app, tmp_path): + """Export creates one OpenVEX JSON file for the selected variant.""" _create_custom_assessment(app) with app.app_context(): runner = app.test_cli_runner() result = runner.invoke(args=[ - "export-custom-assessments", + "export-custom-openvex-assessments", "--project", _PROJECT_NAME, + "--variant", _VARIANT_NAME, "--output-dir", str(tmp_path), ]) assert result.exit_code == 0, result.output - out_file = tmp_path / f"{_VARIANT_NAME}.json" + out_file = tmp_path / f"custom_openvex_{_VARIANT_NAME}.json" assert out_file.exists() doc = json.loads(out_file.read_text()) @@ -548,92 +549,72 @@ def test_export_custom_assessments_success(app, tmp_path): assert len(doc["statements"]) >= 1 -def test_export_custom_assessments_compress(app, tmp_path): - """Export with --compress creates custom_assessments.tar.gz.""" +def test_export_custom_vulnscout_data_success_variant(app, tmp_path): + """Export creates a VulnScout JSON document for the selected variant.""" _create_custom_assessment(app) with app.app_context(): runner = app.test_cli_runner() result = runner.invoke(args=[ - "export-custom-assessments", + "export-custom-vulnscout-data", "--project", _PROJECT_NAME, - "--compress", + "--variant", _VARIANT_NAME, "--output-dir", str(tmp_path), ]) assert result.exit_code == 0, result.output - out_file = tmp_path / "custom_assessments.tar.gz" + out_file = tmp_path / f"custom_vulnscout_data_{_VARIANT_NAME}.json" assert out_file.exists() + document = json.loads(out_file.read_text()) + assert document["version"] == 1 + assert "openvex" not in document.get("@context", "") - import tarfile as _tf - with _tf.open(str(out_file), "r:gz") as tar: - members = tar.getnames() - assert len(members) >= 1 - f = tar.extractfile(members[0]) - doc = json.loads(f.read()) - assert "openvex" in doc["@context"] - assert len(doc["statements"]) >= 1 - -def test_export_custom_assessments_success_variant(app, tmp_path): - """Export creates variant.json OpenVEX.""" +def test_export_custom_vulnscout_data_all_variants(app, tmp_path): + """Export without --variant creates a VulnScout JSON document for the project.""" _create_custom_assessment(app) with app.app_context(): runner = app.test_cli_runner() result = runner.invoke(args=[ - "export-custom-assessments", + "export-custom-vulnscout-data", "--project", _PROJECT_NAME, - "--variant", _VARIANT_NAME, "--output-dir", str(tmp_path), ]) assert result.exit_code == 0, result.output - out_file = tmp_path / f"{_VARIANT_NAME}.json" + out_file = tmp_path / "custom_vulnscout_data_all.json" assert out_file.exists() -def test_import_custom_assessments_file_not_found(app, tmp_path): +def test_import_custom_openvex_assessments_file_not_found(app, tmp_path): """Import with a nonexistent file exits with error code 1.""" with app.app_context(): runner = app.test_cli_runner() result = runner.invoke(args=[ - "import-custom-assessments", + "import-custom-openvex-assessments", "--project", _PROJECT_NAME, - str(tmp_path / "nonexistent.tar.gz"), + "--variant", _VARIANT_NAME, + str(tmp_path / "nonexistent.json"), ]) assert result.exit_code == 1 - assert "file not found" in result.output + assert "file not found" in result.output.lower() -def test_import_custom_assessments_unsupported_type(app, tmp_path): +def test_import_custom_openvex_assessments_unsupported_type(app, tmp_path): """Import with an unsupported file type exits with error code 1.""" bad_file = tmp_path / "data.xml" bad_file.write_text("") with app.app_context(): runner = app.test_cli_runner() result = runner.invoke(args=[ - "import-custom-assessments", + "import-custom-openvex-assessments", "--project", _PROJECT_NAME, + "--variant", _VARIANT_NAME, str(bad_file), ]) assert result.exit_code == 1 assert "unsupported file type" in result.output.lower() -def test_import_custom_assessments_invalid_targz(app, tmp_path): - """Import with a corrupt tar.gz exits with error code 1.""" - bad_archive = tmp_path / "corrupt.tar.gz" - bad_archive.write_bytes(b"not a tar.gz") - with app.app_context(): - runner = app.test_cli_runner() - result = runner.invoke(args=[ - "import-custom-assessments", - "--project", _PROJECT_NAME, - str(bad_archive), - ]) - assert result.exit_code == 1 - assert "unable to open" in result.output.lower() - - -def test_import_custom_assessments_json_unknown_variant(app, tmp_path): - """Import a .json with a filename that doesn't match any variant.""" +def test_import_custom_openvex_assessments_requires_variant(app, tmp_path): + """OpenVEX import requires an explicit target variant.""" doc = { "@context": "https://openvex.dev/ns/v0.2.0", "statements": [], @@ -643,45 +624,47 @@ def test_import_custom_assessments_json_unknown_variant(app, tmp_path): with app.app_context(): runner = app.test_cli_runner() result = runner.invoke(args=[ - "import-custom-assessments", + "import-custom-openvex-assessments", "--project", _PROJECT_NAME, str(json_file), ]) - assert result.exit_code == 1 - assert "no variant found" in result.output.lower() + assert result.exit_code == 2 + assert "missing option '--variant'" in result.output.lower() -def test_import_custom_assessments_json_invalid_json(app, tmp_path): +def test_import_custom_openvex_assessments_invalid_json(app, tmp_path): """Import a .json with invalid JSON exits with error code 1.""" json_file = tmp_path / f"{_VARIANT_NAME}.json" json_file.write_text("{invalid json") with app.app_context(): runner = app.test_cli_runner() result = runner.invoke(args=[ - "import-custom-assessments", + "import-custom-openvex-assessments", "--project", _PROJECT_NAME, + "--variant", _VARIANT_NAME, str(json_file), ]) assert result.exit_code == 1 assert "invalid json" in result.output.lower() -def test_import_custom_assessments_json_not_openvex(app, tmp_path): +def test_import_custom_openvex_assessments_not_openvex(app, tmp_path): """Import a .json that is not OpenVEX exits with error code 1.""" json_file = tmp_path / f"{_VARIANT_NAME}.json" json_file.write_text(json.dumps({"hello": "world"})) with app.app_context(): runner = app.test_cli_runner() result = runner.invoke(args=[ - "import-custom-assessments", + "import-custom-openvex-assessments", "--project", _PROJECT_NAME, + "--variant", _VARIANT_NAME, str(json_file), ]) assert result.exit_code == 1 assert "not a valid openvex" in result.output.lower() -def test_import_custom_assessments_json_success(app, tmp_path): +def test_import_custom_openvex_assessments_success(app, tmp_path): """Import a valid .json creates assessments.""" doc = { "@context": "https://openvex.dev/ns/v0.2.0", @@ -697,15 +680,16 @@ def test_import_custom_assessments_json_success(app, tmp_path): with app.app_context(): runner = app.test_cli_runner() result = runner.invoke(args=[ - "import-custom-assessments", + "import-custom-openvex-assessments", "--project", _PROJECT_NAME, + "--variant", _VARIANT_NAME, str(json_file), ]) assert result.exit_code == 0, result.output - assert "Imported 1 assessments" in result.output + assert "Imported 1 OpenVEX assessments" in result.output -def test_import_custom_assessments_custom_data_format(app, tmp_path): +def test_import_custom_vulnscout_data(app, tmp_path): """Import the web 'export custom data' format (non-OpenVEX) via the CLI. The filename does not match a variant; the embedded ``variant_id`` is used. @@ -723,7 +707,7 @@ def test_import_custom_assessments_custom_data_format(app, tmp_path): with app.app_context(): runner = app.test_cli_runner() result = runner.invoke(args=[ - "import-custom-assessments", + "import-custom-vulnscout-data", "--project", _PROJECT_NAME, str(json_file), ]) @@ -732,229 +716,36 @@ def test_import_custom_assessments_custom_data_format(app, tmp_path): assert "time estimates" in result.output.lower() -def test_import_custom_assessments_json_success_variant_flag(app, tmp_path): - """Import a .json with a filename that doesn't match any variant.""" - doc = { - "@context": "https://openvex.dev/ns/v0.2.0", - "statements": [{ - "vulnerability": {"name": "CVE-2020-35492"}, - "status": "affected", - "products": [{"@id": "cairo@1.16.0"}], - "status_notes": "imported via CLI", - }], - } - json_file = tmp_path / "nonexistent_variant.json" - json_file.write_text(json.dumps(doc)) +def test_import_custom_openvex_assessments_rejects_vulnscout_json(app, tmp_path): + """OpenVEX import rejects a VulnScout JSON document.""" + json_file = tmp_path / "custom.json" + json_file.write_text(json.dumps({"version": 1, "assessments": []})) with app.app_context(): runner = app.test_cli_runner() result = runner.invoke(args=[ - "import-custom-assessments", + "import-custom-openvex-assessments", "--project", _PROJECT_NAME, "--variant", _VARIANT_NAME, str(json_file), ]) - assert result.exit_code == 0, result.output - assert "Imported 1 assessments" in result.output - - -def test_import_custom_assessments_targz_no_matching(app, tmp_path): - """Import a tar.gz with no matching variant files exits 1.""" - import tarfile as _tf - - buf = io.BytesIO() - with _tf.open(fileobj=buf, mode='w:gz') as tar: - content = json.dumps({ - "@context": "https://openvex.dev/ns/v0.2.0", - "statements": [], - }).encode() - info = _tf.TarInfo(name="unknown_variant.json") - info.size = len(content) - tar.addfile(info, io.BytesIO(content)) - - archive = tmp_path / "assessments.tar.gz" - archive.write_bytes(buf.getvalue()) - - with app.app_context(): - runner = app.test_cli_runner() - result = runner.invoke(args=[ - "import-custom-assessments", - "--project", _PROJECT_NAME, - str(archive), - ]) - assert result.exit_code == 1 - assert "no valid openvex" in result.output.lower() - - -def test_import_custom_assessments_targz_variant_flag(app, tmp_path): - """Import a tar.gz with --variant fails.""" - archive = tmp_path / "assessments.tar.gz" - archive.touch() - - with app.app_context(): - runner = app.test_cli_runner() - result = runner.invoke(args=[ - "import-custom-assessments", - "--project", _PROJECT_NAME, - "--variant", _VARIANT_NAME, - str(archive), - ]) - assert result.exit_code == 1 - assert "cannot use the --variant" in result.output.lower() - - -def test_import_custom_assessments_directory_success(app, tmp_path): - """Import from a directory of JSON files.""" - doc = { - "@context": "https://openvex.dev/ns/v0.2.0", - "statements": [{ - "vulnerability": {"name": "CVE-2020-35492"}, - "status": "affected", - "products": [{"@id": "cairo@1.16.0"}], - "status_notes": "imported from dir", - }], - } - json_file = tmp_path / f"{_VARIANT_NAME}.json" - json_file.write_text(json.dumps(doc)) - with app.app_context(): - runner = app.test_cli_runner() - result = runner.invoke(args=[ - "import-custom-assessments", - "--project", _PROJECT_NAME, - str(tmp_path), - ]) - assert result.exit_code == 0, result.output - assert "Imported 1 assessments" in result.output - - -def test_import_custom_assessments_directory_no_matching(app, tmp_path): - """Import from a directory with no matching variant files exits 1.""" - doc = { - "@context": "https://openvex.dev/ns/v0.2.0", - "statements": [], - } - json_file = tmp_path / "unknown_variant.json" - json_file.write_text(json.dumps(doc)) - with app.app_context(): - runner = app.test_cli_runner() - result = runner.invoke(args=[ - "import-custom-assessments", - "--project", _PROJECT_NAME, - str(tmp_path), - ]) - assert result.exit_code == 1 - assert "no valid openvex" in result.output.lower() - - -def test_import_custom_assessments_directory_variant_flag(app, tmp_path): - """Import from a directory with --variant fails.""" - (tmp_path / f"{_VARIANT_NAME}.json").write_text("{}") - with app.app_context(): - runner = app.test_cli_runner() - result = runner.invoke(args=[ - "import-custom-assessments", - "--project", _PROJECT_NAME, - "--variant", _VARIANT_NAME, - str(tmp_path), - ]) - assert result.exit_code == 1 - assert "cannot use the --variant" in result.output.lower() - - -def test_import_custom_assessments_directory_empty(app, tmp_path): - """Import from an empty directory exits 1.""" - empty_dir = tmp_path / "empty" - empty_dir.mkdir() - with app.app_context(): - runner = app.test_cli_runner() - result = runner.invoke(args=[ - "import-custom-assessments", - "--project", _PROJECT_NAME, - str(empty_dir), - ]) assert result.exit_code == 1 - assert "no .json files found" in result.output.lower() - - -def test_export_import_roundtrip_directory(app, tmp_path): - """Export individual files then import directory produces same assessments.""" - _create_custom_assessment(app) - - # Export (individual files) - with app.app_context(): - runner = app.test_cli_runner() - result = runner.invoke(args=[ - "export-custom-assessments", - "--project", _PROJECT_NAME, - "--output-dir", str(tmp_path), - ]) - assert result.exit_code == 0, result.output - - # Delete all to have a clean slate, then import directory - with app.app_context(): - from src.extensions import db as _db - from src.models.assessment import Assessment - for a in Assessment.get_by_origin(): - a.delete() - _db.session.commit() - - runner = app.test_cli_runner() - result = runner.invoke(args=[ - "import-custom-assessments", - "--project", _PROJECT_NAME, - str(tmp_path), - ]) - assert result.exit_code == 0, result.output - assert "Imported 1 assessments" in result.output - - -def test_export_import_roundtrip(app, tmp_path): - """Export then import produces same number of assessments.""" - _create_custom_assessment(app) - - # Export (individual files, new default) - with app.app_context(): - runner = app.test_cli_runner() - result = runner.invoke(args=[ - "export-custom-assessments", - "--project", _PROJECT_NAME, - "--output-dir", str(tmp_path), - ]) - assert result.exit_code == 0, result.output - - # Delete all to have a clean slate, then import the individual file - with app.app_context(): - from src.extensions import db as _db - from src.models.assessment import Assessment - for a in Assessment.get_by_origin(): - a.delete() - _db.session.commit() - - runner = app.test_cli_runner() - result = runner.invoke(args=[ - "import-custom-assessments", - "--project", _PROJECT_NAME, - str(tmp_path / f"{_VARIANT_NAME}.json"), - ]) - assert result.exit_code == 0, result.output - assert "Imported 1 assessments" in result.output + assert "not a valid openvex" in result.output.lower() -def test_export_import_roundtrip_compress(app, tmp_path): - """Export with --compress then import tar.gz produces same assessments.""" +def test_export_import_openvex_roundtrip(app, tmp_path): + """Export then import an OpenVEX document for one variant.""" _create_custom_assessment(app) - # Export (compressed) with app.app_context(): runner = app.test_cli_runner() result = runner.invoke(args=[ - "export-custom-assessments", + "export-custom-openvex-assessments", "--project", _PROJECT_NAME, - "--compress", + "--variant", _VARIANT_NAME, "--output-dir", str(tmp_path), ]) assert result.exit_code == 0, result.output - # Delete all to have a clean slate, then import with app.app_context(): from src.extensions import db as _db from src.models.assessment import Assessment @@ -964,24 +755,25 @@ def test_export_import_roundtrip_compress(app, tmp_path): runner = app.test_cli_runner() result = runner.invoke(args=[ - "import-custom-assessments", + "import-custom-openvex-assessments", "--project", _PROJECT_NAME, - str(tmp_path / "custom_assessments.tar.gz"), + "--variant", _VARIANT_NAME, + str(tmp_path / f"custom_openvex_{_VARIANT_NAME}.json"), ]) assert result.exit_code == 0, result.output - assert "Imported 1 assessments" in result.output + assert "Imported 1 OpenVEX assessments" in result.output -def test_import_custom_assessments_skips_duplicates(app, tmp_path): - """Importing the same data twice skips duplicates.""" +def test_import_custom_openvex_assessments_skips_duplicates(app, tmp_path): + """Importing the same OpenVEX data twice skips duplicates.""" _create_custom_assessment(app) - # Export (individual files) with app.app_context(): runner = app.test_cli_runner() runner.invoke(args=[ - "export-custom-assessments", + "export-custom-openvex-assessments", "--project", _PROJECT_NAME, + "--variant", _VARIANT_NAME, "--output-dir", str(tmp_path), ]) @@ -989,66 +781,15 @@ def test_import_custom_assessments_skips_duplicates(app, tmp_path): with app.app_context(): runner = app.test_cli_runner() result = runner.invoke(args=[ - "import-custom-assessments", + "import-custom-openvex-assessments", "--project", _PROJECT_NAME, - str(tmp_path / f"{_VARIANT_NAME}.json"), + "--variant", _VARIANT_NAME, + str(tmp_path / f"custom_openvex_{_VARIANT_NAME}.json"), ]) assert result.exit_code == 0, result.output assert "1 skipped" in result.output -def test_import_custom_assessments_targz_invalid_json_inside( - app, tmp_path -): - """Import tar.gz with invalid JSON inside continues gracefully.""" - import tarfile as _tf - - buf = io.BytesIO() - with _tf.open(fileobj=buf, mode='w:gz') as tar: - content = b"{" - info = _tf.TarInfo(name=f"{_VARIANT_NAME}.json") - info.size = len(content) - tar.addfile(info, io.BytesIO(content)) - - archive = tmp_path / "bad_inner.tar.gz" - archive.write_bytes(buf.getvalue()) - - with app.app_context(): - runner = app.test_cli_runner() - result = runner.invoke(args=[ - "import-custom-assessments", - "--project", _PROJECT_NAME, - str(archive), - ]) - assert result.exit_code == 1 - - -def test_import_custom_assessments_targz_not_openvex_inside( - app, tmp_path -): - """Import tar.gz with non-OpenVEX JSON inside reports error.""" - import tarfile as _tf - - buf = io.BytesIO() - with _tf.open(fileobj=buf, mode='w:gz') as tar: - content = json.dumps({"hello": "world"}).encode() - info = _tf.TarInfo(name=f"{_VARIANT_NAME}.json") - info.size = len(content) - tar.addfile(info, io.BytesIO(content)) - - archive = tmp_path / "not_vex.tar.gz" - archive.write_bytes(buf.getvalue()) - - with app.app_context(): - runner = app.test_cli_runner() - result = runner.invoke(args=[ - "import-custom-assessments", - "--project", _PROJECT_NAME, - str(archive), - ]) - assert result.exit_code == 1 - - def test_list_projects(cli_runner): out: CliResult = cli_runner.invoke(args=[ "list-projects" diff --git a/tests/unit_tests/test_assessment_io.py b/tests/unit_tests/test_assessment_io.py index 348521120..24f8025d6 100644 --- a/tests/unit_tests/test_assessment_io.py +++ b/tests/unit_tests/test_assessment_io.py @@ -5,11 +5,7 @@ """Unit tests for assessment_io.py helper functions.""" -import io import json -import os -import tarfile -import tempfile from pathlib import Path from unittest import mock @@ -19,11 +15,8 @@ is_openvex_doc, sanitize_variant_name, _get_vuln_info, - import_archive_bytes, - import_directory, build_variant_by_name_map, build_openvex_doc, - build_openvex_archive, import_statements, ) @@ -232,147 +225,6 @@ def test_vuln_cached_on_second_call(self): assert result["url"] == "" -# --------------------------------------------------------------------------- -# import_archive_bytes() tests -# --------------------------------------------------------------------------- - -class TestImportArchiveBytes: - """Test import_archive_bytes function.""" - - def test_invalid_tar_gz_bytes(self): - """GIVEN invalid tar.gz bytes WHEN imported THEN raise ValueError.""" - invalid_bytes = b"this is not a tar file" - variant_by_name = {} - - with pytest.raises(ValueError, match="Unable to open tar.gz archive"): - import_archive_bytes(invalid_bytes, variant_by_name) - - def test_tar_with_no_json_files(self): - """GIVEN a tar.gz with no JSON files WHEN imported THEN skip them.""" - # Create a tar archive with non-JSON files - tar_buffer = io.BytesIO() - with tarfile.open(fileobj=tar_buffer, mode='w:gz') as tar: - # Add non-JSON files - info = tarfile.TarInfo(name="readme.txt") - info.size = 5 - tar.addfile(info, io.BytesIO(b"hello")) - - variant_by_name = {} - created, errors, skipped, variant_files_found = import_archive_bytes( - tar_buffer.getvalue(), - variant_by_name - ) - - assert variant_files_found == 0 - assert created == [] - assert errors == [] - - def test_tar_with_missing_variant(self): - """GIVEN a tar with JSON for unknown variant WHEN imported THEN add error.""" - tar_buffer = io.BytesIO() - with tarfile.open(fileobj=tar_buffer, mode='w:gz') as tar: - doc = {"@context": "https://openvex.dev/ns/v0.2.0", "statements": []} - json_bytes = json.dumps(doc).encode() - info = tarfile.TarInfo(name="unknown_variant.json") - info.size = len(json_bytes) - tar.addfile(info, io.BytesIO(json_bytes)) - - variant_by_name = {} - created, errors, skipped, variant_files_found = import_archive_bytes( - tar_buffer.getvalue(), - variant_by_name - ) - - assert variant_files_found == 0 - assert len(errors) == 1 - assert "No variant found" in errors[0]["error"] - - def test_tar_with_directory_member(self): - """GIVEN a tar with directory entries WHEN imported THEN skip them.""" - tar_buffer = io.BytesIO() - with tarfile.open(fileobj=tar_buffer, mode='w:gz') as tar: - # Add a directory - info = tarfile.TarInfo(name="subdir/") - info.type = tarfile.DIRTYPE - tar.addfile(info) - - variant_by_name = {} - created, errors, skipped, variant_files_found = import_archive_bytes( - tar_buffer.getvalue(), - variant_by_name - ) - - assert variant_files_found == 0 - - -# --------------------------------------------------------------------------- -# import_directory() tests -# --------------------------------------------------------------------------- - -class TestImportDirectory: - """Test import_directory function.""" - - def test_empty_directory(self): - """GIVEN a directory with no JSON files WHEN imported THEN raise ValueError.""" - with tempfile.TemporaryDirectory() as tmpdir: - variant_by_name = {} - - with pytest.raises(ValueError, match="No .json files found"): - import_directory(tmpdir, variant_by_name) - - def test_directory_with_non_json_files(self): - """GIVEN a directory with only non-JSON files WHEN imported THEN raise ValueError.""" - with tempfile.TemporaryDirectory() as tmpdir: - # Create a non-JSON file - with open(os.path.join(tmpdir, "readme.txt"), "w") as f: - f.write("hello") - - variant_by_name = {} - - with pytest.raises(ValueError, match="No .json files found"): - import_directory(tmpdir, variant_by_name) - - def test_directory_with_unknown_variant(self): - """GIVEN a directory with JSON for unknown variant WHEN imported THEN add error.""" - with tempfile.TemporaryDirectory() as tmpdir: - # Create a valid JSON file - doc = {"@context": "https://openvex.dev/ns/v0.2.0", "statements": []} - json_path = os.path.join(tmpdir, "unknown_variant.json") - with open(json_path, "w") as f: - json.dump(doc, f) - - variant_by_name = {} - created, errors, skipped, variant_files_found = import_directory( - tmpdir, - variant_by_name - ) - - assert variant_files_found == 0 - assert len(errors) == 1 - assert "No variant found" in errors[0]["error"] - - def test_directory_sorted_by_filename(self): - """GIVEN a directory with multiple JSON files WHEN imported THEN files are processed.""" - with tempfile.TemporaryDirectory() as tmpdir: - # Create multiple JSON files with different names - for name in ["zebra.json", "apple.json", "monkey.json"]: - doc = {"@context": "https://openvex.dev/ns/v0.2.0", "statements": []} - json_path = os.path.join(tmpdir, name) - with open(json_path, "w") as f: - json.dump(doc, f) - - variant_by_name = {} - - # All 3 files should produce errors for missing variants - created, errors, skipped, variant_files_found = import_directory( - tmpdir, - variant_by_name - ) - - # All 3 files should produce errors for missing variants - assert len(errors) == 3 - - # --------------------------------------------------------------------------- # build_variant_by_name_map() tests # --------------------------------------------------------------------------- @@ -641,91 +493,6 @@ def test_ordering_independent_of_input_order(self): assert names1 == names2 -# --------------------------------------------------------------------------- -# build_openvex_archive() tests -# --------------------------------------------------------------------------- - -class TestBuildOpenvexArchive: - """Unit tests for build_openvex_archive.""" - - def test_empty_assessments_returns_empty_tar(self): - """GIVEN no assessments WHEN building archive THEN tar.gz with no members.""" - result = build_openvex_archive([], {}, "author") - buf = io.BytesIO(result) - with tarfile.open(fileobj=buf, mode="r:gz") as tar: - assert len(tar.getmembers()) == 0 - - def test_assessment_without_variant_goes_to_unassigned(self): - """GIVEN assessment with variant_id=None WHEN building archive THEN unassigned.json.""" - assess = mock.MagicMock() - assess.variant_id = None - assess.to_openvex_dict.return_value = {"status": "affected"} - assess.vuln_id = "CVE-2021-1234" - assess.packages = ["pkg@1.0"] - assess.source = "s" - assess.origin = "o" - with mock.patch("src.helpers.assessment_io._get_vuln_info", return_value=_EMPTY_VULN_INFO): - result = build_openvex_archive([assess], {}, "author") - buf = io.BytesIO(result) - with tarfile.open(fileobj=buf, mode="r:gz") as tar: - names = [m.name for m in tar.getmembers()] - assert "unassigned.json" in names - - def test_assessment_with_variant_uses_variant_name(self): - """GIVEN assessment with known variant WHEN building archive THEN file named after variant.""" - import uuid as _uuid - vid = str(_uuid.uuid4()) - assess = mock.MagicMock() - assess.variant_id = vid - assess.to_openvex_dict.return_value = {"status": "fixed"} - assess.vuln_id = "CVE-2021-5678" - assess.packages = ["lib@0.1"] - assess.source = "s" - assess.origin = "o" - variant_names = {vid: "my-variant"} - with mock.patch("src.helpers.assessment_io._get_vuln_info", return_value=_EMPTY_VULN_INFO): - result = build_openvex_archive([assess], variant_names, "author") - buf = io.BytesIO(result) - with tarfile.open(fileobj=buf, mode="r:gz") as tar: - names = [m.name for m in tar.getmembers()] - assert "my-variant.json" in names - - def test_variant_name_with_slash_is_sanitized(self): - """GIVEN a variant name with '/' WHEN building archive THEN slashes become underscores.""" - import uuid as _uuid - vid = str(_uuid.uuid4()) - assess = mock.MagicMock() - assess.variant_id = vid - assess.to_openvex_dict.return_value = None # skipped - assess.packages = [] - variant_names = {vid: "board/arch"} - with mock.patch("src.helpers.assessment_io._get_vuln_info", return_value=_EMPTY_VULN_INFO): - result = build_openvex_archive([assess], variant_names, "author") - buf = io.BytesIO(result) - with tarfile.open(fileobj=buf, mode="r:gz") as tar: - names = [m.name for m in tar.getmembers()] - assert "board_arch.json" in names - - def test_json_content_is_valid_openvex(self): - """GIVEN an archive built from one assessment WHEN extracting THEN JSON is valid OpenVEX.""" - assess = mock.MagicMock() - assess.variant_id = None - assess.to_openvex_dict.return_value = {"status": "not_affected"} - assess.vuln_id = "CVE-2021-9999" - assess.packages = ["pkg@2.0"] - assess.source = "scanner" - assess.origin = "custom" - with mock.patch("src.helpers.assessment_io._get_vuln_info", return_value={"description": "d", "aliases": [], "url": "https://example.com"}): - result = build_openvex_archive([assess], {}, "author", now_iso="2025-01-01T00:00:00Z") - buf = io.BytesIO(result) - with tarfile.open(fileobj=buf, mode="r:gz") as tar: - member = tar.getmembers()[0] - doc = json.load(tar.extractfile(member)) - assert is_openvex_doc(doc) - assert doc["author"] == "author" - assert doc["timestamp"] == "2025-01-01T00:00:00Z" - - # --------------------------------------------------------------------------- # import_statements() — unit tests for early-exit paths (no DB needed) # --------------------------------------------------------------------------- diff --git a/tests/unit_tests/test_assessment_io_coverage.py b/tests/unit_tests/test_assessment_io_coverage.py index a52620935..275c2b2f2 100644 --- a/tests/unit_tests/test_assessment_io_coverage.py +++ b/tests/unit_tests/test_assessment_io_coverage.py @@ -3,16 +3,14 @@ # SPDX-License-Identifier: GPL-3.0-only """Coverage tests for assessment_io.py targeting the lines missed by CI. -Lines covered: 321-322, 387, 447-449, 452-456, 547-548, 579, +Lines covered: 321-322, 387, 547-548, 579, 684, 698, 706, 710-714, 717-721, 734, 740, 772-773, 784, 787, 810-814, 822, 833-835, 844, 847, 850-854, 862-863, 871-875, 882. """ -import io import json import os -import tarfile import uuid as _uuid from unittest import mock @@ -20,7 +18,6 @@ from src.helpers.assessment_io import ( build_variant_by_name_map, - import_archive_bytes, import_custom_data, import_statements, build_custom_data_export, @@ -56,22 +53,6 @@ def variant_and_project(app): return proj, var -def _make_openvex_tar(entries: "dict[str, bytes]") -> bytes: - """Build a .tar.gz with {filename: bytes} entries.""" - buf = io.BytesIO() - with tarfile.open(fileobj=buf, mode="w:gz") as tar: - for name, data in entries.items(): - info = tarfile.TarInfo(name=name) - info.size = len(data) - tar.addfile(info, io.BytesIO(data)) - return buf.getvalue() - - -def _openvex_bytes(statements=None) -> bytes: - doc = {"@context": "https://openvex.dev/ns/v0.2.0", "statements": statements or []} - return json.dumps(doc).encode() - - # =========================================================================== # import_statements — lines 321-322: except handler when DB.create fails # =========================================================================== @@ -112,53 +93,6 @@ def test_returns_only_variants_for_project(self, app, variant_and_project): assert result["io-cov-var"].id == var.id -# =========================================================================== -# import_archive_bytes — lines 447-456 -# =========================================================================== - -class TestImportArchiveBytesEdgeCases: - """Lines 447-449 (f is None), 452-456 (invalid JSON / not OpenVEX).""" - - def test_extractfile_none_is_skipped(self, app, variant_and_project): - """Line 448-449: tar.extractfile returns None → member silently skipped.""" - _, var = variant_and_project - variant_by_name = {"io-cov-var": var} - - tar_bytes = _make_openvex_tar({"io-cov-var.json": _openvex_bytes()}) - - with mock.patch.object(tarfile.TarFile, "extractfile", return_value=None): - with app.app_context(): - created, errors, skipped, found = import_archive_bytes(tar_bytes, variant_by_name) - - # member was skipped silently — no error, no files found - assert found == 0 - assert errors == [] - - def test_invalid_json_in_archive_appends_error(self, app, variant_and_project): - """Lines 452-453: JSON decode error → error entry appended.""" - _, var = variant_and_project - variant_by_name = {"io-cov-var": var} - tar_bytes = _make_openvex_tar({"io-cov-var.json": b"{ not valid json !!"}) - - with app.app_context(): - created, errors, skipped, found = import_archive_bytes(tar_bytes, variant_by_name) - - assert found == 0 - assert any("Invalid JSON" in e.get("error", "") for e in errors) - - def test_non_openvex_json_in_archive_appends_error(self, app, variant_and_project): - """Lines 452-456: valid JSON but not OpenVEX → error entry appended.""" - _, var = variant_and_project - variant_by_name = {"io-cov-var": var} - tar_bytes = _make_openvex_tar({"io-cov-var.json": b'{"key": "value"}'}) - - with app.app_context(): - created, errors, skipped, found = import_archive_bytes(tar_bytes, variant_by_name) - - assert found == 0 - assert any("Not a valid OpenVEX" in e.get("error", "") for e in errors) - - # =========================================================================== # build_custom_data_export — lines 547-548, 579 # =========================================================================== @@ -196,6 +130,42 @@ def test_variant_ids_filter_and_variant_name_resolved(self, app, variant_and_pro # variant name should be resolved on the exported assessment assert result["assessments"][0]["variant"] == "io-cov-var" + def test_pending_ai_assessments_are_exported(self, app, variant_and_project): + """Pending AI rows are emitted separately for the Review page AI tab.""" + from src.models.package import Package + from src.models.vulnerability import Vulnerability + from src.models.finding import Finding + from src.models.assessment import Assessment + + _, var = variant_and_project + + with app.app_context(): + pkg = Package.create("ai-pkg", "1.0.0") + vuln = Vulnerability.create_record("CVE-2099-AI01") + finding = Finding.create(pkg.id, vuln.id) + Assessment.create( + status="under_investigation", + finding_id=finding.id, + variant_id=var.id, + origin="ai", + ) + + result = build_custom_data_export(variant_ids=[var.id]) + + assert result["assessments"] == [] + assert result["ai_assessments"] == [{ + "vuln_id": "CVE-2099-AI01", + "status": "under_investigation", + "simplified_status": "", + "justification": None, + "impact_statement": None, + "status_notes": None, + "workaround": None, + "packages": ["ai-pkg@1.0.0"], + "variant_id": str(var.id), + "variant": "io-cov-var", + }] + # =========================================================================== # import_custom_data — assessments section @@ -302,6 +272,30 @@ def test_variant_name_resolution_via_variant_by_name(self, app, variant_and_proj result = import_custom_data(data, variant_by_name) assert result["assessments_imported"] == 1 + def test_import_ai_assessment_preserves_pending_origin(self, app, variant_and_project): + """AI JSON rows return to the Review page as pending AI assessments.""" + from src.models.assessment import Assessment + + _, var = variant_and_project + data = { + "ai_assessments": [{ + "vuln_id": "CVE-2099-AI02", + "status": "affected", + "packages": ["ai-import@1.0"], + "variant_id": str(var.id), + }] + } + with app.app_context(): + result = import_custom_data(data, {}) + imported = Assessment.get_by_origin([var.id], origin="ai") + duplicate_result = import_custom_data(data, {}) + + assert result["ai_assessments_imported"] == 1 + assert result["assessments_imported"] == 0 + assert len(imported) == 1 + assert imported[0].vuln_id == "CVE-2099-AI02" + assert duplicate_result["ai_assessments_skipped"] == 1 + # =========================================================================== # import_custom_data — CVSS section diff --git a/tests/webapp_tests/test_review_endpoints.py b/tests/webapp_tests/test_review_endpoints.py index b02bbe245..dee87351d 100644 --- a/tests/webapp_tests/test_review_endpoints.py +++ b/tests/webapp_tests/test_review_endpoints.py @@ -11,7 +11,6 @@ import io import json -import tarfile import uuid import pytest @@ -406,60 +405,69 @@ def test_assessments_list_by_project_invalid(client): # ── GET /api/assessments/review/export ─────────────────────────────────── -def test_export_empty(client): - """No handmade assessments → 404.""" +def test_export_requires_one_variant(client): + """OpenVEX export needs an explicit single variant.""" resp = client.get("/api/assessments/review/export") - assert resp.status_code == 404 + assert resp.status_code == 400 + assert "variant" in json.loads(resp.data)["error"].lower() -def test_export_tar_gz(client): +def test_export_openvex_json(client): _create_handmade_assessment(client) - resp = client.get("/api/assessments/review/export") + resp = client.get(f"/api/assessments/review/export?variant_id={VARIANT_UUID}") assert resp.status_code == 200 - assert resp.content_type == "application/gzip" - buf = io.BytesIO(resp.data) - with tarfile.open(fileobj=buf, mode="r:gz") as tar: - members = tar.getmembers() - assert len(members) >= 1 - # Each member should be a valid OpenVEX JSON - for m in members: - assert m.name.endswith(".json") - f = tar.extractfile(m) - doc = json.load(f) - assert "openvex" in doc.get("@context", "") - assert isinstance(doc.get("statements"), list) - for stmt in doc["statements"]: - assert "vulnerability" in stmt - assert "products" in stmt - assert "status" in stmt + assert resp.content_type == "application/json" + doc = json.loads(resp.data) + assert "openvex" in doc.get("@context", "") + assert isinstance(doc.get("statements"), list) + for stmt in doc["statements"]: + assert "vulnerability" in stmt + assert "products" in stmt + assert "status" in stmt def test_export_contains_variant_name(client): _create_handmade_assessment(client) - resp = client.get("/api/assessments/review/export") - buf = io.BytesIO(resp.data) - with tarfile.open(fileobj=buf, mode="r:gz") as tar: - names = [m.name for m in tar.getmembers()] - # The demo variant is named "default" - assert "default.json" in names + resp = client.get(f"/api/assessments/review/export?variant_id={VARIANT_UUID}") + assert 'review_openvex_default.json' in resp.headers["Content-Disposition"] + + +def test_export_rejects_multiple_variants(client, app): + from src.extensions import db + from src.models import Variant + + second_variant_id = uuid.uuid4() + with app.app_context(): + db.session.add(Variant(id=second_variant_id, project_id=PROJECT_UUID, name="second")) + db.session.commit() + + resp = client.get( + "/api/assessments/review/export" + f"?variant_id={VARIANT_UUID}&variant_id={second_variant_id}" + ) + + assert resp.status_code == 400 + + +def test_export_selected_variants_invalid_uuid(client): + _create_handmade_assessment(client) + resp = client.get("/api/assessments/review/export?variant_id=not-a-uuid") + assert resp.status_code == 400 def test_export_enriched_fields(client): """Exported statements should have enriched vulnerability and product fields.""" _create_handmade_assessment(client) - resp = client.get("/api/assessments/review/export") - buf = io.BytesIO(resp.data) - with tarfile.open(fileobj=buf, mode="r:gz") as tar: - for m in tar.getmembers(): - doc = json.load(tar.extractfile(m)) - for stmt in doc["statements"]: - vuln = stmt["vulnerability"] - assert "name" in vuln - assert "description" in vuln - assert "aliases" in vuln - for prod in stmt["products"]: - assert "identifiers" in prod - assert "scanners" in stmt + resp = client.get(f"/api/assessments/review/export?variant_id={VARIANT_UUID}") + doc = json.loads(resp.data) + for stmt in doc["statements"]: + vuln = stmt["vulnerability"] + assert "name" in vuln + assert "description" in vuln + assert "aliases" in vuln + for prod in stmt["products"]: + assert "identifiers" in prod + assert "scanners" in stmt # ── POST /api/assessments/review/import ────────────────────────────────── @@ -476,18 +484,6 @@ def _make_openvex_json(variant_name, statements): }).encode("utf-8") -def _make_tar_gz(files_dict): - """Build a tar.gz archive from {filename: bytes} dict.""" - buf = io.BytesIO() - with tarfile.open(fileobj=buf, mode="w:gz") as tar: - for name, data in files_dict.items(): - info = tarfile.TarInfo(name=name) - info.size = len(data) - tar.addfile(info, io.BytesIO(data)) - buf.seek(0) - return buf - - def test_import_no_file(client): resp = client.post("/api/assessments/review/import", content_type="multipart/form-data") @@ -495,7 +491,7 @@ def test_import_no_file(client): def test_import_json_valid(client): - """Import a single .json named after the demo variant.""" + """Import a single .json document into the selected variant.""" statements = [{ "vulnerability": {"name": "CVE-2020-35492"}, "products": [{"@id": "cairo@1.16.0"}], @@ -508,7 +504,7 @@ def test_import_json_valid(client): data = _make_openvex_json("default", statements) resp = client.post( "/api/assessments/review/import", - data={"file": (io.BytesIO(data), "default.json")}, + data={"file": (io.BytesIO(data), "default.json"), "variant_id": str(VARIANT_UUID)}, content_type="multipart/form-data", ) assert resp.status_code == 200 @@ -517,7 +513,7 @@ def test_import_json_valid(client): assert result["imported"] >= 1 -def test_import_json_unknown_variant(client): +def test_import_json_requires_selected_variant(client): data = _make_openvex_json("unknown_variant", []) resp = client.post( "/api/assessments/review/import", @@ -525,98 +521,57 @@ def test_import_json_unknown_variant(client): content_type="multipart/form-data", ) assert resp.status_code == 400 - assert "variant" in json.loads(resp.data)["error"].lower() + assert "variant_id" in json.loads(resp.data)["error"] -def test_import_json_invalid_json(client): +def test_import_json_selected_variant_ignores_filename(client): + data = _make_openvex_json("unrelated", []) resp = client.post( "/api/assessments/review/import", - data={"file": (io.BytesIO(b"not json"), "default.json")}, + data={ + "file": (io.BytesIO(data), "unrelated.json"), + "variant_id": str(VARIANT_UUID), + }, content_type="multipart/form-data", ) - assert resp.status_code == 400 - -def test_import_json_not_openvex(client): - data = json.dumps({"foo": "bar"}).encode() - resp = client.post( - "/api/assessments/review/import", - data={"file": (io.BytesIO(data), "default.json")}, - content_type="multipart/form-data", - ) - assert resp.status_code == 400 - assert "openvex" in json.loads(resp.data)["error"].lower() - - -def test_import_tar_gz_valid(client): - """Import a tar.gz with one file named after the demo variant.""" - statements = [{ - "vulnerability": {"name": "CVE-2020-35492"}, - "products": [{"@id": "cairo@1.16.0"}], - "status": "not_affected", - "justification": "component_not_present", - "impact_statement": "not present", - "status_notes": "", - "action_statement": "", - }] - content = _make_openvex_json("default", statements) - tar_buf = _make_tar_gz({"default.json": content}) - resp = client.post( - "/api/assessments/review/import", - data={"file": (tar_buf, "review.tar.gz")}, - content_type="multipart/form-data", - ) assert resp.status_code == 200 - result = json.loads(resp.data) - assert result["status"] == "success" - assert result["imported"] >= 1 + assert json.loads(resp.data)["status"] == "success" -def test_import_tar_gz_unknown_variant(client): - """Archive with a .json not matching any variant → error.""" - content = _make_openvex_json("nonexistent", [{ - "vulnerability": {"name": "CVE-2020-35492"}, - "products": [{"@id": "cairo@1.16.0"}], - "status": "affected", - }]) - tar_buf = _make_tar_gz({"nonexistent.json": content}) +def test_import_json_selected_variant_not_found(client): + data = _make_openvex_json("unrelated", []) resp = client.post( "/api/assessments/review/import", - data={"file": (tar_buf, "review.tar.gz")}, + data={ + "file": (io.BytesIO(data), "unrelated.json"), + "variant_id": str(uuid.uuid4()), + }, content_type="multipart/form-data", ) - assert resp.status_code == 400 - -def test_import_tar_gz_invalid_archive(client): - resp = client.post( - "/api/assessments/review/import", - data={"file": (io.BytesIO(b"notatar"), "bad.tar.gz")}, - content_type="multipart/form-data", - ) - assert resp.status_code == 400 + assert resp.status_code == 404 + assert "variant" in json.loads(resp.data)["error"].lower() -def test_import_tar_gz_invalid_json_inside(client): - tar_buf = _make_tar_gz({"default.json": b"not json"}) +def test_import_json_invalid_json(client): resp = client.post( "/api/assessments/review/import", - data={"file": (tar_buf, "review.tar.gz")}, + data={"file": (io.BytesIO(b"not json"), "default.json"), "variant_id": str(VARIANT_UUID)}, content_type="multipart/form-data", ) - # The bad JSON is reported as error but request succeeds if no valid files - assert resp.status_code in (200, 400) + assert resp.status_code == 400 -def test_import_tar_gz_not_openvex_inside(client): - content = json.dumps({"not": "openvex"}).encode() - tar_buf = _make_tar_gz({"default.json": content}) +def test_import_json_not_openvex(client): + data = json.dumps({"foo": "bar"}).encode() resp = client.post( "/api/assessments/review/import", - data={"file": (tar_buf, "review.tar.gz")}, + data={"file": (io.BytesIO(data), "default.json"), "variant_id": str(VARIANT_UUID)}, content_type="multipart/form-data", ) - assert resp.status_code in (200, 400) + assert resp.status_code == 400 + assert "openvex" in json.loads(resp.data)["error"].lower() def test_import_unsupported_file_type(client): @@ -652,7 +607,7 @@ def test_import_duplicate_skipped(client): # First import resp1 = client.post( "/api/assessments/review/import", - data={"file": (io.BytesIO(data), "default.json")}, + data={"file": (io.BytesIO(data), "default.json"), "variant_id": str(VARIANT_UUID)}, content_type="multipart/form-data", ) assert resp1.status_code == 200 @@ -662,7 +617,7 @@ def test_import_duplicate_skipped(client): # Second import — same data resp2 = client.post( "/api/assessments/review/import", - data={"file": (io.BytesIO(data), "default.json")}, + data={"file": (io.BytesIO(data), "default.json"), "variant_id": str(VARIANT_UUID)}, content_type="multipart/form-data", ) assert resp2.status_code == 200 @@ -681,7 +636,7 @@ def test_import_statement_missing_vuln(client): data = _make_openvex_json("default", statements) resp = client.post( "/api/assessments/review/import", - data={"file": (io.BytesIO(data), "default.json")}, + data={"file": (io.BytesIO(data), "default.json"), "variant_id": str(VARIANT_UUID)}, content_type="multipart/form-data", ) assert resp.status_code == 200 @@ -697,7 +652,7 @@ def test_import_statement_missing_status(client): data = _make_openvex_json("default", statements) resp = client.post( "/api/assessments/review/import", - data={"file": (io.BytesIO(data), "default.json")}, + data={"file": (io.BytesIO(data), "default.json"), "variant_id": str(VARIANT_UUID)}, content_type="multipart/form-data", ) assert resp.status_code == 200 @@ -713,7 +668,7 @@ def test_import_statement_missing_products(client): data = _make_openvex_json("default", statements) resp = client.post( "/api/assessments/review/import", - data={"file": (io.BytesIO(data), "default.json")}, + data={"file": (io.BytesIO(data), "default.json"), "variant_id": str(VARIANT_UUID)}, content_type="multipart/form-data", ) assert resp.status_code == 200 @@ -735,7 +690,7 @@ def test_import_product_string_format(client): data = _make_openvex_json("default", statements) resp = client.post( "/api/assessments/review/import", - data={"file": (io.BytesIO(data), "default.json")}, + data={"file": (io.BytesIO(data), "default.json"), "variant_id": str(VARIANT_UUID)}, content_type="multipart/form-data", ) assert resp.status_code == 200 @@ -757,7 +712,7 @@ def test_import_product_without_version(client): data = _make_openvex_json("default", statements) resp = client.post( "/api/assessments/review/import", - data={"file": (io.BytesIO(data), "default.json")}, + data={"file": (io.BytesIO(data), "default.json"), "variant_id": str(VARIANT_UUID)}, content_type="multipart/form-data", ) assert resp.status_code == 200 @@ -771,12 +726,12 @@ def test_export_import_round_trip(client): """Export Review → Import Review should be a valid round-trip.""" _create_handmade_assessment(client, status="affected") # Export - export_resp = client.get("/api/assessments/review/export") + export_resp = client.get(f"/api/assessments/review/export?variant_id={VARIANT_UUID}") assert export_resp.status_code == 200 # Import the exported file back import_resp = client.post( "/api/assessments/review/import", - data={"file": (io.BytesIO(export_resp.data), "review.tar.gz")}, + data={"file": (io.BytesIO(export_resp.data), "review.json"), "variant_id": str(VARIANT_UUID)}, content_type="multipart/form-data", ) assert import_resp.status_code == 200 @@ -920,8 +875,8 @@ def test_review_time_estimates_matches_export_via_patch_flow(app, client): rows_all = [e for e in json.loads(resp_all.data) if e["vuln_id"] == "CVE-2020-35492"] assert len({e["variant_id"] for e in rows_all}) == 2 - # Export endpoint (project scope) → two time-estimate entries, matching. - exp = client.get(f"/api/assessments/review/export-custom-data?project_id={PROJECT_UUID}") + # The all-variant export retains both time-estimate entries. + exp = client.get("/api/assessments/review/export-custom-data") assert exp.status_code == 200 exported = json.loads(exp.data)["time_estimates"] exp_rows = [e for e in exported if e["vuln_id"] == "CVE-2020-35492"] @@ -995,14 +950,13 @@ def test_export_custom_data_basic(client): def test_export_custom_data_by_variant(client): - """Filter by variant_id returns only that variant's assessments.""" _create_handmade_assessment(client) resp = client.get(f"/api/assessments/review/export-custom-data?variant_id={VARIANT_UUID}") assert resp.status_code == 200 data = json.loads(resp.data) assert len(data["assessments"]) >= 1 - for a in data["assessments"]: - assert a["variant_id"] == str(VARIANT_UUID) + for assessment in data["assessments"]: + assert assessment["variant_id"] == str(VARIANT_UUID) def test_export_custom_data_by_project(client): @@ -1650,6 +1604,19 @@ def test_export_import_custom_data_round_trip(client): assert result["status"] == "success" +def test_export_custom_data_with_only_pending_ai_assessments(client): + """The Review page can export a pending AI assessment without custom data.""" + _create_ai_assessment(client) + + response = client.get("/api/assessments/review/export-custom-data") + + assert response.status_code == 200 + exported = json.loads(response.data) + assert exported["assessments"] == [] + assert len(exported["ai_assessments"]) == 1 + assert exported["ai_assessments"][0]["vuln_id"] == "CVE-2020-35492" + + # ── review_custom_cvss: variant/project filtering ──────────────────────── def test_review_custom_cvss_by_variant(client): diff --git a/vulnscout b/vulnscout index 82541b94f..6fe9d97e7 100755 --- a/vulnscout +++ b/vulnscout @@ -78,9 +78,10 @@ Scan & output commands: --export-spdx Export project as SPDX 3.0 SBOM --export-cdx Export project as CycloneDX 1.6 SBOM --export-openvex Export project as OpenVEX document - --export-custom-assessments Export custom (review) assessments as individual OpenVEX files - --compress Compress export output into a .tar.gz archive - --import-custom-assessments Import custom assessments from .json, .tar.gz, or directory + --export-custom-vulnscout-data Export custom VulnScout JSON data + --import-custom-vulnscout-data Import custom VulnScout JSON data + --export-custom-openvex-assessments Export custom OpenVEX assessments for --variant + --import-custom-openvex-assessments Import custom OpenVEX assessments into --variant --delete-scan Delete a past scan by its ID Data retrieval commands: @@ -336,18 +337,27 @@ parse_and_run() { ensure_running; EXPORT_ARGS+=(--export-cdx); shift ;; export-openvex|--export-openvex) ensure_running; EXPORT_ARGS+=(--export-openvex); shift ;; - export-custom-assessments|--export-custom-assessments) - ensure_running; EXPORT_ARGS+=(--export-custom-assessments); shift ;; - compress|--compress) - EXPORT_ARGS+=(--compress); shift ;; - import-custom-assessments|--import-custom-assessments) + export-custom-vulnscout-data|--export-custom-vulnscout-data) + ensure_running; EXPORT_ARGS+=(--export-custom-vulnscout-data); shift ;; + export-custom-openvex-assessments|--export-custom-openvex-assessments) + ensure_running; EXPORT_ARGS+=(--export-custom-openvex-assessments); shift ;; + import-custom-vulnscout-data|--import-custom-vulnscout-data) ensure_running local import_file import_file="$(readlink -f "$2")" local import_staged import_staged="/tmp/vulnscout_stage_$(basename "$import_file")" call_container_engine cp "$import_file" "$CONTAINER_NAME:$import_staged" - EXPORT_ARGS+=(--import-custom-assessments "$import_staged") + EXPORT_ARGS+=(--import-custom-vulnscout-data "$import_staged") + shift 2 ;; + import-custom-openvex-assessments|--import-custom-openvex-assessments) + ensure_running + local import_file + import_file="$(readlink -f "$2")" + local import_staged + import_staged="/tmp/vulnscout_stage_$(basename "$import_file")" + call_container_engine cp "$import_file" "$CONTAINER_NAME:$import_staged" + EXPORT_ARGS+=(--import-custom-openvex-assessments "$import_staged") shift 2 ;; delete-scan|--delete-scan) ensure_running