From 7b6c19d8d940ede0e7edcd77cf625055111d5eb4 Mon Sep 17 00:00:00 2001 From: Florian Frank Date: Sun, 20 Sep 2026 19:47:39 +0200 Subject: [PATCH] Implement functionality in order to generate report directly from the artifacts folder. --- README.md | 22 +++- report_generator/__main__.py | 12 ++- report_generator/commands/generate.py | 74 +++++++++++-- report_generator/sources.py | 147 +++++++++++++++++++++++++- 4 files changed, 239 insertions(+), 16 deletions(-) diff --git a/README.md b/README.md index 844e1ee..eaca9e5 100644 --- a/README.md +++ b/README.md @@ -17,10 +17,30 @@ pointing at a module that was never put on the path. ## Running +Two ways to say where the artifacts are. + +**Out of the store** — Postgres for the rows, the bucket for the files: + ``` -report_generator --scan_id [--path ] --config config.json +report_generator generate --scan_id [--path ] --config config.json ``` +**Out of a folder** — neither Postgres nor the bucket, just the artifacts as the pipeline exported +them: + +``` +report_generator from-folder --artifacts ./sample [--path ] +``` + +`--scan_id` is optional there: the pipeline writes its own id into the paths it records in +`metadata.json`, so an export names its own run. Pass it for a folder assembled by hand, or one +whose metadata is gone. + +The two produce the same document. Everything after the artifacts are in hand is shared code, and +the sample export in `sample/` was checked against the report built from the store for the same +scan: 17 of 17 pages identical. The tests cover the folder path; comparing the two needs a store +and so is not among them. + `--scan_id` takes either id a run is known by. A run is recorded under two: the **pipeline id**, which the GUI shows, cipherverifier names its working directories after, and archon's `core_scan` row is keyed by; and the **stored scan id**, minted when the artifacts are uploaded, which `scans`, diff --git a/report_generator/__main__.py b/report_generator/__main__.py index 1675840..0a183e4 100644 --- a/report_generator/__main__.py +++ b/report_generator/__main__.py @@ -4,20 +4,26 @@ from cc_common.api.cli import CLIManager from cc_common.config import load_environment from persistent_storage.shared import inject_secrets -from .commands.generate import GenerateProperties, generate +from .commands.generate import FolderProperties, GenerateProperties, from_folder, generate def run() -> None: """Entrypoint for CLI.""" load_environment() - cli = CLIManager("report_generator", "Generates the PDF report of a stored scan.") + cli = CLIManager("report_generator", "Generates the PDF report of a scan.") cli.register_command( "generate", _with_storage_secrets, GenerateProperties, - help="Builds the PDF report of one scan from the artifacts it stored.", + help="Builds the report of a scan held in the store.", + ) + cli.register_command( + "from-folder", + from_folder, + FolderProperties, + help="Builds the report from a folder of exported artifacts, without Postgres or the bucket.", ) cli.run() diff --git a/report_generator/commands/generate.py b/report_generator/commands/generate.py index fe58500..21f394f 100644 --- a/report_generator/commands/generate.py +++ b/report_generator/commands/generate.py @@ -1,4 +1,9 @@ -"""report_generator generate command.""" +"""The two ways to ask for a report: out of the store, or out of a folder. + +Two commands rather than one with a mode switch, because the two need different things and a +command that demands a database host in order to read a directory is a command nobody will use for +that. What they share is everything after the artifacts are in hand, which is :func:`_report`. +""" from pathlib import Path from uuid import UUID @@ -7,10 +12,12 @@ from pydantic import BaseModel, Field from cc_common.api import PipelineError, context from cc_common.utils import print + +# Imported for the properties model; the store half of the tool is the only part that needs it. from persistent_storage.shared import StorageConfiguration from ..rendering import render -from ..sources import collect +from ..sources import ReportSources, collect, collect_from_directory # Where the report goes when the caller names no path. DEFAULT_DIRECTORY = Path("reports") @@ -20,7 +27,7 @@ class GenerateProperties(BaseModel): """Configuration properties for the generate CLI command. Attributes: - scan_id: The scan to report on. + scan_id: The scan to report on. The store is keyed by it. path: Where to write the PDF. A directory is written into under the name the renderer chose; leaving it out writes into ``reports/``. storage_config: The store the scan's artifacts are read from. @@ -31,6 +38,23 @@ class GenerateProperties(BaseModel): storage_config: StorageConfiguration +class FolderProperties(BaseModel): + """Configuration properties for the from-folder CLI command. + + Attributes: + artifacts: The folder holding the artifacts, as the pipeline exported them. + scan_id: The scan to report under. Read out of the artifacts when not + given -- an export names its own run. + path: Where to write the PDF, as above. + """ + + artifacts: Path = Field(description="Folder of exported artifacts to build the report from.") + scan_id: UUID | None = Field( + default=None, description="The scan to report under. Read from the artifacts when omitted." + ) + path: Path | None = Field(default=None, description="Where to write the PDF report.") + + class GeneratedReport(BaseModel): """The report that was written. @@ -51,10 +75,10 @@ class GeneratedReport(BaseModel): def generate(properties: GenerateProperties) -> GeneratedReport: - """Build the PDF report of one scan and write it. + """Build the report of a scan held in the store. Args: - properties: Parsed command properties naming the scan and the destination. + properties: The scan, the store and the destination. Returns: The report that was written. @@ -63,9 +87,45 @@ def generate(properties: GenerateProperties) -> GeneratedReport: PipelineError: The scan cannot be read, is missing an input the report needs, or the destination cannot be written. """ - sources = collect(properties.storage_config, properties.scan_id) + return _report(collect(properties.storage_config, properties.scan_id), properties.path) + + +def from_folder(properties: FolderProperties) -> GeneratedReport: + """Build the report of a scan that was exported to a folder. + + Neither Postgres nor the bucket is touched. The same report, from a directory somebody was sent. + + Args: + properties: The folder, optionally the scan to report under, and the destination. + + Returns: + The report that was written. + + Raises: + PipelineError: The folder cannot be read, names no scan, is missing an + input the report needs, or the destination cannot be written. + """ + return _report(collect_from_directory(properties.artifacts, properties.scan_id), properties.path) + + +def _report(sources: ReportSources, path: Path | None) -> GeneratedReport: + """Render the collected artifacts and write the result. + + Everything after the artifacts are in hand, which is the same whichever of the two collected + them -- and the reason the two commands are two callbacks rather than two pipelines. + + Args: + sources: The scan and its files. + path: Where the caller asked for the PDF, if anywhere in particular. + + Returns: + The report that was written. + + Raises: + PipelineError: The report cannot be built or the destination not written. + """ name, content = render(sources) - destination = _destination(properties.path, name) + destination = _destination(path, name) try: destination.parent.mkdir(parents=True, exist_ok=True) diff --git a/report_generator/sources.py b/report_generator/sources.py index 89f2434..a4b78e2 100644 --- a/report_generator/sources.py +++ b/report_generator/sources.py @@ -1,14 +1,23 @@ """Collects everything a scan's report is built from. -The one place that knows where the bytes come from. Swapping the store -- persistent_storage today, -archon tomorrow -- is a change to :func:`collect` and to nothing else: what it returns is files by -name, and a renderer cannot tell which store handed them over. +The one place that knows where the bytes come from. Two ways in: + +* :func:`collect` reads a scan out of the store -- Postgres for the rows, the bucket for the files. +* :func:`collect_from_directory` reads a folder of exported artifacts, and needs neither. + +Both return the same thing: files by name. Nothing downstream can tell which of the two handed them +over, which is what makes the second one worth having rather than a second pipeline -- a report of +an exported run is the same report, built by the same code. """ from __future__ import annotations +import json +import re from dataclasses import dataclass -from typing import TYPE_CHECKING, cast +from datetime import UTC, datetime +from pathlib import Path +from typing import TYPE_CHECKING, Any, cast from uuid import UUID from sqlalchemy import create_engine, text @@ -20,7 +29,6 @@ from persistent_storage.utils.s3_utils import fetch_artifact_files if TYPE_CHECKING: from collections.abc import Sequence - from datetime import datetime from mypy_boto3_s3 import S3Client from sqlalchemy.sql.elements import TextClause @@ -254,3 +262,132 @@ def collect(configuration: StorageConfiguration, scan_id: UUID) -> ReportSources artifacts=scan.artifacts, files=dict(zip(names, contents, strict=True)), ) + + +# A UUID wherever it appears inside a longer string, which is how the pipeline writes its own id +# into the paths it records. +_UUID = re.compile(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", re.IGNORECASE) + +# The metadata fields that carry a path ending in the run's own id. Both are written by the +# pipeline itself, so a folder exported from a run identifies itself without being told. +_ID_BEARING = ("export_folder", "repository_path") + +# Read whole rather than streamed: the report needs every file at once, and the largest export in +# the sample set comes to a few megabytes. A folder that does not fit in memory is not an export. +_METADATA = ("metadata.json", "meta.json") + + +def collect_from_directory(directory: Path, scan_id: UUID | None = None) -> ReportSources: + """Read a folder of exported artifacts, without touching Postgres or the bucket. + + The same report from a directory somebody was sent: a scan that was exported, a pipeline run on + a machine with no store attached, a bug report that arrived as a zip. Nothing downstream knows + the difference. + + Args: + directory: The folder holding the artifacts, as the pipeline exported them. + scan_id: The scan to report under. Read out of the artifacts when not + given -- the pipeline writes its own id into the paths it records in + ``metadata.json``, and failing that the folder may be named by it. + + Returns: + The files, under the id the run is known by. + + Raises: + PipelineError: The folder does not exist, holds no files, or holds nothing + that names the run and none was given. + """ + context.logger.info(f"Collecting the artifacts in {directory}...") + + if not directory.is_dir(): + message = f"'{directory}' is not a directory." + + raise PipelineError(message) + + files = {entry.name: entry.read_bytes() for entry in sorted(directory.iterdir()) if entry.is_file()} + + if not files: + message = f"'{directory}' holds no files." + + raise PipelineError(message) + + context.logger.info(f"Collected {len(files)} file(s) ✔") + + metadata = _metadata_of(files) + reported = scan_id or _scan_id_of(metadata, directory) + + if reported is None: + message = ( + f"Nothing in '{directory}' names the scan it belongs to. Pass --scan_id, or keep the " + f"metadata.json the pipeline exported alongside the other artifacts." + ) + + raise PipelineError(message) + + return ReportSources( + scan_id=reported, + stored_scan_id=reported, + run_at=_run_at_of(metadata, directory), + artifacts=[], + files=files, + ) + + +def _metadata_of(files: dict[str, bytes]) -> dict[str, Any]: + """The pipeline metadata among the collected files, or nothing. + + Read here as well as in the renderer because the id and the date are wanted before a context is + built, and a folder without metadata still has to produce a report. + """ + for name in _METADATA: + content = files.get(name) + + if content is None: + continue + + try: + parsed = json.loads(content) + except ValueError: + return {} + + return parsed if isinstance(parsed, dict) else {} + + return {} + + +def _scan_id_of(metadata: dict[str, Any], directory: Path) -> UUID | None: + """Work out which run a folder of artifacts belongs to. + + The pipeline records where it worked and where it exported to, and both paths end in its own + id, so an export identifies itself. Where the metadata is gone, the folder may still be named + by the run -- which is how these are usually kept. + """ + for field in _ID_BEARING: + stated = metadata.get(field) + + if isinstance(stated, str): + found = _UUID.search(stated) + + if found: + return UUID(found.group(0)) + + found = _UUID.fullmatch(directory.resolve().name) + + return UUID(found.group(0)) if found else None + + +def _run_at_of(metadata: dict[str, Any], directory: Path) -> datetime: + """When the run happened. + + Its own timestamp where the metadata kept one; otherwise the folder's, which is when it was + exported rather than when it ran -- close enough to date a report by, and better than now. + """ + stated = metadata.get("pipeline_start_iso") + + if isinstance(stated, str): + try: + return datetime.fromisoformat(stated) + except ValueError: + pass + + return datetime.fromtimestamp(directory.stat().st_mtime, tz=UTC)