diff --git a/CHANGELOG.md b/CHANGELOG.md index 91b8bc6..73eed73 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,102 @@ schemas and the `paramify` CLI — not the internal code. ## [Unreleased] +### Added + +- `paramify programs` — a new command group over the Paramify workspace. + `programs list` shows each program's readable name next to its project UUID; + `programs target` selects programs (interactively, by name/id, or `--all`) and + writes them as fanout targets, filling in the shared config they need. The API + identifies programs by UUID while people know them by name; this closes that + gap without anyone copying a UUID by hand. The shared values (`--cert-uri`, + `--report-from`) are shown on every interactive run with what the manifest + holds today as the prompt default and a note of where it comes from: enter + keeps it and writes nothing, typing over it updates the category value. So the + same command adds a program and rolls the report window forward, and neither + requires opening the manifest to see what the next run will carry. Entries that + resolve to different values get no default — either one offered as *the* answer + would misreport the other. +- `program_name` — an optional target field on the Paramify VER fetchers. The + fetcher uses it for its evidence filename and the uploader for the artifact + title, so per-program artifacts read as `… - Alpha Cloud Services` rather than + a bare UUID. A UUID prefix stays in the filename because program names are not + guaranteed unique. + +### Changed + +- The `tui` extra pins `textual>=8,<9` (was `>=1.0,<2.0`). The old range was not + what anyone ran, and focus / `Input` behaviour differs enough across those lines + that the TUI is not the same app on 1.x. `tests/test_tui_keys.py` (new) drives + the real app through Textual's pilot to hold the key-and-focus contract: what + each tab focuses, that the globals survive a repeat tab press, and that enter + reaches an action wherever the footer says it does. +- **TUI**: the footer hint bar lists `esc` (the only way out of a focused text + field back to the shortcut keys — an `Input` consumes every printable key) and + the Run tab shows `enter/ctrl+r`, since focus opens on the ▶ Run button and + `enter` presses it. +- **Paramify VER fetchers**: `report_from` / `report_to` / `api_base_url` / + `http_timeout` moved out of `secrets[]`. Every declared secret is mandatory, so + declaring optional knobs there made them required, contradicting their + documented defaults. `cert_package_uri`, `api_base_url` and `http_timeout` are + now category config (`fetchers/_categories/paramify.yaml`) — one value per + workspace, set once under `platforms.paramify.config` instead of copied onto + every target. +- Each VER report's `_summary` now carries a `collection` block (status + the + API-failure ledger). `/issues` is the only call these fetchers make, so a + failure yields empty report arrays; without this a failed report was + indistinguishable from a genuinely clean one to anything reading the payload. +- The uploader prefers a target's `program_name` over its opaque id when titling + an artifact. Fetchers whose id is already readable are unaffected. +- **Every timestamp in a VER report is now emitted in one format** — UTC, second + precision, literal `Z` (`2026-07-30T09:00:00Z`). Values from the Paramify API + (`detectedAt`, `evaluationCompletedAt`, the `dueDate` quoted in an overdue + explanation) were previously passed through with the API's millisecond + precision, so a single document mixed notations; they are normalized on the way + in, and non-UTC offsets are converted rather than preserved. A `report_from` / + `report_to` given as a bare date is expanded, with a date-only end reported as + that day's last second (`2026-06-30` → `2026-06-30T23:59:59Z`) to match the + window actually collected. + +### Fixed + +- **TUI**: pressing the number of the tab you are already on no longer clears + focus. Assigning `TabbedContent.active` the value it already holds fires no + `TabActivated`, so nothing re-homed focus after it was cleared — and because a + page's bindings only resolve while focus is inside that page, every page + shortcut (`a`/`e`/`x`, `ctrl+r`, `j`/`k`, the arrows) silently went dead until + you pressed escape or a different tab. +- **TUI**: `ctrl+p` on the Paramify tab runs Preview instead of opening Textual's + command palette, which claims that key as a *priority* binding — checked ahead + of the focused widget, so the page's own binding could never fire. `p` now does + it too, mirroring the Manifest tab's preview key. +- **TUI**: `enter` does what the footer promises on the two tables where it did + nothing at all — on a run it drills into that run's evidence files (where enter + opens one), and on a manifest row it opens the entry editor. +- **TUI**: editing the manifest's output dir no longer loses the path. Textual + selects an `Input`'s value on focus, so the first keystroke replaced the whole + path; and an edit never submitted with `enter` was silently reverted by the next + rebuild. Focus no longer selects the value, and leaving the field commits it. +- **TUI**: `enter` in a confirmation dialog now means No. Yes is composed first, + so it took the default focus — on the dialogs that delete a manifest file, + remove an entry, and upload to Paramify. `y` still confirms. +- **TUI**: config set at the category level showed as unset on every entry that + inherited it — the manifest screen read only the entry's own `config` block and + had no notion of `platforms..config`. Both the detail pane and the + summary count now render `api.effective_config()`, the same merge the runner + performs, and show which layer each value came from. +- **Paramify VER fetchers**: a pending or rejected `RISK_ADJUSTMENT` no longer + reports `finalDisposition: "Partially Mitigated"` — mitigation now requires an + accepted deviation, not an unapproved request. +- An issue carrying neither `poamId` nor `id` no longer raises `KeyError` and + kills the whole report. +- `PARAMIFY_HTTP_TIMEOUT` is parsed at call time and falls back to the default on + a malformed value, instead of aborting the run with a bare `ValueError` at + import. +- A timestamped `report_to` no longer over-includes up to a day beyond the + declared reporting period. +- `PARAMIFY_REPORT_TO` is now declared, so it can actually be set through a + manifest (the runner passes only declared env vars). + ## [0.3.1-beta] - 2026-07-28 ### Changed diff --git a/README.md b/README.md index 4764f87..2fb17b0 100644 --- a/README.md +++ b/README.md @@ -162,9 +162,42 @@ paramify run # run it paramify runs # past runs under an output dir (newest first) paramify evidence # read one evidence file (normalizing the envelope) paramify upload [run-dir] # push a run's evidence to Paramify (default: latest run) +paramify programs # list workspace programs; turn them into targets paramify manifest # build/edit a manifest (see below) ``` +Fanning a fetcher out across the programs in a Paramify workspace is its own +step, because the API takes project UUIDs while people know their programs by +name. `paramify programs` closes that gap — list what's there, pick by name, and +it writes the targets for you: + +```bash +paramify programs list # name + UUID for every program +paramify programs target # choose interactively, then wire them up +paramify programs target --all \ + --cert-uri https://example.gov/cpo --report-from 2026-01-01 +``` + +With no fetcher argument it targets every manifest entry that takes a program, so +one command fans all of them out at once. Re-running it tops the manifest up +rather than duplicating targets. + +A target carries only what varies per program — `project_id` and its readable +`program_name`. Everything shared is written once to `platforms.paramify.config`: +the Certification Package Overview URI (not in Paramify's API, one value per +workspace) and the report period start. Every interactive run shows both with +whatever the manifest holds today as the prompt default — enter keeps it and +writes nothing, typing over it updates it — so `paramify programs target` is +equally how you add a program and how you roll the report window forward. +Passing `--cert-uri`/`--report-from` overwrites what's there without asking. + +`--report-from` is checked for an ISO date up front — an unparseable one produces +an empty report window, which drops every closed issue from the report without +failing. + +Both subcommands need `PARAMIFY_API_TOKEN` with read scope and accept `--json` +(under `--json` nothing prompts, so pass the flags). + > Back-compat: `python -m framework.runner ` and `python -m framework.tui` > still work and are exactly equivalent to the corresponding `paramify` > subcommands. diff --git a/docs/fetcher_contract.md b/docs/fetcher_contract.md index cd0fa16..d270302 100644 --- a/docs/fetcher_contract.md +++ b/docs/fetcher_contract.md @@ -110,9 +110,17 @@ paramify run # collect: enveloped JSON + _run_metadata.js paramify runs # past runs under the output dir (newest first) paramify evidence # read one evidence file (normalizing the envelope) paramify upload [run-dir] # push one run's evidence to Paramify (default: latest run) +paramify programs list # programs in the Paramify workspace: readable name + project UUID +paramify programs target [fetcher ...] # select programs by name and write them as fanout targets paramify manifest # build/edit a manifest file (init/new/add/remove/set-config/set-secret/add-target/remove-target/...) ``` +`paramify programs` is the only command that reads live workspace state (`GET +/projects`, needs `PARAMIFY_API_TOKEN`); it exists because the API identifies +programs by UUID while operators know them by name. `target` composes +`add_target` under the hood, so it produces exactly the manifest a hand-written +`manifest add-target` would. + Every `manifest` subcommand also accepts `--json`, emitting a stable `{ok, path, errors}` object so an agent can build a manifest step by step and read `errors` to see what's still missing. `list`/`validate` fail with a non-zero exit if any `fetcher.yaml` is schema-invalid. The envelope the runner produces is validated against `envelope_schema.json`, but a fetcher's *runtime* behavior (exit codes, output paths, etc.) is not yet automatically verified — that arrives with integration tests. diff --git a/docs/run_manifest_reference.md b/docs/run_manifest_reference.md index ee9fa71..e5e0fbc 100644 --- a/docs/run_manifest_reference.md +++ b/docs/run_manifest_reference.md @@ -241,6 +241,47 @@ the AI CLI; without it you get the human-readable rendering. Mutating commands return `{ok, path, errors}` under `--json` so a caller can confirm the write landed and surface any validation messages. +### Targets from a Paramify workspace + +For fetchers whose target is a Paramify program, the targets can be filled in +from the workspace instead of by hand — the API needs project UUIDs, which nobody +wants to copy: + +```bash +paramify programs list # readable name + UUID +paramify programs target # pick interactively, write targets +paramify programs target --all \ # every program + --cert-uri https://… --report-from 2026-01-01 # non-interactive / --json +``` + +With no fetcher argument it targets every manifest entry whose `target_schema` +declares `project_id`. Programs already targeted are skipped, so re-running tops +the manifest up instead of duplicating entries. + +A target gets only what varies per program: `project_id` and `program_name` (the +readable label — the fetcher uses it for its evidence filename, the uploader for +the artifact title). + +Anything the targeted fetchers need that *doesn't* vary per program is written to +`platforms..config`: `--cert-uri` (the Certification Package Overview +URI) and `--report-from` (the report period start). Each is shown on every +interactive run, with the value in force as the prompt default and a line saying +where it comes from (`platforms.paramify`, an entry's own config, or not set +yet); enter keeps it and leaves the manifest alone, typing over it updates the +category value. Entries that resolve to *different* values are reported as such +and no default is offered, since either one shown as the answer would misreport +the other. Supplying the flag skips the prompt and overwrites an existing value. +`--report-from` is validated as an ISO date before it's written. + +Needs `PARAMIFY_API_TOKEN` with read scope; under `--json` nothing prompts, so +pass `--program`/`--all` plus whichever shared values are still missing. + +The same split applies generally: values that vary per fanout iteration belong in +`targets[]`, and values shared across a category belong under `platforms..config` +— which the runner merges as *platform defaults ← platform values ← per-fetcher +values*, so a manifest can set **any** field a fetcher declares once at the +platform level, even one declared in the fetcher's own `config_schema`. + ### Build / edit a manifest The `manifest` subcommands read each fetcher's `fetcher.yaml` and write the diff --git a/fetchers/_categories/paramify.yaml b/fetchers/_categories/paramify.yaml new file mode 100644 index 0000000..e55fefc --- /dev/null +++ b/fetchers/_categories/paramify.yaml @@ -0,0 +1,49 @@ +# Category-level metadata for the Paramify FedRAMP report fetchers. +# +# Unlike most categories (which pull evidence FROM a third-party system INTO +# Paramify), these fetchers read FROM Paramify's own REST API and produce the +# FedRAMP Consolidated Rules 2026 vulnerability-reporting artifacts +# (VER-RPT-AVI, VER-RPT-VDT, VER-TFR-MRH). +# +# Access: Paramify REST API v0 with a Bearer token that has read scope on the +# target project's issues and deviations. + +name: paramify +description: >- + FedRAMP 20x vulnerability-reporting artifacts generated from Paramify issue + data (accepted vulnerabilities, vulnerability detail report, and the + historical VER-activity snapshot). + +# Config shared by every paramify fetcher. The runner injects these env vars for +# any fetcher with category: paramify. Set values once per run under manifest +# platforms.paramify.config. +config_schema: + cert_package_uri: + type: string + required: true + env: PARAMIFY_CERT_PACKAGE_URI + description: >- + Public Certification Package Overview URI, written into every VER report as + certificationPackageOverviewUri. One workspace publishes one such URI, so it + lives here rather than per fetcher or per program — set it once and all + three reports, across every program, carry it. + api_base_url: + type: string + default: https://app.paramify.com/api/v0 + env: PARAMIFY_API_BASE_URL + description: Paramify REST API base URL. Point at a non-production instance for testing. + http_timeout: + type: integer + default: 300 + env: PARAMIFY_HTTP_TIMEOUT + description: >- + Per-request timeout in seconds. The unfiltered /issues call is large + (~1.9 MB / ~75-120 s on a ~2k-issue project), hence the high default. + +auth: + # The fetchers read the token from PARAMIFY_API_TOKEN (falling back to + # PARAMIFY_UPLOAD_API_TOKEN). Base URL defaults to app.paramify.com and can + # be overridden per environment. + description: >- + Paramify REST API Bearer token with read scope on the project's issues and + deviations. See fetchers/paramify/README.md. diff --git a/fetchers/paramify/README.md b/fetchers/paramify/README.md new file mode 100644 index 0000000..183e3f2 --- /dev/null +++ b/fetchers/paramify/README.md @@ -0,0 +1,122 @@ +# Paramify FedRAMP VER Report Fetchers + +Unlike most categories (which pull evidence *from* a third-party system into +Paramify), these fetchers read *from* Paramify's own REST API and generate the +FedRAMP Consolidated Rules 2026 vulnerability-reporting artifacts: + +| Fetcher | Report | Evidence set | +|---|---|---| +| `paramify_accepted_vulnerabilities` | VER-RPT-AVI | `EVD-PARAMIFY-VER-RPT-AVI` | +| `paramify_vulnerability_detail_report` | VER-RPT-VDT | `EVD-PARAMIFY-VER-RPT-VDT` | +| `paramify_historical_ver_activity` | VER-TFR-MRH | `EVD-PARAMIFY-VER-TFR-MRH` | + +AVI and VDT are exact partition complements: every project issue is reported in +exactly one of them (accepted vs. not-accepted). MRH is a point-in-time snapshot +carrying both partitions in one document. All three share a single definition of +"accepted" and one issue-fetch/mapping implementation in +[`_shared/ver_common.py`](_shared/ver_common.py), so the reports cannot drift +apart. + +## Credentials + +A Paramify REST API Bearer token with **read** scope on the target project's +issues and deviations. It is the only `secret` these fetchers declare; everything +else is a `target` field or non-secret `config`. + +| Env var | Declared as | Required | Purpose | +|---|---|---|---| +| `PARAMIFY_API_TOKEN` | secret `api_token` | yes | Bearer token (falls back to `PARAMIFY_UPLOAD_API_TOKEN` when run standalone). | +| `PARAMIFY_PROJECT_ID` | **target** `project_id` | yes | Project UUID to scope the report. One target per program. | +| `PARAMIFY_PROGRAM_NAME` | **target** `program_name` | no | Readable program name; used for the evidence filename and artifact title. Falls back to the UUID. | +| `PARAMIFY_REPORT_FROM` | fetcher config `report_from` | yes | ISO start of the report period. | +| `PARAMIFY_REPORT_TO` | fetcher config `report_to` | no | ISO end; defaults to run time. | +| `PARAMIFY_CERT_PACKAGE_URI` | **category** config `cert_package_uri` | yes | Certification Package Overview URI written into every report. | +| `PARAMIFY_API_BASE_URL` | **category** config `api_base_url` | no | Defaults to `https://app.paramify.com/api/v0`. Point at stage for testing. | +| `PARAMIFY_HTTP_TIMEOUT` | **category** config `http_timeout` | no | Per-request timeout (seconds). Default 300 — the unfiltered `/issues` call is large. | + +Three layers, by what the value actually varies with: + +- **target** — differs per program, so it's per fanout iteration. +- **category config** (`fetchers/_categories/paramify.yaml`, set under + `platforms.paramify.config`) — one value for the whole workspace, shared by all + three fetchers. The package URI belongs here: one workspace publishes one, and + copying it onto every target would mean editing N×3 places to change it. +- **fetcher config** — the report period, which is a property of the report. + +Nothing non-secret is declared under `secrets[]`, deliberately: every declared +secret is **mandatory** (the runner raises when a manifest omits one), while +config is optional and defaultable. + +## Running across several programs + +All three fetchers fan out: one invocation per program, one evidence file per +program, all files landing in that report's single evidence set. Fill the targets +in from the workspace rather than by hand: + +```bash +paramify programs list # readable name + project UUID +paramify programs target # pick programs, get targets on all three fetchers +``` + +It asks for the Certification Package Overview URI and the report period start, +storing both as category config — no per-program bookkeeping. A later run shows +both again with the stored values as the defaults, so adding a program is enter, +enter, and moving the report window forward is typing a new date over the old one. + +`report_from` is declared per-fetcher (it's a property of the report, not the +platform) but set once at the platform level: the runner merges *platform +defaults ← platform values ← per-fetcher values*, so a manifest can set any +declared field once under `platforms.paramify.config`. Override it for a single +report by putting `report_from` in that fetcher entry's own `config`. + +Each program's file is named for its program (`..._Alpha_Cloud_Services_aaaaaaaa.json`, +UUID prefix appended because program names are not guaranteed unique), and the +uploader titles the artifact the same way. + +## Timestamps + +Every instant in a generated report uses one format — UTC, second precision, +literal `Z`: + +``` +2026-07-30T09:00:00Z +``` + +That holds regardless of source. Values the fetcher generates (`generatedAt`, +a defaulted `reportPeriod.to`) are produced in it; values from the Paramify API +(`detectedAt`, `evaluationCompletedAt`, the `dueDate` quoted in an overdue +explanation) are **normalized on the way in**, since the API returns +milliseconds; and a `report_from` / `report_to` supplied as a bare date is +expanded. A non-UTC offset is converted rather than preserved, so +`2026-02-01T09:00:00+02:00` is emitted as `2026-02-01T07:00:00Z`. + +A date-only `report_to` is reported as that day's **last second** +(`2026-06-30` → `2026-06-30T23:59:59Z`), because a date-only end means "through +the end of that day" to the coverage filter — reporting its midnight would +understate the period by a day. + +A value the parser can't read is passed through unchanged rather than dropped or +blanked; schema verification is the right place for a malformed source value to +surface. `tests/test_ver_timestamps.py` pins all of this. + +## Notes + +- **Coverage:** the fetchers keep every OPEN issue regardless of when its status + last changed, plus anything whose status changed inside the report window. + This avoids silently dropping open issues with a missing/epoch `statusDate`. +- **Epoch sentinel:** issues with a missing or pre-2000 (`1970-…`) + `evaluationDate` are treated as never-evaluated — they are not time-accepted + (the VER-TFR-MAV 192-day clock never started) and are surfaced in a + VER-TFR-EVU warning to stderr. +- **`_summary`:** each report carries a top-level `_summary` object (count + breakdowns computed from the report's own arrays). It is a vendor extension — + the FedRAMP report arrays remain the source of truth. +- **`_summary.collection`:** records the collection outcome and the API-failure + ledger *inside* the payload. `/issues` is the only call these fetchers make, so + a failure leaves the report arrays empty; without this block an empty **failed** + report would look identical to a genuinely clean one to anything reading the + payload alone (the uploader's `skip_failed` defaults to false, so failed + evidence is uploaded unless configured otherwise). A failed collection still + exits non-zero. +- **Milestones** are read from the `milestones` array embedded in the `/issues` + response; there are no per-issue milestone calls. diff --git a/fetchers/paramify/_shared/ver_common.py b/fetchers/paramify/_shared/ver_common.py new file mode 100644 index 0000000..b39f73a --- /dev/null +++ b/fetchers/paramify/_shared/ver_common.py @@ -0,0 +1,495 @@ +""" +Shared logic for the Paramify FedRAMP VER-* report fetchers. + +One source of truth for the three reports (VER-RPT-AVI, VER-RPT-VDT, +VER-TFR-MRH): the "accepted vulnerability" definition, the Paramify /issues +fetch + coverage rule, the epoch/sentinel evaluation-date handling, the +VDT field mapping (disposition, overdue, rating), the acceptance rationale, +the VER-TFR-EVU backlog warning, and the per-report _summary builders. + +Consolidating here means the AVI/VDT partition can never drift: all three +fetchers import the SAME is_accepted() and map_vulnerability_detail(), so a +change is made once and applies everywhere. + +Env reads (interim v0.x: fetchers read env directly; the runner sets these): + PARAMIFY_API_TOKEN (falls back to PARAMIFY_UPLOAD_API_TOKEN) + PARAMIFY_PROJECT_ID + PARAMIFY_CERT_PACKAGE_URI + PARAMIFY_REPORT_FROM + PARAMIFY_REPORT_TO (optional; defaults to run time) + PARAMIFY_API_BASE_URL (optional; defaults to app.paramify.com/api/v0) + PARAMIFY_HTTP_TIMEOUT (optional; default 300s) +""" + +import logging +import os +import re +from collections import Counter +from datetime import datetime, timedelta, timezone +from typing import Any, Dict, List, Optional, Tuple + +import requests + +logger = logging.getLogger("paramify_ver_common") + +# --- Shared "accepted" definition (AVI and VDT must agree exactly) ---------- +ACCEPTED_DEVIATION_TYPES = ( + "OPERATIONAL_REQUIREMENT", + "VENDOR_DEPENDENCY", + "RISK_ADJUSTMENT", +) +ACCEPTED_STATUS = "ACCEPTED" +ACCEPTANCE_DAYS = 192 # VER-TFR-MAV +OPEN_ISSUE_STATUSES = ("OPEN",) +CLOSED_ISSUE_STATUSES = ("CLOSED",) + +# Potential Agency Impact N-rating. INTERIM positional mapping (confirmed with +# the FedRAMP package owner). Absent level => no rating emitted. +LEVEL_TO_NRATING = {"CHILL": 1, "LOW": 2, "MODERATE": 3, "HIGH": 4, "CRITICAL": 5} + +DISPOSITION_FULLY = "Fully Mitigated" +DISPOSITION_PARTIALLY = "Partially Mitigated" +DISPOSITION_FALSE_POSITIVE = "False Positive" + +# Paramify records some issues with a Unix-epoch evaluationDate +# ("1970-01-01T00:00:00.000Z"). An epoch (or otherwise implausibly ancient) +# timestamp is a missing-data sentinel, not a real evaluation event. Any date +# before this floor is treated as "no evaluation recorded". +MIN_PLAUSIBLE_EVALUATION = datetime(2000, 1, 1, tzinfo=timezone.utc) + +# HTTP timeout (seconds) for Paramify API calls; override with +# PARAMIFY_HTTP_TIMEOUT. The unfiltered /issues call is large (~1.9 MB / +# ~75-120 s on a ~2k-issue project), so the shipped default failsafe is 300s. +DEFAULT_HTTP_TIMEOUT = 300 + + +def http_timeout() -> int: + """Per-request timeout, resolved at call time. + + Read lazily (not as an import-time constant) so a malformed value degrades to + the default with a warning instead of aborting the run with a bare int() + ValueError before main() can log anything useful. + """ + raw = os.environ.get("PARAMIFY_HTTP_TIMEOUT", "").strip() + if not raw: + return DEFAULT_HTTP_TIMEOUT + try: + return int(raw) + except ValueError: + logger.warning( + "PARAMIFY_HTTP_TIMEOUT=%r is not an integer; using the %ds default", + raw, DEFAULT_HTTP_TIMEOUT, + ) + return DEFAULT_HTTP_TIMEOUT + + +# --- Environment / API ------------------------------------------------------ +# The single timestamp format every value in these reports is emitted in: +# UTC, second precision, literal Z ("2026-07-30T09:00:00Z"). RFC 3339, and the +# same shape the runner stamps into envelope metadata. +TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%SZ" + + +def current_timestamp() -> str: + return datetime.now(timezone.utc).strftime(TIMESTAMP_FORMAT) + + +def get_env(name: str) -> str: + value = os.environ.get(name, "") + if not value: + raise RuntimeError(f"Missing required env var: {name}") + return value + + +def resolve_common_env() -> Dict[str, str]: + """Resolve the env every VER fetcher needs. Token falls back to the upload + token name. Raises RuntimeError naming the first missing required var.""" + token = os.environ.get("PARAMIFY_API_TOKEN") or os.environ.get("PARAMIFY_UPLOAD_API_TOKEN") + if not token: + raise RuntimeError("Missing required env var: PARAMIFY_API_TOKEN (or PARAMIFY_UPLOAD_API_TOKEN)") + now = current_timestamp() + return { + "token": token, + "base_url": os.environ.get("PARAMIFY_API_BASE_URL", "https://app.paramify.com/api/v0"), + "project_id": get_env("PARAMIFY_PROJECT_ID"), + "cert_package_uri": get_env("PARAMIFY_CERT_PACKAGE_URI"), + "report_from": get_env("PARAMIFY_REPORT_FROM"), + "report_to": os.environ.get("PARAMIFY_REPORT_TO") or now, + "generated_at": now, + # Optional readable label for this program; only used for the filename. + "program_name": os.environ.get("PARAMIFY_PROGRAM_NAME", ""), + } + + +def sanitize_for_filename(value: str) -> str: + """Make a target identifier safe for a filename (mirrors the gitlab fetcher). + + Fanout writes one file per program; the runner discovers outputs by diffing + the evidence dir, so each invocation MUST write a distinct name or the second + program silently overwrites the first and its outputs list comes back empty. + """ + return re.sub(r"[^a-zA-Z0-9_-]", "_", str(value)) + + +def target_slug(env: Dict[str, str]) -> str: + """Filename discriminator for this target: the readable program name when the + manifest supplied one, else the project UUID. + + Program names are not guaranteed unique in a workspace, so two same-named + programs would collide on one filename -- and the runner's dir-diff output + discovery would report the second invocation as having produced nothing. The + UUID tail keeps every name distinct while staying readable. + """ + name = (env.get("program_name") or "").strip() + if not name: + return sanitize_for_filename(env["project_id"]) + return f"{sanitize_for_filename(name)}_{sanitize_for_filename(env['project_id'])[:8]}" + + +def paramify_get(base_url: str, token: str, path: str, params: Dict[str, Any]) -> Any: + url = f"{base_url.rstrip('/')}{path}" + resp = requests.get( + url, + headers={"Accept": "application/json", "Authorization": f"Bearer {token}"}, + params=params, + timeout=http_timeout(), + ) + resp.raise_for_status() + return resp.json() + + +def _window_bounds(status_start: str, status_end: str) -> Tuple[Optional[datetime], Optional[datetime]]: + """Half-open [start, end) report window from two ISO strings. + + A date-only upper bound ("2026-06-30") means "through the end of that day", + so it is pushed to the following midnight. A timestamped bound is used as + given -- callers must NOT pre-truncate to 10 chars, or a timestamped + report_to silently over-includes up to a day past the declared period. + """ + start = _parse_iso(status_start) + end = _parse_iso(status_end) + if end is not None and len(status_end.strip()) == 10: + end = end + timedelta(days=1) + return start, end + + +def fetch_all_issues( + base_url: str, + token: str, + project_id: str, + status_start: str, + status_end: str, + api_failures: List[Dict[str, Any]], +) -> List[Dict]: + """Fetch every issue in the project, then keep those that are OPEN (an open, + unresolved vulnerability is ongoing activity regardless of when its status + last changed) OR whose statusDate falls in the report window (captures + closures/changes in the period). + + The /issues API has no status filter, and filtering the query by statusDate + silently excluded open issues whose statusDate is missing or an epoch + sentinel. Fetching by projectId alone and filtering in code closes that gap. + Pagination is not documented on this endpoint; extend here if large projects + turn out to paginate.""" + try: + payload = paramify_get(base_url, token, "/issues", {"projectId": project_id}) + except requests.exceptions.RequestException as e: + api_failures.append({"query": "all_issues", "type": type(e).__name__, "message": str(e)}) + return [] + issues = payload.get("issues", []) if isinstance(payload, dict) else [] + + start, end = _window_bounds(status_start, status_end) + + def in_window(issue: Dict) -> bool: + sd = _parse_iso(issue.get("statusDate")) + if sd is None or start is None or end is None: + return False + return start <= sd < end + + return [i for i in issues if i.get("status") in OPEN_ISSUE_STATUSES or in_window(i)] + + +# --- Date handling ---------------------------------------------------------- +def _parse_iso(value: str) -> Optional[datetime]: + if not value: + return None + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed + + +def to_utc_z(value: Optional[str]) -> Optional[str]: + """Normalize a timestamp to the one format these reports emit: UTC, second + precision, literal Z -- "2026-07-30T09:00:00Z". + + Every instant in a report goes through here. Paramify returns milliseconds + ("2026-02-01T00:00:00.000Z") and config may supply a bare date, so passing + values straight through produced a document mixing three notations. Offsets + are converted to UTC rather than preserved, so "…T09:00:00+02:00" emits as + "…T07:00:00Z". + + Unparseable input is returned unchanged: dropping or blanking a value the + source gave us is worse than an off-format one, and schema verification is + the right place for that to surface. + """ + if not value: + return value + parsed = _parse_iso(value) + if parsed is None: + return value + return parsed.astimezone(timezone.utc).strftime(TIMESTAMP_FORMAT) + + +def report_period_bounds(report_from: str, report_to: str) -> Tuple[str, str]: + """The reportPeriod as declared in a report: both ends in to_utc_z() form. + + A date-only bound is reported as that day's last second rather than its + midnight, because a date-only end means "through the end of that day" to the + coverage filter (see _window_bounds). Emitting "2026-06-30T00:00:00Z" for a + window that collected all of June 30 would understate the period in a + compliance artifact by a day. + """ + end_day = _parse_iso(report_to) if report_to and len(report_to.strip()) == 10 else None + end = ( + (end_day + timedelta(days=1, seconds=-1)).strftime(TIMESTAMP_FORMAT) + if end_day is not None else (to_utc_z(report_to) or report_to) + ) + return to_utc_z(report_from) or report_from, end + + +def effective_evaluation_date(issue: Dict) -> Optional[datetime]: + """Real completed-evaluation date, or None when missing, unparseable, or a + pre-2000 sentinel (e.g. Unix epoch).""" + evaluated = _parse_iso(issue.get("evaluationDate")) + if evaluated is None or evaluated < MIN_PLAUSIBLE_EVALUATION: + return None + return evaluated + + +# --- Accepted-vulnerability test (shared by AVI + VDT) ---------------------- +def _accepted_deviations(issue: Dict, types: Tuple[str, ...]) -> List[Dict]: + """Deviations of the given types that have actually been ACCEPTED. A pending + or rejected deviation is a request, not a decision.""" + return [ + d for d in issue.get("deviations", []) + if d.get("type") in types + and (d.get("deviationMetadata") or {}).get("status") == ACCEPTED_STATUS + ] + + +def accepted_deviation(issue: Dict) -> Optional[Dict]: + """The most recently accepted qualifying deviation, or None.""" + return max( + _accepted_deviations(issue, ACCEPTED_DEVIATION_TYPES), + key=lambda d: (d.get("deviationMetadata") or {}).get("acceptanceStatusDate") or "", + default=None, + ) + + +def is_192_day_accepted(issue: Dict, now: Optional[datetime] = None) -> bool: + """VER-TFR-MAV: open AND evaluated 192+ days ago. Missing/sentinel evaluation + dates mean no evaluation happened, so the clock has not started.""" + if issue.get("status") not in OPEN_ISSUE_STATUSES: + return False + evaluated = effective_evaluation_date(issue) + if evaluated is None: + return False + now = now or datetime.now(timezone.utc) + return (now - evaluated).days >= ACCEPTANCE_DAYS + + +def is_accepted(issue: Dict) -> bool: + """Accepted deviation OR 192-day-open. The single partition test.""" + return accepted_deviation(issue) is not None or is_192_day_accepted(issue) + + +def acceptance_rationale(issue: Dict) -> str: + """Rationale text from the qualifying accepted deviation, or a default. + + Shared by AVI and MRH: both wrap an accepted issue in the same + {vulnerabilityDetail, acceptanceRationale} object, so the text must match. + """ + dev = accepted_deviation(issue) + if dev and dev.get("description"): + return dev["description"] + if is_192_day_accepted(issue): + return "Open beyond the VER-TFR-MAV 192-day threshold without full mitigation." + return "Accepted vulnerability." + + +# --- VDT field derivations -------------------------------------------------- +def _final_disposition(issue: Dict) -> Optional[str]: + """False Positive (accepted FP deviation) > Fully Mitigated (closed) > + Partially Mitigated (open with accepted risk-adjustment or milestone) > omit. + Milestones are read from the `milestones` array embedded in the /issues + response -- no per-issue calls.""" + if _accepted_deviations(issue, ("FALSE_POSITIVE",)): + return DISPOSITION_FALSE_POSITIVE + if issue.get("status") in CLOSED_ISSUE_STATUSES: + return DISPOSITION_FULLY + if issue.get("status") in OPEN_ISSUE_STATUSES and ( + _accepted_deviations(issue, ("RISK_ADJUSTMENT",)) or issue.get("milestones") + ): + return DISPOSITION_PARTIALLY + return None + + +def _overdue_status(issue: Dict, now: Optional[datetime] = None) -> Dict: + """INTERIM: open past dueDate => overdue (explanation required by schema).""" + if issue.get("status") not in OPEN_ISSUE_STATUSES: + return {"isOverdue": False} + due = _parse_iso(issue.get("dueDate")) + if due is None: + return {"isOverdue": False} + now = now or datetime.now(timezone.utc) + if now > due: + return { + "isOverdue": True, + "explanation": ( + f"Open past its remediation due date " + f"({due.astimezone(timezone.utc).strftime(TIMESTAMP_FORMAT)}); " + "not yet fully mitigated or remediated." + ), + } + return {"isOverdue": False} + + +def map_vulnerability_detail(issue: Dict) -> Dict: + """Build one FedRAMP vulnerabilityDetail object (used by VDT + MRH active, + and wrapped for AVI/MRH accepted).""" + origin = issue.get("origin") or {} + detail: Dict[str, Any] = { + # An issue carrying neither identifier is a source-data defect; emit it + # empty so schema verification flags the record, rather than raising a + # KeyError that kills the whole report. + "providerTrackingId": issue.get("poamId") or issue.get("id") or "", + "detection": { + # Normalized, not passed through: the API returns milliseconds. + "detectedAt": to_utc_z(issue.get("createdAt")), + "detectionSource": origin.get("name") or "Unspecified", + }, + "vulnerabilityDescription": issue.get("description") or issue.get("title") or "", + } + if issue.get("internetReachableVulnerability") is not None: + detail["isInternetReachable"] = issue["internetReachableVulnerability"] + if issue.get("likelyExploitableVulnerability") is not None: + detail["isLikelyExploitable"] = issue["likelyExploitableVulnerability"] + evaluated = effective_evaluation_date(issue) + if evaluated is not None: + detail["evaluationCompletedAt"] = evaluated.astimezone(timezone.utc).strftime(TIMESTAMP_FORMAT) + rating = LEVEL_TO_NRATING.get(issue.get("level")) + if rating is not None: + detail["currentRating"] = rating + detail["overdueStatus"] = _overdue_status(issue) + disposition = _final_disposition(issue) + if disposition is not None: + detail["finalDisposition"] = disposition + return detail + + +# --- Shared reporting helpers ----------------------------------------------- +def warn_unevaluated_backlog( + issues: List[Dict], log: logging.Logger, consequence: str +) -> List[Dict]: + """Log the VER-TFR-EVU open-but-never-evaluated backlog; return it. + + Open, not deviation-accepted, and with no plausible evaluationDate: the + VER-TFR-MAV 192-day clock never started for these, so each report states + what it did with them via `consequence`. + """ + unevaluated = [ + i for i in issues + if i.get("status") in OPEN_ISSUE_STATUSES + and effective_evaluation_date(i) is None + and accepted_deviation(i) is None + ] + if unevaluated: + log.warning( + "%d open issue(s) have no real completed-evaluation date " + "(missing or epoch sentinel); %s " + "(VER-TFR-EVU: evaluate within 5 days of detection).", + len(unevaluated), consequence, + ) + return unevaluated + + +def build_collection_status(api_failures: List[Dict[str, Any]]) -> Dict[str, Any]: + """Collection outcome, carried inside the report's own _summary. + + House pattern (20+ fetchers): a failed collection still writes its evidence + file with the failure ledger inside it. That matters twice over here -- a + dropped /issues call yields EMPTY report arrays, and the uploader's + skip_failed defaults to false, so without this block an empty failed report + is indistinguishable from a genuinely clean one to anything reading the + payload alone. + """ + return { + "status": "failed" if api_failures else "success", + "apiFailures": api_failures, + } + + +# --- _summary builders (vendor extension carried in the payload) ------------ +DISPOSITION_IN_PROGRESS = "In Progress" + + +def _detail_counts(details: List[Dict]) -> Dict[str, Any]: + """Disposition / overdue / unevaluated tallies over mapped vulnerabilityDetails. + + Shared by VDT and MRH, which count identically and differ only in the key + names they file the result under. Keyed off the DISPOSITION_* constants the + emitter uses, so renaming a label can't leave the summaries reporting zeros. + """ + disp = Counter(v.get("finalDisposition", DISPOSITION_IN_PROGRESS) for v in details) + return { + "dispositions": { + "fullyMitigated": disp[DISPOSITION_FULLY], + "partiallyMitigated": disp[DISPOSITION_PARTIALLY], + "falsePositive": disp[DISPOSITION_FALSE_POSITIVE], + "inProgress": disp[DISPOSITION_IN_PROGRESS], + }, + "overdue": sum(1 for v in details if (v.get("overdueStatus") or {}).get("isOverdue") is True), + "withoutCompletedEvaluation": sum(1 for v in details if "evaluationCompletedAt" not in v), + } + + +def build_vdt_summary(vulns: List[Dict], report_from: str, report_to: str) -> Dict: + counts = _detail_counts(vulns) + return { + "report": "VER-RPT-VDT", + "reportPeriod": {"from": report_from, "to": report_to}, + "nonAcceptedVulnerabilities": len(vulns), + "dispositions": counts["dispositions"], + "overdue": counts["overdue"], + "notOverdue": len(vulns) - counts["overdue"], + "withoutCompletedEvaluation": counts["withoutCompletedEvaluation"], + } + + +def build_avi_summary(accepted: List[Dict], report_from: str, report_to: str) -> Dict: + with_eval = sum(1 for a in accepted if a["vulnerabilityDetail"].get("evaluationCompletedAt")) + return { + "report": "VER-RPT-AVI", + "reportPeriod": {"from": report_from, "to": report_to}, + "acceptedVulnerabilities": len(accepted), + "withCompletedEvaluation": with_eval, + "withoutCompletedEvaluation": len(accepted) - with_eval, + } + + +def build_mrh_summary(active: List[Dict], accepted: List[Dict], generated_at: str) -> Dict: + counts = _detail_counts(active) + return { + "report": "VER-TFR-MRH", + "generatedAt": generated_at, + "totalVulnerabilities": len(active) + len(accepted), + "active": len(active), + "accepted": len(accepted), + "activeDispositions": counts["dispositions"], + "activeOverdue": counts["overdue"], + "activeWithoutCompletedEvaluation": counts["withoutCompletedEvaluation"], + } diff --git a/fetchers/paramify/accepted_vulnerabilities/fetcher.py b/fetchers/paramify/accepted_vulnerabilities/fetcher.py new file mode 100644 index 0000000..6a86f10 --- /dev/null +++ b/fetchers/paramify/accepted_vulnerabilities/fetcher.py @@ -0,0 +1,111 @@ +#!/usr/bin/env python3 +""" +VER-RPT-AVI: Paramify Accepted Vulnerability Info + +Generates the FedRAMP 20x Accepted Vulnerability Info report from Paramify +issues. An issue is an accepted vulnerability if it has an accepted deviation +(OPERATIONAL_REQUIREMENT / VENDOR_DEPENDENCY / RISK_ADJUSTMENT) or is open 192+ +days past a real completed evaluation (VER-TFR-MAV). Issues with a missing or +epoch-sentinel evaluation date are NOT time-accepted (the 192-day clock never +started) and are surfaced as an unevaluated-backlog warning (VER-TFR-EVU). + +Output: $EVIDENCE_DIR/paramify_accepted_vulnerabilities.json +Env: PARAMIFY_API_TOKEN (or PARAMIFY_UPLOAD_API_TOKEN), PARAMIFY_PROJECT_ID, + PARAMIFY_CERT_PACKAGE_URI, PARAMIFY_REPORT_FROM, PARAMIFY_REPORT_TO (opt), + PARAMIFY_API_BASE_URL (opt), PARAMIFY_HTTP_TIMEOUT (opt). +""" + +import json +import logging +import os +import sys +from pathlib import Path + +from dotenv import load_dotenv + +SCRIPT_DIR = Path(__file__).resolve().parent +sys.path.insert(0, str(SCRIPT_DIR.parent / "_shared")) +import ver_common as vc # noqa: E402 + +logger = logging.getLogger("paramify_accepted_vulnerabilities") + + +def build_report(issues, cert_package_uri, report_from, report_to): + accepted = [ + { + "vulnerabilityDetail": vc.map_vulnerability_detail(i), + "acceptanceRationale": vc.acceptance_rationale(i), + } + for i in issues if vc.is_accepted(i) + ] + return { + "certificationPackageOverviewUri": cert_package_uri, + "reportPeriod": {"from": report_from, "to": report_to}, + "acceptedVulnerabilities": accepted, + } + + +def main() -> int: + logging.basicConfig( + level=os.environ.get("LOG_LEVEL", "INFO"), + format="%(asctime)s %(levelname)s %(name)s %(message)s", + ) + load_dotenv() # interim v0.x: fetcher loads .env itself + + env = vc.resolve_common_env() + output_dir = Path(os.environ.get("EVIDENCE_DIR", "./evidence")) + output_dir.mkdir(parents=True, exist_ok=True) + + api_failures = [] + issues = vc.fetch_all_issues( + env["base_url"], env["token"], env["project_id"], + env["report_from"], env["report_to"], api_failures, + ) + + # Visibility: open issues with no real completed evaluation (VER-TFR-EVU). + vc.warn_unevaluated_backlog( + issues, logger, + "excluded from VER-TFR-MAV time-based acceptance", + ) + + # Declared period in the report's own timestamp format. The RAW env values + # still drive fetch_all_issues above -- normalizing before the window is + # computed would turn a date-only bound into a midnight instant and silently + # drop that day's closures. + period_from, period_to = vc.report_period_bounds(env["report_from"], env["report_to"]) + + report = build_report( + issues, env["cert_package_uri"], period_from, period_to + ) + report["_summary"] = vc.build_avi_summary( + report["acceptedVulnerabilities"], period_from, period_to + ) + report["_summary"]["collection"] = vc.build_collection_status(api_failures) + + output_path = output_dir / f"paramify_accepted_vulnerabilities_{vc.target_slug(env)}.json" + with open(output_path, "w") as f: + json.dump(report, f, indent=2) + + s = report["_summary"] + logger.info( + "Evidence saved to %s (%d accepted; %d with eval date, %d without)", + output_path, s["acceptedVulnerabilities"], + s["withCompletedEvaluation"], s["withoutCompletedEvaluation"], + ) + + # Exit non-zero if collection encountered API failures (repo convention). + # /issues is the only call, so any failure means the report arrays above are + # empty for want of data -- NOT because the program has no accepted + # vulnerabilities. _summary.collection records that inside the payload. + if api_failures: + logger.error( + "%d API failure(s) during collection; the report is incomplete and " + "its counts must not be read as a clean result", + len(api_failures), + ) + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/fetchers/paramify/accepted_vulnerabilities/fetcher.yaml b/fetchers/paramify/accepted_vulnerabilities/fetcher.yaml new file mode 100644 index 0000000..0d122b6 --- /dev/null +++ b/fetchers/paramify/accepted_vulnerabilities/fetcher.yaml @@ -0,0 +1,69 @@ +name: paramify_accepted_vulnerabilities +version: 0.1.0 +description: >- + Generates the FedRAMP 20x Accepted Vulnerability Info (VER-RPT-AVI) report + from Paramify issues: vulnerabilities with an accepted deviation + (OPERATIONAL_REQUIREMENT / VENDOR_DEPENDENCY / RISK_ADJUSTMENT) or that are + open 192+ days past a real completed evaluation (VER-TFR-MAV). +category: paramify + +supports_targets: true + +target_schema: + project_id: + type: string + required: true + env: PARAMIFY_PROJECT_ID + description: Paramify program (project) UUID. One target per program; from GET /projects. + program_name: + type: string + required: false + env: PARAMIFY_PROGRAM_NAME + description: >- + Human-readable program name from the Paramify workspace. Used for the + evidence filename and the uploaded artifact title; falls back to the + project UUID when absent. `paramify programs target` fills this in. + +runtime: + type: python + entry: fetcher.py + timeout: 600 + +output: + type: json + path: paramify_accepted_vulnerabilities.json + aggregation: per_target + +secrets: + - name: api_token + env: PARAMIFY_API_TOKEN + +# Report-period knobs. Declared here (not under secrets) because every entry in +# secrets[] is mandatory -- the runner raises when a manifest omits one -- while +# config is optional and defaultable. Platform-wide knobs (cert_package_uri, +# api_base_url, http_timeout) live in fetchers/_categories/paramify.yaml; a +# manifest can still set any of these once under platforms.paramify.config. +config_schema: + report_from: + type: string + required: true + env: PARAMIFY_REPORT_FROM + description: ISO start of the report period, date ("2026-01-01") or timestamp. + report_to: + type: string + required: false + env: PARAMIFY_REPORT_TO + description: ISO end of the report period. Defaults to run time when omitted. + +evidence_set: + reference_id: EVD-PARAMIFY-VER-RPT-AVI + name: Paramify Accepted Vulnerability Info (VER-RPT-AVI) + instructions: >- + Script: fetcher.py. Reads issues and deviations for PARAMIFY_PROJECT_ID from + the Paramify REST API (GET /issues), applies the shared accepted-vulnerability + definition (accepted deviation or 192-day-open per VER-TFR-MAV), and writes + the FedRAMP Accepted Vulnerability Info JSON. + +ksis: + - KSI-MLA-06 + - KSI-MLA-03 diff --git a/fetchers/paramify/historical_ver_activity/fetcher.py b/fetchers/paramify/historical_ver_activity/fetcher.py new file mode 100644 index 0000000..d6dcb13 --- /dev/null +++ b/fetchers/paramify/historical_ver_activity/fetcher.py @@ -0,0 +1,108 @@ +#!/usr/bin/env python3 +""" +VER-TFR-MRH: Paramify Historical VER Activity (snapshot) + +Point-in-time snapshot containing BOTH partitions in one document: + activeVulnerabilities -- all non-accepted vulnerabilities (VER-RPT-VDT fields) + acceptedVulnerabilities -- all accepted vulnerabilities (VER-RPT-AVI fields) + +Contains no acceptance logic of its own: it partitions a SINGLE issue fetch +using the shared accepted definition in _shared/ver_common.py, so the two arrays +are consistent by construction (same issue set, same instant) and can never +disagree with the individually generated AVI/VDT reports. + +Output: $EVIDENCE_DIR/paramify_historical_ver_activity.json +Env: PARAMIFY_API_TOKEN (or PARAMIFY_UPLOAD_API_TOKEN), PARAMIFY_PROJECT_ID, + PARAMIFY_CERT_PACKAGE_URI, PARAMIFY_REPORT_FROM, PARAMIFY_REPORT_TO (opt), + PARAMIFY_API_BASE_URL (opt), PARAMIFY_HTTP_TIMEOUT (opt). +""" + +import json +import logging +import os +import sys +from pathlib import Path + +from dotenv import load_dotenv + +SCRIPT_DIR = Path(__file__).resolve().parent +sys.path.insert(0, str(SCRIPT_DIR.parent / "_shared")) +import ver_common as vc # noqa: E402 + +logger = logging.getLogger("paramify_historical_ver_activity") + + +def build_report(issues, cert_package_uri, generated_at): + active, accepted = [], [] + for issue in issues: + if vc.is_accepted(issue): + accepted.append({ + "vulnerabilityDetail": vc.map_vulnerability_detail(issue), + "acceptanceRationale": vc.acceptance_rationale(issue), + }) + else: + active.append(vc.map_vulnerability_detail(issue)) + return { + "certificationPackageOverviewUri": cert_package_uri, + "generatedAt": generated_at, + "activeVulnerabilities": active, + "acceptedVulnerabilities": accepted, + } + + +def main() -> int: + logging.basicConfig( + level=os.environ.get("LOG_LEVEL", "INFO"), + format="%(asctime)s %(levelname)s %(name)s %(message)s", + ) + load_dotenv() + + env = vc.resolve_common_env() + output_dir = Path(os.environ.get("EVIDENCE_DIR", "./evidence")) + output_dir.mkdir(parents=True, exist_ok=True) + + api_failures = [] + issues = vc.fetch_all_issues( + env["base_url"], env["token"], env["project_id"], + env["report_from"], env["report_to"], api_failures, + ) + + vc.warn_unevaluated_backlog( + issues, logger, + "reported as active without evaluationCompletedAt", + ) + + report = build_report(issues, env["cert_package_uri"], env["generated_at"]) + report["_summary"] = vc.build_mrh_summary( + report["activeVulnerabilities"], report["acceptedVulnerabilities"], + env["generated_at"], + ) + report["_summary"]["collection"] = vc.build_collection_status(api_failures) + + output_path = output_dir / f"paramify_historical_ver_activity_{vc.target_slug(env)}.json" + with open(output_path, "w") as f: + json.dump(report, f, indent=2) + + s = report["_summary"] + logger.info( + "Evidence saved to %s (%d total: %d active, %d accepted; " + "active overdue=%d, without-eval=%d)", + output_path, s["totalVulnerabilities"], s["active"], s["accepted"], + s["activeOverdue"], s["activeWithoutCompletedEvaluation"], + ) + + # /issues is the only call, so any failure means both arrays above are empty + # for want of data -- NOT because the program has no vulnerabilities. + # _summary.collection records that inside the payload. + if api_failures: + logger.error( + "%d API failure(s) during collection; the snapshot is incomplete and " + "its counts must not be read as a clean result", + len(api_failures), + ) + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/fetchers/paramify/historical_ver_activity/fetcher.yaml b/fetchers/paramify/historical_ver_activity/fetcher.yaml new file mode 100644 index 0000000..81a426f --- /dev/null +++ b/fetchers/paramify/historical_ver_activity/fetcher.yaml @@ -0,0 +1,74 @@ +name: paramify_historical_ver_activity +version: 0.1.0 +description: >- + Generates the FedRAMP 20x Historical VER Activity (VER-TFR-MRH) snapshot from + Paramify issues: a point-in-time document containing both active (non-accepted) + and accepted vulnerabilities, partitioned by the shared accepted definition so + the two arrays are consistent by construction. +category: paramify + +supports_targets: true + +target_schema: + project_id: + type: string + required: true + env: PARAMIFY_PROJECT_ID + description: Paramify program (project) UUID. One target per program; from GET /projects. + program_name: + type: string + required: false + env: PARAMIFY_PROGRAM_NAME + description: >- + Human-readable program name from the Paramify workspace. Used for the + evidence filename and the uploaded artifact title; falls back to the + project UUID when absent. `paramify programs target` fills this in. + +runtime: + type: python + entry: fetcher.py + timeout: 600 + +output: + type: json + path: paramify_historical_ver_activity.json + aggregation: per_target + +secrets: + - name: api_token + env: PARAMIFY_API_TOKEN + +# Report-period knobs. Declared here (not under secrets) because every entry in +# secrets[] is mandatory -- the runner raises when a manifest omits one -- while +# config is optional and defaultable. Platform-wide knobs (cert_package_uri, +# api_base_url, http_timeout) live in fetchers/_categories/paramify.yaml; a +# manifest can still set any of these once under platforms.paramify.config. +config_schema: + report_from: + type: string + required: true + env: PARAMIFY_REPORT_FROM + description: >- + ISO start of the issue-coverage window, date ("2026-01-01") or timestamp. + The snapshot itself is point-in-time; this bounds which non-open issues it + carries. + report_to: + type: string + required: false + env: PARAMIFY_REPORT_TO + description: ISO end of the coverage window. Defaults to run time when omitted. + +evidence_set: + reference_id: EVD-PARAMIFY-VER-TFR-MRH + name: Paramify Historical VER Activity (VER-TFR-MRH) + instructions: >- + Script: fetcher.py. Reads all project issues for PARAMIFY_PROJECT_ID from the + Paramify REST API (GET /issues) once, partitions into active (non-accepted) + and accepted vulnerabilities using the shared accepted definition, and writes + the FedRAMP Historical VER Activity snapshot JSON. Per VER-TFR-MRH, Class C + providers should refresh this at least every 14 days (scheduling is + operational, outside the fetcher). + +ksis: + - KSI-MLA-06 + - KSI-MLA-03 diff --git a/fetchers/paramify/vulnerability_detail_report/fetcher.py b/fetchers/paramify/vulnerability_detail_report/fetcher.py new file mode 100644 index 0000000..39db58c --- /dev/null +++ b/fetchers/paramify/vulnerability_detail_report/fetcher.py @@ -0,0 +1,108 @@ +#!/usr/bin/env python3 +""" +VER-RPT-VDT: Paramify Vulnerability Detail Report + +Generates the FedRAMP 20x Vulnerability Detail Report from Paramify issues. This +report covers NON-accepted vulnerabilities only -- the exact partition complement +of VER-RPT-AVI, using the shared accepted definition so every issue lands in +exactly one report. Derives overdueStatus and finalDisposition. Open issues with +a missing or epoch-sentinel evaluation date are reported without an +evaluationCompletedAt and surfaced in a VER-TFR-EVU warning. + +Output: $EVIDENCE_DIR/paramify_vulnerability_detail_report.json +Env: PARAMIFY_API_TOKEN (or PARAMIFY_UPLOAD_API_TOKEN), PARAMIFY_PROJECT_ID, + PARAMIFY_CERT_PACKAGE_URI, PARAMIFY_REPORT_FROM, PARAMIFY_REPORT_TO (opt), + PARAMIFY_API_BASE_URL (opt), PARAMIFY_HTTP_TIMEOUT (opt). +""" + +import json +import logging +import os +import sys +from pathlib import Path + +from dotenv import load_dotenv + +SCRIPT_DIR = Path(__file__).resolve().parent +sys.path.insert(0, str(SCRIPT_DIR.parent / "_shared")) +import ver_common as vc # noqa: E402 + +logger = logging.getLogger("paramify_vulnerability_detail_report") + + +def build_report(issues, cert_package_uri, report_from, report_to): + vulnerabilities = [ + vc.map_vulnerability_detail(i) for i in issues if not vc.is_accepted(i) + ] + return { + "certificationPackageOverviewUri": cert_package_uri, + "reportPeriod": {"from": report_from, "to": report_to}, + "vulnerabilities": vulnerabilities, + } + + +def main() -> int: + logging.basicConfig( + level=os.environ.get("LOG_LEVEL", "INFO"), + format="%(asctime)s %(levelname)s %(name)s %(message)s", + ) + load_dotenv() + + env = vc.resolve_common_env() + output_dir = Path(os.environ.get("EVIDENCE_DIR", "./evidence")) + output_dir.mkdir(parents=True, exist_ok=True) + + api_failures = [] + issues = vc.fetch_all_issues( + env["base_url"], env["token"], env["project_id"], + env["report_from"], env["report_to"], api_failures, + ) + + vc.warn_unevaluated_backlog( + issues, logger, + "reported without evaluationCompletedAt", + ) + + # Declared period in the report's own timestamp format. The RAW env values + # still drive fetch_all_issues above -- normalizing before the window is + # computed would turn a date-only bound into a midnight instant and silently + # drop that day's closures. + period_from, period_to = vc.report_period_bounds(env["report_from"], env["report_to"]) + + report = build_report( + issues, env["cert_package_uri"], period_from, period_to + ) + report["_summary"] = vc.build_vdt_summary( + report["vulnerabilities"], period_from, period_to + ) + report["_summary"]["collection"] = vc.build_collection_status(api_failures) + + output_path = output_dir / f"paramify_vulnerability_detail_report_{vc.target_slug(env)}.json" + with open(output_path, "w") as f: + json.dump(report, f, indent=2) + + s = report["_summary"] + d = s["dispositions"] + logger.info( + "Evidence saved to %s (%d non-accepted; FM=%d PM=%d FP=%d InProgress=%d; " + "overdue=%d, without-eval=%d)", + output_path, s["nonAcceptedVulnerabilities"], + d["fullyMitigated"], d["partiallyMitigated"], d["falsePositive"], + d["inProgress"], s["overdue"], s["withoutCompletedEvaluation"], + ) + + # /issues is the only call, so any failure means the report arrays above are + # empty for want of data -- NOT because the program has no open + # vulnerabilities. _summary.collection records that inside the payload. + if api_failures: + logger.error( + "%d API failure(s) during collection; the report is incomplete and " + "its counts must not be read as a clean result", + len(api_failures), + ) + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/fetchers/paramify/vulnerability_detail_report/fetcher.yaml b/fetchers/paramify/vulnerability_detail_report/fetcher.yaml new file mode 100644 index 0000000..33b6ca9 --- /dev/null +++ b/fetchers/paramify/vulnerability_detail_report/fetcher.yaml @@ -0,0 +1,69 @@ +name: paramify_vulnerability_detail_report +version: 0.1.0 +description: >- + Generates the FedRAMP 20x Vulnerability Detail Report (VER-RPT-VDT) from + Paramify issues: all non-accepted vulnerabilities (the exact partition + complement of VER-RPT-AVI), with derived overdue status and final disposition. +category: paramify + +supports_targets: true + +target_schema: + project_id: + type: string + required: true + env: PARAMIFY_PROJECT_ID + description: Paramify program (project) UUID. One target per program; from GET /projects. + program_name: + type: string + required: false + env: PARAMIFY_PROGRAM_NAME + description: >- + Human-readable program name from the Paramify workspace. Used for the + evidence filename and the uploaded artifact title; falls back to the + project UUID when absent. `paramify programs target` fills this in. + +runtime: + type: python + entry: fetcher.py + timeout: 600 + +output: + type: json + path: paramify_vulnerability_detail_report.json + aggregation: per_target + +secrets: + - name: api_token + env: PARAMIFY_API_TOKEN + +# Report-period knobs. Declared here (not under secrets) because every entry in +# secrets[] is mandatory -- the runner raises when a manifest omits one -- while +# config is optional and defaultable. Platform-wide knobs (cert_package_uri, +# api_base_url, http_timeout) live in fetchers/_categories/paramify.yaml; a +# manifest can still set any of these once under platforms.paramify.config. +config_schema: + report_from: + type: string + required: true + env: PARAMIFY_REPORT_FROM + description: ISO start of the report period, date ("2026-01-01") or timestamp. + report_to: + type: string + required: false + env: PARAMIFY_REPORT_TO + description: ISO end of the report period. Defaults to run time when omitted. + +evidence_set: + reference_id: EVD-PARAMIFY-VER-RPT-VDT + name: Paramify Vulnerability Detail Report (VER-RPT-VDT) + instructions: >- + Script: fetcher.py. Reads all project issues for PARAMIFY_PROJECT_ID from the + Paramify REST API (GET /issues; milestones read from the embedded array), + excludes accepted vulnerabilities via the shared accepted definition, derives + overdueStatus and finalDisposition, and writes the FedRAMP Vulnerability + Detail Report JSON. + +ksis: + - KSI-MLA-06 + - KSI-MLA-03 diff --git a/framework/api.py b/framework/api.py index 8bb71f7..d78dd45 100644 --- a/framework/api.py +++ b/framework/api.py @@ -26,6 +26,7 @@ from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable, Dict, List, Optional +from urllib.parse import urlparse import yaml @@ -106,6 +107,18 @@ def _fetcher_descriptor(f) -> dict: } +def discover(root: Path) -> Dict[str, dict]: + """One discovery pass, as {"fetchers": …, "platforms": …}. + + Every api function that needs these takes them as optional arguments so a + caller can scan once and thread the result through. Worth doing: each scan + walks and jsonschema-validates all ~125 fetcher.yaml files (~165 ms), and the + tree is immutable for the life of a command or a TUI redraw. Splat it into a + call — `api.validate(m, root, **discovered)`. + """ + return {"fetchers": discover_fetchers(root), "platforms": discover_platforms(root)} + + def catalog(root: Path) -> dict: """Discover all fetchers, group them by category, and describe every editable field. This single structure is both the UI form schema and the AI-readable @@ -1053,3 +1066,289 @@ def new_manifest_path(root, name: str, output_dir: str = "./evidence") -> Path: raise FileExistsError(str(path)) path.write_text(yaml.safe_dump(init_manifest(output_dir), sort_keys=False)) return path + + +# --------------------------------------------------------------------------- # +# Paramify workspace — program (project) discovery +# +# The Paramify API identifies programs by UUID, but the product UI shows people +# names. These helpers let a front-end offer the readable pick list and hand the +# UUID to the manifest, so nobody has to copy a UUID by hand. +# --------------------------------------------------------------------------- # + +_PROGRAMS_PATH = "/projects" # the UI's "programs" are the API's projects +_PROGRAMS_TIMEOUT = 30 + + +def program_display_name(program: dict) -> str: + """Best human-readable label for a program, falling back to its UUID.""" + return ( + program.get("name") + or program.get("system_name") + or program.get("short_name") + or program.get("id", "") + ) + + +def list_programs() -> List[dict]: + """Fetch the workspace's programs via GET /projects. + + Returns [{"id", "name", "system_name", "short_name"}] sorted by display name. + Raises RuntimeError with an actionable message on missing credentials, a + non-https endpoint, or a transport/HTTP failure — the CLI turns that into + {"ok": false, "errors": [...]}. + """ + import requests # local: keeps `paramify list`/`tui` startup free of it + + token = os.environ.get("PARAMIFY_API_TOKEN") or os.environ.get("PARAMIFY_UPLOAD_API_TOKEN") + if not token: + raise RuntimeError( + "No Paramify API token: set PARAMIFY_API_TOKEN (or " + "PARAMIFY_UPLOAD_API_TOKEN) to a token with read scope on the workspace" + ) + base_url = os.environ.get("PARAMIFY_API_BASE_URL") or "https://app.paramify.com/api/v0" + # Same rule the uploader enforces (uploader._base_url_error): a Bearer token + # must not go out over plaintext. Localhost is exempt so a local stub works. + host = urlparse(base_url).hostname or "" + if urlparse(base_url).scheme != "https" and host not in ("localhost", "127.0.0.1", "::1"): + raise RuntimeError( + f"PARAMIFY_API_BASE_URL must be https to protect the API token (got {base_url!r}); " + "only localhost may use http" + ) + url = f"{base_url.rstrip('/')}{_PROGRAMS_PATH}" + try: + resp = requests.get( + url, + headers={"Accept": "application/json", "Authorization": f"Bearer {token}"}, + timeout=_PROGRAMS_TIMEOUT, + ) + except Exception as e: # noqa: BLE001 — transport errors become one clean message + raise RuntimeError(f"could not reach {url}: {e}") from e + if resp.status_code in (401, 403): + raise RuntimeError( + f"Paramify rejected the token (HTTP {resp.status_code}) for {url}; " + "check that it has read scope on this workspace" + ) + if resp.status_code != 200: + raise RuntimeError(f"GET {url} failed (HTTP {resp.status_code}): {resp.text[:300]}") + try: + payload = resp.json() + except ValueError as e: + raise RuntimeError(f"GET {url} returned a non-JSON body: {resp.text[:200]}") from e + + # List endpoints wrap in {"projects": [...]}; tolerate a bare list. + raw = payload.get("projects", []) if isinstance(payload, dict) else payload + programs = [ + { + "id": p.get("id", ""), + "name": p.get("name") or "", + "system_name": p.get("systemName") or "", + "short_name": p.get("systemShortName") or "", + } + for p in raw + if isinstance(p, dict) and p.get("id") + ] + programs.sort(key=lambda p: program_display_name(p).lower()) + return programs + + +def resolve_program(programs: List[dict], selector: str) -> dict: + """Resolve one program from a user-supplied id or name. + + Exact id, then exact case-insensitive display name, then a unique + case-insensitive substring of the display name. Raises LookupError when + nothing matches and ValueError when a substring is ambiguous — an ambiguous + pick must never silently target the wrong program. + """ + needle = selector.strip() + for p in programs: + if p["id"] == needle: + return p + lowered = needle.lower() + exact = [p for p in programs if program_display_name(p).lower() == lowered] + if len(exact) == 1: + return exact[0] + if len(exact) > 1: + raise ValueError( + f"{selector!r} matches {len(exact)} programs by name; use the program id instead" + ) + partial = [p for p in programs if lowered in program_display_name(p).lower()] + if len(partial) == 1: + return partial[0] + if len(partial) > 1: + names = ", ".join(f"{program_display_name(p)} ({p['id']})" for p in partial[:5]) + raise ValueError(f"{selector!r} is ambiguous — matches: {names}") + raise LookupError(f"no program matches {selector!r}") + + +# Fetchers whose target IS a Paramify program. Scoped by category, not by field +# name alone: gitlab's fetchers also declare a `project_id` target field, and +# selecting on that name would write Paramify program UUIDs into gitlab targets +# in any manifest holding both. The category is the honest discriminator while +# this command group is Paramify-specific; a target_schema field declaring what +# it identifies would let it generalize (follow-up). +PROGRAM_CATEGORY = "paramify" +PROGRAM_ID_FIELD = "project_id" +PROGRAM_NAME_FIELD = "program_name" + + +def program_target_fetchers(m: dict, root: Path, fetchers: Optional[dict] = None) -> List[str]: + """Manifest entries whose target is a Paramify program — the default set, so + `programs target` with no fetcher argument does the obvious thing. + + `fetchers` is the {name: Fetcher} mapping from discover_fetchers(), passed in + when the caller already has one.""" + discovered = fetchers if fetchers is not None else discover_fetchers(root) + return [ + entry["use"] for entry in _entries(m) + if (f := discovered.get(entry.get("use"))) is not None + and f.category == PROGRAM_CATEGORY + and f.supports_targets + and PROGRAM_ID_FIELD in f.target_schema + ] + + +def _targeted_program_ids(m: dict, use: str) -> List[str]: + """project_id values already targeted on a fetcher entry — so re-running + `programs target` tops up the manifest instead of duplicating targets.""" + entry = _find_entry(m, use) or {} + return [t[PROGRAM_ID_FIELD] for t in (entry.get("targets") or []) if t.get(PROGRAM_ID_FIELD)] + + +def add_program_targets( + m: dict, uses: List[str], programs: List[dict], root: Optional[Path] = None, + fetchers: Optional[dict] = None, +) -> dict: + """Add one target per (fetcher x program), skipping programs already targeted. + + A target carries only what varies per program: project_id and its readable + program_name. Everything uniform across the workspace — the Certification + Package Overview URI, the API base URL — is category config, set once under + platforms..config. + + Mutates `m` in place and returns a JSON-able report rather than the manifest: + this is a composite of add_target() calls, and the caller needs to know what + landed and what was already there. + """ + discovered = fetchers if fetchers is not None else (discover_fetchers(root) if root else {}) + added: List[dict] = [] + skipped: List[dict] = [] + for use in uses: + existing = set(_targeted_program_ids(m, use)) + for program in programs: + label = program_display_name(program) + if program["id"] in existing: + skipped.append({"use": use, "program_id": program["id"], "program_name": label, + "reason": "already targeted"}) + continue + values: Dict[str, Any] = {PROGRAM_ID_FIELD: program["id"]} + f = discovered.get(use) + declares_name = f is not None and PROGRAM_NAME_FIELD in f.target_schema + if declares_name and label and label != program["id"]: + values[PROGRAM_NAME_FIELD] = label + add_target(m, use, values) + existing.add(program["id"]) + added.append({"use": use, "program_id": program["id"], "program_name": label}) + return {"added": added, "skipped": skipped} + + +def effective_config( + m: dict, uses: List[str], root: Path, fetchers: Optional[dict] = None, + platforms: Optional[dict] = None, +) -> Dict[str, List[dict]]: + """Per entry, every config field that applies to it, with its value and where + that value comes from — the merge the runner actually performs: + + platform defaults <- platform values <- per-fetcher values + + Returns {use: [descriptor + {"value", "source"}]}, where source is "entry", + "platforms.", "default", or None when nothing supplies it. A field + the category declares (and the fetcher doesn't) is included too, since it is + injected into that fetcher's environment just the same. + + Front-ends need this to render config honestly: reading only the entry's own + `config` block reports a value set once at the category level as unset on + every entry that inherits it. Batched over `uses` so a caller redrawing a + table scans the fetcher tree once, not once per row. + """ + discovered = fetchers if fetchers is not None else discover_fetchers(root) + specs = platforms if platforms is not None else discover_platforms(root) + all_platforms = _run(m).get("platforms") or {} + + out: Dict[str, List[dict]] = {} + for use in uses: + f = discovered.get(use) + if f is None: + out[use] = [] + continue + spec = specs.get(f.category) if f.category else None + schema: Dict[str, ConfigField] = {} + if spec: + schema.update(spec.config_schema) + schema.update(f.config_schema) # fetcher overrides platform on a name clash + + platform_values = ((all_platforms.get(f.category or "") or {}).get("config")) or {} + entry_values = (_find_entry(m, use) or {}).get("config") or {} + + fields: List[dict] = [] + for name, fdef in schema.items(): + d = _config_descriptor(fdef) + d["category"] = f.category + if name in entry_values: + d["value"], d["source"] = entry_values[name], "entry" + elif name in platform_values: + d["value"], d["source"] = platform_values[name], f"platforms.{f.category}" + elif fdef.default is not None: + d["value"], d["source"] = fdef.default, "default" + else: + d["value"], d["source"] = None, None + fields.append(d) + out[use] = fields + return out + +def shared_config_state( + m: dict, uses: List[str], field_name: str, root: Path, *, + fetchers: Optional[dict] = None, platforms: Optional[dict] = None, +) -> dict: + """What a front-end needs to show and edit one config field shared across + `uses` — the kind set once per category rather than per entry: + + categories every category among `uses` that accepts `field_name` + value the value they all resolve to today, or None when nothing + supplies one or the entries disagree + sources where those values come from ("entry", "platforms.", + "default"), distinct, in the order met + conflict True when the entries resolve to different values + missing categories where nothing supplies a required value + + A view over effective_config() rather than a second merge, so "is it set" + can't mean membership here and truthiness there — which is exactly how the + two functions this replaced had already drifted apart. Offering `value` as an + edit default is honest only while `conflict` is False: one entry's override + presented as the manifest's answer would misreport the other entries. + """ + categories: List[str] = [] + missing: List[str] = [] + values: List[Any] = [] + sources: List[str] = [] + for fields in effective_config(m, uses, root, fetchers, platforms).values(): + for d in fields: + if d["name"] != field_name or not d["category"]: + continue + if d["category"] not in categories: + categories.append(d["category"]) + if d["required"] and d["source"] is None and d["category"] not in missing: + missing.append(d["category"]) + if d["source"] is None: + continue + if d["value"] not in values: + values.append(d["value"]) + if d["source"] not in sources: + sources.append(d["source"]) + return { + "categories": categories, + "value": values[0] if len(values) == 1 else None, + "sources": sources, + "conflict": len(values) > 1, + "missing": missing, + } diff --git a/framework/cli.py b/framework/cli.py index 400d8d5..da4cc9d 100644 --- a/framework/cli.py +++ b/framework/cli.py @@ -18,6 +18,11 @@ paramify evidence [--json] # read one evidence file paramify upload [run-dir] [--dry-run] [--json] +Paramify workspace (live lookups; needs PARAMIFY_API_TOKEN with read scope): + paramify programs list [--json] # programs in the workspace: name + id + paramify programs target [fetcher ...] [--program NAME|ID ...] [--all] + [--cert-uri URI] [--report-from DATE] [-f FILE] [--json] + Manifest editing (writes the manifest file; -f/--file, default ./manifest.yaml; every subcommand accepts --json, emitting {"ok", "path", "errors"}): paramify manifest init [--output-dir DIR] @@ -47,8 +52,11 @@ from __future__ import annotations import json +import re +import sys +from datetime import datetime from pathlib import Path -from typing import List, Optional +from typing import List, NoReturn, Optional import typer @@ -81,6 +89,13 @@ ) app.add_typer(scripts_app, name="scripts") +programs_app = typer.Typer( + no_args_is_help=True, + context_settings=_HELP_OPTS, + help="List the workspace's programs and turn them into manifest targets.", +) +app.add_typer(programs_app, name="programs") + # --------------------------------------------------------------------------- # # Small shared helpers (ported verbatim from the previous argparse CLI) @@ -99,7 +114,7 @@ def _coerce(raw: str, typ: str): return raw -def _fail(path, msg: str, json_out: bool): +def _fail(path, msg: str, json_out: bool) -> NoReturn: """Report a command-level argument error honoring --json, then exit 1. Keeps the {ok, path, errors} contract on mutator argument-error paths (a @@ -955,6 +970,244 @@ def tui_cmd( raise typer.Exit(1) launch(manifest, at) +# --------------------------------------------------------------------------- # +# Paramify workspace — pick programs by name, target them by UUID +# +# The API only accepts project UUIDs; people know their programs by name. These +# commands close that gap: list what's in the workspace, let the operator choose, +# then reuse the manifest mutators to write the targets. Selection is interactive +# by default and fully flag-driven under --json, so an AI caller never hits a +# prompt it can't answer. +# --------------------------------------------------------------------------- # + +def _is_iso_datish(value: str) -> bool: + """Accept what the fetchers' date parser accepts: an ISO date or timestamp, + with a trailing Z allowed. Mirrors ver_common._parse_iso — fetchers aren't an + importable package, so the rule is restated rather than shared.""" + try: + datetime.fromisoformat(value.strip().replace("Z", "+00:00")) + except ValueError: + return False + return True + + +def _can_prompt(json_out: bool) -> bool: + """--json has no way to answer a prompt, and neither does a piped stdin.""" + return not json_out and sys.stdin.isatty() + + +def _config_origin(state: dict) -> str: + """Where a shared config field's value comes from, for the line above its + prompt. The value itself is the prompt's default, so this says only what the + bracketed default can't: whether it's already stored, and where — an entry's + own config outranks the category value this command writes.""" + labels = ", ".join( + "this entry's own config" if s == "entry" else + "the fetcher default" if s == "default" else s + for s in state["sources"] + ) + if state["conflict"]: + return f"differs across entries ({labels}) — one value replaces them all" + return f"set in {labels}" if labels else "not set yet" + + +def _programs_or_exit(json_out: bool) -> List[dict]: + try: + return api.list_programs() + except RuntimeError as e: + _fail(None, str(e), json_out) + + +def _parse_selection(raw: str, count: int) -> List[int]: + """Parse "1,3,5", "1-3", "2 4", or "all" into zero-based indices. + + Raises ValueError on anything out of range so a typo can't silently target + the wrong program. + """ + text = raw.strip().lower() + if text in ("all", "*"): + return list(range(count)) + picked: List[int] = [] + for token in re.split(r"[,\s]+", text): + if not token: + continue + lo_s, _, hi_s = token.partition("-") + lo, hi = int(lo_s), int(hi_s or lo_s) + if not 1 <= lo <= hi <= count: + raise ValueError(f"{token!r} is outside 1-{count}") + picked.extend(range(lo - 1, hi)) + ordered = list(dict.fromkeys(picked)) # de-dupe, keep the order typed + if not ordered: + raise ValueError("no programs selected") + return ordered + + +@programs_app.command("list") +def programs_list(json_out: bool = typer.Option(False, "--json", help="Emit JSON")): + """List the programs in the Paramify workspace (name + id).""" + programs = _programs_or_exit(json_out) + if json_out: + typer.echo(json.dumps({"ok": True, "programs": programs}, indent=2)) + return + if not programs: + typer.echo("No programs found in this workspace.") + return + width = max(len(api.program_display_name(p)) for p in programs) + typer.echo(f"{len(programs)} program(s):\n") + for p in programs: + short = f" [{p['short_name']}]" if p["short_name"] else "" + typer.echo(f" {api.program_display_name(p):<{width}} {p['id']}{short}") + + +@programs_app.command("target") +def programs_target( + fetchers: Optional[List[str]] = typer.Argument( + None, help="Fanout fetcher(s) to add targets to. Default: every manifest entry that takes a program." + ), + program: Optional[List[str]] = typer.Option( + None, "--program", "-p", help="Program name or id (repeatable). Omit to choose interactively." + ), + all_programs: bool = typer.Option(False, "--all", help="Target every program in the workspace"), + cert_uri: Optional[str] = typer.Option( + None, "--cert-uri", + help="Certification Package Overview URI for the workspace. Set once as category " + "config; shown for confirmation on every interactive run.", + ), + report_from: Optional[str] = typer.Option( + None, "--report-from", + help="Report period start (ISO date, e.g. 2026-01-01). Set once as category " + "config; shown for confirmation on every interactive run.", + ), + file: str = typer.Option(_DEFAULT_MANIFEST, "-f", "--file", help="Manifest path"), + json_out: bool = typer.Option(False, "--json", help="Emit JSON"), +): + """Select programs from the workspace and add them as manifest targets.""" + root = api.find_repo_root() + path = Path(file).resolve() + m = _read_for_edit(path, json_out) + # Discovered once and threaded through every api call below. Each of these + # walks + schema-validates all ~125 fetcher.yaml files; the tree is immutable + # for the life of the command, so one pass is enough. + discovered = api.discover(root) + + uses = list(fetchers or []) + if not uses: + uses = api.program_target_fetchers(m, root, discovered["fetchers"]) + if not uses: + _fail( + path, + "No manifest entry takes a program as a target. Add one first " + "(e.g. paramify manifest add paramify_accepted_vulnerabilities), " + "or name the fetcher explicitly.", + json_out, + ) + + programs = _programs_or_exit(json_out) + if not programs: + _fail(path, "No programs found in this workspace.", json_out) + + # --- selection ---------------------------------------------------------- # + selected: List[dict] = [] + if all_programs: + selected = programs + elif program: + for selector in program: + try: + selected.append(api.resolve_program(programs, selector)) + except (LookupError, ValueError) as e: + _fail(path, str(e), json_out) + selected = list({p["id"]: p for p in selected}.values()) + else: + if not _can_prompt(json_out): + _fail( + path, + "No programs chosen and no terminal to prompt on: pass --program " + "NAME|ID (repeatable) or --all.", + json_out, + ) + typer.echo(f"Programs in this workspace (targeting: {', '.join(uses)})\n") + for i, p in enumerate(programs, 1): + short = f" [{p['short_name']}]" if p["short_name"] else "" + typer.echo(f" {i:>3}. {api.program_display_name(p)}{short}") + typer.echo("") + try: + indices = _parse_selection( + typer.prompt("Select programs (e.g. 1,3 or 1-3 or all)"), len(programs) + ) + except ValueError as e: + _fail(path, f"Invalid selection: {e}", json_out) + selected = [programs[i] for i in indices] + + # --- shared config ------------------------------------------------------- # + # Values that don't vary per program are written once to + # platforms..config, where every fetcher in the category picks them + # up. Interactively each one is shown on every run with the value in force as + # the prompt default, because a re-run is also how you fix a wrong URI or roll + # the report window forward — enter keeps what's there and writes nothing. + # Nothing is asked when the flag supplied it (that's an override) or when + # there's no terminal, where only a genuinely missing value is an error. + fields = [ + (name, flag, prompt_text, is_date, supplied, + api.shared_config_state(m, uses, name, root, **discovered)) + for name, flag, prompt_text, is_date, supplied in ( + ("cert_package_uri", "--cert-uri", + "Certification Package Overview URI (used for every program)", False, cert_uri), + ("report_from", "--report-from", + "Report period start — ISO date, e.g. 2026-01-01 (used for every program)", True, report_from), + ) + ] + # The header promises "enter keeps it" only when something is actually stored + # to keep — on a first run there is nothing to show and every prompt is bare. + header = "\nShared config — one value for every program." + ( + " Enter keeps what's shown." if any(s["value"] for *_, s in fields) else "" + ) + announced = False + for field_name, flag, prompt_text, is_date, supplied, state in fields: + if not state["categories"]: + continue + current = "" if state["conflict"] else str(state["value"] or "") + value = (supplied or "").strip() + if not value: + if not _can_prompt(json_out): + if state["missing"]: + _fail( + path, + f"{field_name} is not set for " + + ", ".join(f"platforms.{c}.config" for c in state["missing"]) + + f". Pass {flag} .", + json_out, + ) + continue + if not announced: + typer.echo(header) + announced = True + typer.echo(f"\n {field_name}: {_config_origin(state)}") + value = typer.prompt(prompt_text, default=current or None).strip() + if not value: + _fail(path, f"No {field_name} given; nothing written.", json_out) + if is_date and not _is_iso_datish(value): + # A date the fetcher can't parse yields an empty report window, which + # silently drops every closed issue rather than failing — so reject it + # here, where it's still a typo instead of a wrong report. + _fail( + path, + f"{field_name}: {value!r} is not an ISO date or timestamp " + "(e.g. 2026-01-01 or 2026-01-01T00:00:00Z).", + json_out, + ) + if value == current and not state["missing"] and not supplied: + continue # enter on a value already in force everywhere: leave it be + for category in state["categories"]: + api.set_platform_config(m, category, field_name, value) + + report = api.add_program_targets(m, uses, selected, fetchers=discovered["fetchers"]) + if not json_out: + for rec in report["added"]: + typer.echo(f" + {rec['use']} -> {rec['program_name']} ({rec['program_id']})") + for rec in report["skipped"]: + typer.echo(f" = {rec['use']} -> {rec['program_name']} ({rec['reason']})") + _save_and_report(m, path, root, json_out, verb="Updated") + if __name__ == "__main__": app() diff --git a/framework/tui/app.py b/framework/tui/app.py index e9cf424..a597c67 100644 --- a/framework/tui/app.py +++ b/framework/tui/app.py @@ -27,6 +27,11 @@ class FetcherApp(App): CSS_PATH = "styles/index.tcss" TITLE = "paramify-fetchers" + # Textual binds ctrl+p to its command palette as a *priority* binding, which + # is checked before the focused widget — it swallowed the Paramify tab's + # ctrl+p (preview) entirely. We register no command providers, so the palette + # only offers Textual's own built-ins; turn it off and keep the key. + ENABLE_COMMAND_PALETTE = False def __init__( self, manifest_path: Optional[str] = None, root_override: Optional[str] = None diff --git a/framework/tui/modals.py b/framework/tui/modals.py index 0737142..bd1593d 100644 --- a/framework/tui/modals.py +++ b/framework/tui/modals.py @@ -293,6 +293,12 @@ def action_cancel(self) -> None: class ConfirmModal(ModalScreen[bool]): """A yes/no confirmation. Returns True on confirm, False otherwise.""" + # Yes is composed first, so the default AUTO_FOCUS ("*") put enter on the + # destructive answer — this dialog gates deleting a manifest file, removing + # an entry, and uploading to Paramify. Focus No: enter and escape both mean + # no, y means yes. + AUTO_FOCUS = "#no" + BINDINGS = [ Binding("escape", "no", "No"), Binding("n", "no", "No"), diff --git a/framework/tui/render.py b/framework/tui/render.py index d77fc18..043d5dc 100644 --- a/framework/tui/render.py +++ b/framework/tui/render.py @@ -111,7 +111,12 @@ def _status(set_: bool, required: bool) -> Text: return Text("required — unset", style=palette.WARN) if required else Text("unset", style="dim") -def entry_detail(descriptor: Optional[dict], entry: dict, errors: Optional[List[str]] = None) -> RenderableType: +def entry_detail( + descriptor: Optional[dict], + entry: dict, + errors: Optional[List[str]] = None, + config_view: Optional[List[dict]] = None, +) -> RenderableType: """Render one manifest entry: its current config/secrets/targets vs the contract.""" use = entry.get("use", "?") if descriptor is None: @@ -125,7 +130,6 @@ def entry_detail(descriptor: Optional[dict], entry: dict, errors: Optional[List[ header.append(use, style=f"bold {palette.FG}") header.append(" [fanout]" if fanout else " [single]", style="dim") - cfg = entry.get("config") or {} secs = entry.get("secrets") or {} parts: List[RenderableType] = [header, Text()] @@ -139,15 +143,23 @@ def entry_detail(descriptor: Optional[dict], entry: dict, errors: Optional[List[ rows.append((s["name"], value)) parts += [Text("secrets", style="bold"), _kv_table(rows), Text()] - # config - config_fields = descriptor.get("config", []) + # config — api.effective_config()'s merged view (platform defaults <- platform + # values <- entry values), so a value set once at the category level shows as + # set, and says where it came from, on every entry inheriting it. Empty when + # the merge failed: no config block beats a knowingly-wrong one. + config_fields = config_view or [] if config_fields: rows = [] for c in config_fields: - if c["name"] in cfg: - rows.append((c["name"], Text(str(cfg[c["name"]]), style=palette.FG))) - elif c.get("default") is not None: - rows.append((c["name"], Text(f"{c['default']} (default)", style="dim"))) + source = c.get("source") + if source == "entry": + rows.append((c["name"], Text(str(c["value"]), style=palette.FG))) + elif source and source.startswith("platforms."): + value = Text(str(c["value"]), style=palette.FG) + value.append(f" ({source})", style="dim") + rows.append((c["name"], value)) + elif source == "default": + rows.append((c["name"], Text(f"{c['value']} (default)", style="dim"))) else: rows.append((c["name"], _status(False, c.get("required", False)))) parts += [Text("config", style="bold"), _kv_table(rows), Text()] diff --git a/framework/tui/screens/evidence.py b/framework/tui/screens/evidence.py index 2dd8f0a..fba5177 100644 --- a/framework/tui/screens/evidence.py +++ b/framework/tui/screens/evidence.py @@ -26,7 +26,7 @@ class EvidencePage(Vertical): - HINTS = [("↑↓", "runs"), ("enter", "view"), ("ctrl+r", "refresh")] + HINTS = [("↑↓", "runs"), ("enter", "open / view"), ("ctrl+r", "refresh")] BINDINGS = [Binding("ctrl+r", "refresh_runs", "Refresh")] @@ -135,7 +135,16 @@ def on_data_table_row_highlighted(self, event: DataTable.RowHighlighted) -> None self._show_run(event.row_key.value) def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None: - if event.data_table.id == "evidence-files" and event.row_key.value: + if event.data_table.id == "evidence-runs": + # Enter drills into the run. Focus opens on the runs table, and the + # files table — where enter actually views evidence — was reachable + # only by tab/click, so enter looked broken on the way in. + files = self.query_one("#evidence-files", DataTable) + if files.row_count: + files.focus() + else: + self.notify("This run has no evidence files.") + elif event.data_table.id == "evidence-files" and event.row_key.value: self._open_file(event.row_key.value) @on(Button.Pressed, "#evidence-refresh") diff --git a/framework/tui/screens/manifest.py b/framework/tui/screens/manifest.py index dffe1c5..b12bf3f 100644 --- a/framework/tui/screens/manifest.py +++ b/framework/tui/screens/manifest.py @@ -50,7 +50,9 @@ class ManifestPage(Vertical): def compose(self) -> ComposeResult: with Horizontal(id="manifest-top"): yield Static("output dir:", classes="inline-label") - yield Input(placeholder="./evidence", id="manifest-output-dir") + # select_on_focus off: Textual selects the whole value on focus, so + # the first keystroke replaced the existing path wholesale. + yield Input(placeholder="./evidence", id="manifest-output-dir", select_on_focus=False) yield Button("Add fetcher", variant="primary", id="btn-add") yield Button("Save", id="btn-save") with Horizontal(id="manifest-body"): @@ -64,6 +66,8 @@ def compose(self) -> ComposeResult: def on_mount(self) -> None: self._selected: Optional[str] = None self._errors: List[str] = [] + # {use: merged config view}, rebuilt with the table (api.effective_config) + self._config_view: Dict[str, List[dict]] = {} self.query_one("#manifest-entries-panel", Vertical).border_title = "fetchers" self.query_one("#manifest-detail-scroll", VerticalScroll).border_title = "detail" dt = self.query_one("#manifest-entries", DataTable) @@ -113,11 +117,25 @@ def rebuild(self) -> None: descriptors = self._descriptors() entries = self._entries() + # One discovery pass shared by validate + effective_config: each would + # otherwise walk and schema-validate all ~125 fetcher.yaml files, and + # rebuild() runs on every mutation and tab switch. try: - self._errors = api.validate(self._manifest, self.app.root_path) + discovered = api.discover(self.app.root_path) + except Exception: # never let a discovery failure kill the UI + discovered = {"fetchers": {}, "platforms": {}} + try: + self._errors = api.validate(self._manifest, self.app.root_path, **discovered) except Exception as exc: # never let a validation crash kill the UI self._errors = [f"validation error: {exc}"] by_use = self._bucket_errors(self._errors, entries) + try: + self._config_view = api.effective_config( + self._manifest, [e.get("use", "") for e in entries], + self.app.root_path, **discovered, + ) + except Exception: # never let a config-merge failure kill the UI + self._config_view = {} dt.clear() row_keys: List[str] = [] @@ -126,7 +144,7 @@ def rebuild(self) -> None: d = descriptors.get(use) fanout = bool(d and d.get("supports_targets")) sset, stot = self._secret_counts(d, e) - cset, ctot = self._config_counts(d, e) + cset, ctot = self._config_counts(self._config_view.get(use)) ntargets = len(e.get("targets") or []) errs = by_use.get(use, []) status = palette.pill("✓", "ok") if not errs else palette.pill(f"⚠ {len(errs)}", "warn") @@ -179,7 +197,7 @@ def _refresh_detail(self) -> None: # Bucket against the full entry list so index-prefixed (entry[i]) errors # attribute correctly, then take this entry's slice. errs = self._bucket_errors(self._errors, self._entries()).get(use, []) - detail.update(render.entry_detail(d, entry, errs)) + detail.update(render.entry_detail(d, entry, errs, self._config_view.get(use))) def _set_issues(self, errors: List[str]) -> None: issues = self.query_one("#manifest-issues", Static) @@ -203,12 +221,15 @@ def _secret_counts(d: Optional[dict], e: dict) -> tuple: return (sum(1 for s in top if s["name"] in have), len(top)) @staticmethod - def _config_counts(d: Optional[dict], e: dict) -> tuple: - if not d: + def _config_counts(view: Optional[List[dict]]) -> tuple: + """(explicitly set, total applicable) from api.effective_config()'s view. + + "Set" means a value was supplied — in the entry or at the category level. + Counting only the entry's own block reported category config as unset. + """ + if not view: return (0, 0) - fields = d.get("config", []) - have = e.get("config") or {} - return (sum(1 for f in fields if f["name"] in have), len(fields)) + return (sum(1 for c in view if c.get("source") not in (None, "default")), len(view)) @staticmethod def _bucket_errors(errors: List[str], entries: List[dict]) -> Dict[str, List[str]]: @@ -261,11 +282,27 @@ def on_data_table_row_highlighted(self, event: DataTable.RowHighlighted) -> None self._selected = event.row_key.value self._refresh_detail() + def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None: + # Enter on a row edits it — what a table row implies, and previously the + # one key on this page that did nothing at all. + self.action_edit_entry() + @on(Input.Submitted, "#manifest-output-dir") def _on_output_dir(self, event: Input.Submitted) -> None: - if self._manifest is None: - return - api.set_output_dir(self._manifest, event.value.strip() or "./evidence") + self._commit_output_dir(event.value.strip() or "./evidence") + + @on(Input.Blurred, "#manifest-output-dir") + def _on_output_dir_blurred(self, event: Input.Blurred) -> None: + # Commit on blur as well as enter: an edit that was never submitted got + # silently reverted by the next rebuild(). A cleared field is left alone + # (that's an empty field, not a request for the default). + if event.value.strip(): + self._commit_output_dir(event.value.strip()) + + def _commit_output_dir(self, value: str) -> None: + if self._manifest is None or value == (self._run().get("output_dir") or ""): + return # blur fires on every focus change; only a real change commits + api.set_output_dir(self._manifest, value) self.notify("Output dir updated.") self.rebuild() diff --git a/framework/tui/screens/run.py b/framework/tui/screens/run.py index 53d0d9a..fd3cb2c 100644 --- a/framework/tui/screens/run.py +++ b/framework/tui/screens/run.py @@ -44,7 +44,11 @@ def __init__(self, ev: dict) -> None: class RunPage(Vertical): - HINTS = [("ctrl+r", "run")] + # Focus opens on the ▶ Run button, and Button binds enter — so enter runs the + # manifest too. Advertised rather than changed: the status table can't hold + # focus before the first run (.panel.empty hides it), so moving focus there + # would leave ctrl+r dead instead. + HINTS = [("enter/ctrl+r", "run")] BINDINGS = [Binding("ctrl+r", "run_manifest", "Run")] diff --git a/framework/tui/screens/upload.py b/framework/tui/screens/upload.py index 591d63e..91daefe 100644 --- a/framework/tui/screens/upload.py +++ b/framework/tui/screens/upload.py @@ -44,11 +44,14 @@ def __init__(self, ev: dict) -> None: class UploadPage(Vertical): - HINTS = [("ctrl+u", "upload"), ("ctrl+p", "preview"), ("ctrl+s", "sync"), ("ctrl+r", "refresh")] + HINTS = [("ctrl+u", "upload"), ("p", "preview"), ("ctrl+s", "sync"), ("ctrl+r", "refresh")] BINDINGS = [ Binding("ctrl+u", "upload_run", "Upload"), - Binding("ctrl+p", "preview_scripts", "Preview"), + # p mirrors the Manifest tab's preview key (this page has no Input to eat + # it); ctrl+p stays as an alias, which needs App.ENABLE_COMMAND_PALETTE + # off — Textual's palette claims ctrl+p as a priority binding. + Binding("p,ctrl+p", "preview_scripts", "Preview"), Binding("ctrl+s", "sync_scripts", "Sync Scripts"), Binding("ctrl+r", "refresh_upload", "Refresh"), ] diff --git a/framework/tui/screens/workspace.py b/framework/tui/screens/workspace.py index 13e6619..fc885d7 100644 --- a/framework/tui/screens/workspace.py +++ b/framework/tui/screens/workspace.py @@ -31,7 +31,10 @@ class WorkspaceScreen(Screen): # Screen-level bindings shown on every tab's footer (after the page-specific # hints). Keep in sync with BINDINGS below. - WORKSPACE_HINTS = [("1-5", "tabs"), ("m", "manifest"), ("q", "quit")] + # esc is listed because it is the only way out of a focused text field back + # to the shortcut keys: an Input consumes every printable key, so while one + # holds focus none of the hints above it are live. + WORKSPACE_HINTS = [("1-5", "tabs"), ("m", "manifest"), ("esc", "leave field"), ("q", "quit")] BINDINGS = [ Binding("1", "go_tab(0)", "Catalog"), @@ -105,8 +108,16 @@ def _update_chrome(self) -> None: self.query_one(HintFooter).set_hints(page_hints + self.WORKSPACE_HINTS) def _go_to_tab(self, tab_id: str) -> None: + tabs = self.query_one(TabbedContent) + if tabs.active == tab_id: + # Assigning the active tab it already has fires no TabActivated, so + # nothing would restore focus after set_focus(None) below — pressing + # the number of the tab you're on would silently kill every page + # binding. Re-home focus directly instead. + self.call_after_refresh(self._focus_active_pane) + return self.set_focus(None) # Textual reverts an active-change while focus is in the outgoing pane - self.query_one(TabbedContent).active = tab_id + tabs.active = tab_id # Focus follows via on_tabbed_content_tab_activated (fires for programmatic # changes too), so this is the single place pane focus is decided. diff --git a/pyproject.toml b/pyproject.toml index c176cd7..a7cd16e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,11 +24,15 @@ dependencies = [ ] [project.optional-dependencies] -tui = ["textual>=1.0,<2.0"] +# textual 8.x is what the TUI is developed and tested against (tests/test_tui_keys.py). +# The old >=1.0,<2.0 range was a fiction — nobody ran it, and focus/Input behaviour +# (select_on_focus, blurred cursor styles) differs enough that the TUI is not the +# same app on 1.x. +tui = ["textual>=8,<9"] checkov = ["checkov"] dev = ["pytest", "ruff", "mypy"] # Convenience: every front-end + dev tooling in one install. -all = ["textual>=1.0,<2.0", "checkov", "pytest", "ruff", "mypy"] +all = ["textual>=8,<9", "checkov", "pytest", "ruff", "mypy"] # The single entry point. `paramify` steers every front-end: the headless # commands, plus `paramify tui`. (Renaming later is a one-line change here; add diff --git a/requirements.txt b/requirements.txt index ec5f044..403d1b0 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,8 +8,9 @@ jsonschema # steers every front-end (headless commands + `paramify tui`). typer -# Terminal UI (framework/tui) — optional unless you run the console -textual>=1.0,<2.0 +# Terminal UI (framework/tui) — optional unless you run the console. +# Pinned to the 8.x line the TUI is tested against; see pyproject's [tui] extra. +textual>=8,<9 # Checkov category (fetchers/checkov/*) — the bash fetchers shell out to the # `checkov` CLI. Also requires system `git` (clone) + `jq`, which are not pip deps. diff --git a/tests/test_cli.py b/tests/test_cli.py index 5177ff6..68b979c 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -27,7 +27,7 @@ from typer.main import get_command from typer.testing import CliRunner -from framework import api +from framework import api, cli from framework.cli import app REPO_ROOT = Path(__file__).resolve().parent.parent @@ -49,8 +49,9 @@ def _registered(): EXPECTED_TOP = { "list", "catalog", "describe", "ksi", "doctor", "manifests", "runs", - "evidence", "validate", "run", "upload", "manifest", "scripts", "tui", + "evidence", "validate", "run", "upload", "manifest", "scripts", "programs", "tui", } +EXPECTED_PROGRAMS = {"list", "target"} EXPECTED_MANIFEST = { "init", "new", "add", "remove", "set-config", "set-secret", "add-target", "remove-target", "set-platform-config", @@ -66,6 +67,17 @@ def test_all_expected_commands_registered(): assert EXPECTED_SCRIPTS <= scripts, f"missing scripts subcommands: {EXPECTED_SCRIPTS - scripts}" +def test_programs_subcommands_registered(): + """Assert the subcommands are actually attached to the sub-app. + + Worth its own test: a @programs_app.command decorator placed after the + module's `if __name__ == "__main__": app()` line never runs before dispatch, + so the group loads but reports "No such command" at runtime. + """ + programs = set(get_command(app).commands["programs"].commands.keys()) + assert EXPECTED_PROGRAMS <= programs, f"missing programs subcommands: {EXPECTED_PROGRAMS - programs}" + + def test_doctor_json_ok_without_manifest(): """Without a manifest, doctor is a Python-version gate; tools are advisory.""" result = runner.invoke(app, ["doctor", "--json"]) @@ -112,9 +124,14 @@ def _tui_api_calls() -> set[str]: # its own command. Keep this in sync with the TUI; the test below enforces it. API_TO_CLI = { "find_repo_root": "", + "discover": "", "catalog": "list / catalog / describe", "list_manifests": "manifests", "read_manifest": "manifest show", + # Read-only render helper: the runner's merged config view (platform <- entry) + # behind what the manifest screen displays. No command of its own — the CLI + # surfaces the same facts through `manifest show` + `validate`. + "effective_config": "", "init_manifest": "manifest init", "new_manifest_path": "manifest new", "add_entry": "manifest add", @@ -531,3 +548,334 @@ def test_manifest_new_creates_under_manifests_dir(in_repo): finally: if target.exists(): target.unlink() + + +# --------------------------------------------------------------------------- # +# programs — pick a program by name, target it by UUID +# +# GET /projects is stubbed at the api boundary (never over the wire), so these +# assert the selection/resolution/manifest-wiring logic, not the HTTP client. +# --------------------------------------------------------------------------- # + +_PROGRAMS = [ + {"id": "aaaa1111-0000-0000-0000-000000000000", "name": "Alpha Cloud Services", + "system_name": "Alpha Cloud Services", "short_name": "ACS"}, + {"id": "bbbb2222-0000-0000-0000-000000000000", "name": "Beta Platform", + "system_name": "Beta Platform", "short_name": "BETA"}, + {"id": "cccc3333-0000-0000-0000-000000000000", "name": "Gamma Analytics", + "system_name": "Gamma Analytics", "short_name": "GAM"}, +] + +_VER_FETCHER = "paramify_accepted_vulnerabilities" + + +@pytest.fixture +def stub_programs(monkeypatch): + """Stub the workspace lookup so no test touches the network.""" + monkeypatch.setattr(api, "list_programs", lambda *a, **k: list(_PROGRAMS)) + return _PROGRAMS + + +@pytest.fixture +def ver_manifest(tmp_path, in_repo): + """A manifest at the point `programs target` is normally reached: the entry + added and its secret wired, but no targets and no shared config yet — those + are exactly what the command fills in. + """ + path = tmp_path / "m.yaml" + m = api.init_manifest(str(tmp_path / "out")) + api.add_entry(m, _VER_FETCHER) + api.set_secret(m, _VER_FETCHER, "api_token", "PARAMIFY_API_TOKEN") + api.dump_manifest(m, path, in_repo) + return path + + +def test_programs_list_json(stub_programs): + rep = _json(runner.invoke(app, ["programs", "list", "--json"])) + assert rep["ok"] is True + assert [p["name"] for p in rep["programs"]] == [p["name"] for p in _PROGRAMS] + + +def test_programs_list_human_shows_name_and_id(stub_programs): + result = runner.invoke(app, ["programs", "list"]) + assert result.exit_code == 0, result.output + assert "Alpha Cloud Services" in result.output + assert "aaaa1111-0000-0000-0000-000000000000" in result.output + + +def test_programs_list_reports_missing_token_as_json_error(monkeypatch): + def boom(*a, **k): + raise RuntimeError("No Paramify API token: set PARAMIFY_API_TOKEN") + monkeypatch.setattr(api, "list_programs", boom) + rep = _json_err(runner.invoke(app, ["programs", "list", "--json"])) + assert rep["ok"] is False + assert "PARAMIFY_API_TOKEN" in rep["errors"][0] + + +@pytest.mark.parametrize("selector", [ + "Alpha Cloud Services", # exact name + "alpha cloud services", # case-insensitive + "Alpha", # unique substring + "aaaa1111-0000-0000-0000-000000000000", # id +]) +def test_resolve_program_accepts_name_or_id(selector): + assert api.resolve_program(_PROGRAMS, selector)["short_name"] == "ACS" + + +def test_resolve_program_rejects_ambiguous_substring(): + """'a' hits all three — resolving it silently would target the wrong program.""" + with pytest.raises(ValueError, match="ambiguous"): + api.resolve_program(_PROGRAMS, "a") + + +def test_resolve_program_rejects_unknown(): + with pytest.raises(LookupError): + api.resolve_program(_PROGRAMS, "Nope") + + +@pytest.mark.parametrize("raw,expected", [ + ("1,3", [0, 2]), + ("1-3", [0, 1, 2]), + ("2 3", [1, 2]), + ("all", [0, 1, 2]), + ("3,1,3", [2, 0]), # de-duped, in the order typed +]) +def test_parse_selection(raw, expected): + from framework.cli import _parse_selection + assert _parse_selection(raw, 3) == expected + + +@pytest.mark.parametrize("raw", ["0", "4", "2-9", "", "nope"]) +def test_parse_selection_rejects_out_of_range(raw): + from framework.cli import _parse_selection + with pytest.raises(ValueError): + _parse_selection(raw, 3) + + +def _platform_cfg(manifest_dict, category="paramify"): + return (manifest_dict["run"].get("platforms") or {}).get(category, {}).get("config", {}) + + +# The shared config `programs target` needs; supplied by default so each test's +# argv shows only what that test is actually varying. +_SHARED_ARGS = ["--cert-uri", "https://example.gov/cpo", "--report-from", "2026-01-01"] + + +def _target(manifest, *args, shared=True): + return runner.invoke(app, [ + "programs", "target", *args, *(_SHARED_ARGS if shared else []), + "-f", str(manifest), "--json", + ]) + + +def test_programs_target_writes_targets_by_name(stub_programs, ver_manifest): + rep = _json(_target(ver_manifest, _VER_FETCHER, + "--program", "Alpha Cloud Services", "--program", "Gamma")) + assert rep["ok"] is True, rep["errors"] + m = api.read_manifest(ver_manifest) + targets = _entry(m, _VER_FETCHER)["targets"] + assert [t["project_id"] for t in targets] == [_PROGRAMS[0]["id"], _PROGRAMS[2]["id"]] + assert [t["program_name"] for t in targets] == ["Alpha Cloud Services", "Gamma Analytics"] + # A target carries ONLY what varies per program. + assert all("cert_package_uri" not in t for t in targets) + + +def test_programs_target_writes_cert_uri_as_category_config(stub_programs, ver_manifest): + """One workspace, one URI: it lands once under platforms.paramify.config.""" + uri = "https://example.gov/cpo?package=abc&v=2" # '=' in the query must survive + # shared=False: _SHARED_ARGS is appended after *args, so its --cert-uri would + # win over this one. + _json(_target(ver_manifest, _VER_FETCHER, "--all", "--cert-uri", uri, + "--report-from", "2026-01-01", shared=False)) + m = api.read_manifest(ver_manifest) + assert _platform_cfg(m)["cert_package_uri"] == uri + assert len(_entry(m, _VER_FETCHER)["targets"]) == len(_PROGRAMS) + + +def test_programs_target_under_json_reuses_existing_cert_uri(stub_programs, ver_manifest): + """Second run with the URI already in the manifest must not need --cert-uri. + + Under --json there is no prompt to fall back on, so if the command still + considered it missing this would fail instead of succeeding. + """ + _json(_target(ver_manifest, _VER_FETCHER, "--program", "Alpha")) + rep = _json(_target(ver_manifest, _VER_FETCHER, "--program", "Beta", shared=False)) + assert rep["ok"] is True, rep["errors"] + m = api.read_manifest(ver_manifest) + assert _platform_cfg(m)["cert_package_uri"] == "https://example.gov/cpo" + assert len(_entry(m, _VER_FETCHER)["targets"]) == 2 + + +def test_programs_target_all_covers_every_program(stub_programs, ver_manifest): + rep = _json(_target(ver_manifest, _VER_FETCHER, "--all")) + assert rep["ok"] is True, rep["errors"] + targets = _entry(api.read_manifest(ver_manifest), _VER_FETCHER)["targets"] + assert [t["project_id"] for t in targets] == [p["id"] for p in _PROGRAMS] + + +def test_programs_target_is_idempotent(stub_programs, ver_manifest): + _json(_target(ver_manifest, _VER_FETCHER, "--program", "Beta")) + _json(_target(ver_manifest, _VER_FETCHER, "--program", "Beta")) # same command again + targets = _entry(api.read_manifest(ver_manifest), _VER_FETCHER)["targets"] + assert len(targets) == 1, "re-targeting the same program must not duplicate it" + + +def test_programs_target_defaults_to_program_taking_entries(stub_programs, ver_manifest): + """No fetcher argument: every manifest entry that takes a program gets it.""" + rep = _json(_target(ver_manifest, "--program", "Beta")) + assert rep["ok"] is True, rep["errors"] + assert _entry(api.read_manifest(ver_manifest), _VER_FETCHER)["targets"] + + +def test_programs_target_requires_cert_uri_under_json(stub_programs, ver_manifest): + """--json can't prompt, so a missing URI must fail loudly and say where it goes.""" + rep = _json_err(_target(ver_manifest, _VER_FETCHER, "--program", "Beta", shared=False)) + assert "cert_package_uri" in rep["errors"][0] + assert "platforms.paramify.config" in rep["errors"][0] + assert "--cert-uri" in rep["errors"][0] + + +def test_programs_target_writes_report_from_as_category_config(stub_programs, ver_manifest): + """report_from is declared per-fetcher but set once at the platform level — + the runner merges platform config over any field a fetcher declares.""" + _json(_target(ver_manifest, _VER_FETCHER, "--all")) + m = api.read_manifest(ver_manifest) + assert _platform_cfg(m)["report_from"] == "2026-01-01" + assert "report_from" not in (_entry(m, _VER_FETCHER).get("config") or {}) + + +@pytest.mark.parametrize("bad", ["Jan 1 2026", "2026-13-45", "01/01/2026", "soon"]) +def test_programs_target_rejects_non_iso_report_from(stub_programs, ver_manifest, bad): + """An unparseable date yields an empty report window, which silently drops + every closed issue — so it has to fail here, not at run time.""" + rep = _json_err(_target(ver_manifest, _VER_FETCHER, "--all", + "--cert-uri", "https://example.gov/cpo", + "--report-from", bad, shared=False)) + assert "report_from" in rep["errors"][0] + assert "ISO" in rep["errors"][0] + + +@pytest.mark.parametrize("good", ["2026-01-01", "2026-01-01T00:00:00Z", "2026-06-30T12:00:00+00:00"]) +def test_programs_target_accepts_iso_report_from(stub_programs, ver_manifest, good): + rep = _json(_target(ver_manifest, _VER_FETCHER, "--all", + "--cert-uri", "https://example.gov/cpo", + "--report-from", good, shared=False)) + assert rep["ok"] is True, rep["errors"] + assert _platform_cfg(api.read_manifest(ver_manifest))["report_from"] == good + + +def test_programs_target_requires_report_from_under_json(stub_programs, ver_manifest): + rep = _json_err(_target(ver_manifest, _VER_FETCHER, "--all", + "--cert-uri", "https://example.gov/cpo", shared=False)) + assert "report_from" in rep["errors"][0] + assert "--report-from" in rep["errors"][0] + + +def test_programs_target_flag_overrides_existing_shared_config(stub_programs, ver_manifest): + """Passing a flag is an override — it applies even when a value is already set.""" + _json(_target(ver_manifest, _VER_FETCHER, "--all", "--cert-uri", + "https://old.example.gov/cpo", "--report-from", "2026-01-01", shared=False)) + _json(_target(ver_manifest, _VER_FETCHER, "--all", "--cert-uri", + "https://new.example.gov/cpo", "--report-from", "2026-04-01", shared=False)) + cfg = _platform_cfg(api.read_manifest(ver_manifest)) + assert cfg["cert_package_uri"] == "https://new.example.gov/cpo" + assert cfg["report_from"] == "2026-04-01" + + +def test_programs_target_requires_a_selection_under_json(stub_programs, ver_manifest): + rep = _json_err(_target(ver_manifest, _VER_FETCHER, shared=False)) + assert "--program" in rep["errors"][0] + + +# --------------------------------------------------------------------------- # +# Shared config is shown and editable on every interactive run — the manifest is +# never a black box you have to open to see what a re-run will carry forward. +# --------------------------------------------------------------------------- # + +@pytest.fixture +def tty(monkeypatch): + """Let the command prompt. CliRunner's stdin isn't a tty, so `_can_prompt` + is patched rather than sys.stdin: click reads the runner's piped `input=` + either way, and this keeps the seam to one function.""" + monkeypatch.setattr(cli, "_can_prompt", lambda json_out: not json_out) + + +def _target_tty(manifest, *args, keys=""): + """Invoke without --json, answering the shared-config prompts with `keys`.""" + return runner.invoke( + app, ["programs", "target", *args, "-f", str(manifest)], input=keys + ) + + +def _seeded(manifest): + """A manifest that already carries both shared values, as a second run finds it.""" + _json(_target(manifest, _VER_FETCHER, "--program", "Alpha")) + return manifest + + +def test_programs_target_shows_shared_config_on_every_run(stub_programs, ver_manifest, tty): + """Values already in the manifest are still displayed — a re-run shows what + it's about to carry forward instead of silently reusing it.""" + result = _target_tty(_seeded(ver_manifest), _VER_FETCHER, "--program", "Beta", keys="\n\n") + assert result.exit_code == 0, result.output + assert "https://example.gov/cpo" in result.output # the URI, as the prompt default + assert "2026-01-01" in result.output # the report start, likewise + assert "set in platforms.paramify" in result.output # ...and where it lives + + +def test_programs_target_enter_keeps_shared_config(stub_programs, ver_manifest, tty): + """Enter at both prompts leaves the platform block byte-identical.""" + before = _platform_cfg(api.read_manifest(_seeded(ver_manifest))) + result = _target_tty(ver_manifest, _VER_FETCHER, "--program", "Beta", keys="\n\n") + assert result.exit_code == 0, result.output + m = api.read_manifest(ver_manifest) + assert _platform_cfg(m) == before + assert len(_entry(m, _VER_FETCHER)["targets"]) == 2, "the run still added its target" + + +def test_programs_target_prompt_updates_shared_config(stub_programs, ver_manifest, tty): + """Typing over the default is how you fix a wrong URI or roll the window.""" + result = _target_tty(_seeded(ver_manifest), _VER_FETCHER, "--program", "Beta", + keys="https://new.example.gov/cpo\n2026-04-01\n") + assert result.exit_code == 0, result.output + cfg = _platform_cfg(api.read_manifest(ver_manifest)) + assert cfg["cert_package_uri"] == "https://new.example.gov/cpo" + assert cfg["report_from"] == "2026-04-01" + + +def test_programs_target_rejects_a_bad_date_typed_at_the_prompt(stub_programs, ver_manifest, tty): + """The ISO check guards the prompt too, and failing writes nothing at all.""" + result = _target_tty(_seeded(ver_manifest), _VER_FETCHER, "--program", "Beta", + keys="\nsoon\n") + assert result.exit_code == 1 + assert "ISO" in result.output + m = api.read_manifest(ver_manifest) + assert _platform_cfg(m)["report_from"] == "2026-01-01" + assert len(_entry(m, _VER_FETCHER)["targets"]) == 1, "no target written on a failed run" + + +def test_programs_target_offers_no_default_when_entries_disagree(stub_programs, ver_manifest, tty): + """Two entries, two different report starts — either one shown as *the* + default would misreport the other, so it says so and asks outright: with no + default, enter re-asks instead of quietly picking a side.""" + other = "paramify_vulnerability_detail_report" + m = api.read_manifest(_seeded(ver_manifest)) + api.add_entry(m, other) + api.set_secret(m, other, "api_token", "PARAMIFY_API_TOKEN") + api.set_fetcher_config(m, other, "report_from", "2025-06-01") # diverges from the platform value + api.dump_manifest(m, ver_manifest, REPO_ROOT) + result = _target_tty(ver_manifest, "--program", "Beta", keys="\n\n2026-05-05\n") + assert result.exit_code == 0, result.output + assert "differs across entries" in result.output + for value in ("[2026-01-01]", "[2025-06-01]"): + assert value not in result.output, "a disputed value must not be offered as the default" + assert _platform_cfg(api.read_manifest(ver_manifest))["report_from"] == "2026-05-05" + + +def test_programs_target_errors_when_no_entry_takes_a_program(stub_programs, tmp_path, in_repo): + path = tmp_path / "empty.yaml" + api.dump_manifest(api.init_manifest(str(tmp_path / "out")), path, in_repo) + rep = _json_err(runner.invoke(app, [ + "programs", "target", "--program", "Beta", "-f", str(path), "--json", + ])) + assert "No manifest entry takes a program" in rep["errors"][0] diff --git a/tests/test_tui_keys.py b/tests/test_tui_keys.py new file mode 100644 index 0000000..22bd81f --- /dev/null +++ b/tests/test_tui_keys.py @@ -0,0 +1,298 @@ +"""TUI key-routing regression tests, driven through Textual's pilot. + +The footer hint bar is a promise: every key it advertises must reach its action +from the focus the app actually lands on. That promise is easy to break silently, +because a page's BINDINGS only fire while focus is *inside* that page — so a +dropped focus, a stolen key, or an unhandled Enter turns a documented shortcut +into a no-op with no error anywhere. These lock in the invariants: + + * each tab focuses a widget inside its own page (so page keys are live) + * pressing the number of the tab you're already on keeps that focus + * ctrl+p belongs to the Paramify page, not Textual's command palette + * enter does something wherever the footer says it does + * enter in a confirm dialog means the safe answer + * a focused text field eats the global keys (documented, not fixed — esc is + the way out, which is why the footer lists it) + +Written sync (asyncio.run per test) so the suite needs no async pytest plugin. +""" + +from __future__ import annotations + +import asyncio +import json +from pathlib import Path + +import pytest + +pytest.importorskip("textual", reason="TUI tests need the 'tui' extra") + +from textual.widgets import DataTable, Input, TabbedContent # noqa: E402 + +from framework import api # noqa: E402 +from framework.tui.app import FetcherApp # noqa: E402 +from framework.tui.modals import ConfirmModal, MultiPickerModal # noqa: E402 +from framework.tui.screens.manifest import ManifestPage # noqa: E402 +from framework.tui.screens.upload import UploadPage # noqa: E402 + +REPO_ROOT = Path(__file__).resolve().parents[1] +SIZE = (180, 50) + + +def _write_manifest(tmp_path: Path, fetchers: int = 1) -> Path: + """A real, schema-valid manifest with `fetchers` discovered entries and its + evidence under tmp_path (nothing here touches the repo's own evidence/).""" + catalog = api.catalog(REPO_ROOT) + names = [f["name"] for c in catalog["categories"] for f in c["fetchers"]][:fetchers] + assert names, "no fetchers discovered — cannot build a test manifest" + manifest = api.init_manifest() + api.set_output_dir(manifest, str(tmp_path / "evidence")) + for name in names: + api.add_entry(manifest, name) + path = tmp_path / "keys-test.yaml" + api.dump_manifest(manifest, path, REPO_ROOT) + return path + + +def _fake_run(tmp_path: Path, *, files: int = 1) -> None: + """Plant one completed run under the manifest's output dir, as api.list_runs + expects to find it (metadata + the output files its invocations name).""" + run_dir = tmp_path / "evidence" / "run-2026-07-30T00-00-00Z" + run_dir.mkdir(parents=True) + outputs = [f"evidence_{i}.json" for i in range(files)] + for name in outputs: + (run_dir / name).write_text(json.dumps({"payload": {"ok": True}})) + (run_dir / "_run_metadata.json").write_text( + json.dumps({ + "started_at": "2026-07-30T00:00:00Z", + "completed_at": "2026-07-30T00:00:10Z", + "invocations": [ + {"fetcher_name": "test_fetcher", "exit_code": 0, "outputs": outputs} + ], + }) + ) + + +def _run(coro_fn, manifest: Path): + """Boot the app on `manifest` and hand (app, pilot) to an async callback.""" + + async def main(): + app = FetcherApp(manifest_path=str(manifest), root_override=str(REPO_ROOT)) + async with app.run_test(size=SIZE) as pilot: + await pilot.pause() + return await coro_fn(app, pilot) + + return asyncio.run(main()) + + +def _focus_id(app) -> str | None: + return None if app.focused is None else app.focused.id + + +# --------------------------------------------------------------------------- # +# focus: every tab must land inside its own page, or its BINDINGS are dead +# --------------------------------------------------------------------------- # + +def test_each_tab_focuses_a_widget_in_its_own_page(tmp_path): + manifest = _write_manifest(tmp_path) + + async def body(app, pilot): + landed = {} + for key, tab in zip("12345", app.screen.TAB_IDS): + await pilot.press(key) + await pilot.pause() + assert app.screen.query_one(TabbedContent).active == tab + assert app.focused is not None, f"tab {tab} left focus cleared" + landed[tab] = _focus_id(app) + # the focused widget must live inside the active pane, so the page's + # own bindings (a/e/x, ctrl+r, ...) resolve + pane = app.screen.query_one(TabbedContent).active_pane + assert app.focused in pane.walk_children(), f"{tab} focused outside its pane" + return landed + + landed = _run(body, manifest) + assert landed == { + "tab-catalog": "catalog-tree", + "tab-manifest": "manifest-entries", + "tab-run": "btn-run", + "tab-evidence": "evidence-runs", + "tab-upload": "scripts-preview", + } + + +def test_repeat_tab_press_keeps_pane_focus(tmp_path): + """Pressing the number of the tab you're on used to clear focus, killing + every page binding until you pressed escape or another tab.""" + manifest = _write_manifest(tmp_path) + + async def body(app, pilot): + await pilot.press("2") + await pilot.pause() + await pilot.press("2") + await pilot.pause() + assert _focus_id(app) == "manifest-entries" + # and a page-level binding still resolves + await pilot.press("a") + await pilot.pause() + return isinstance(app.screen, MultiPickerModal) + + assert _run(body, manifest) is True + + +def test_escape_returns_focus_to_the_pane_default(tmp_path): + manifest = _write_manifest(tmp_path) + + async def body(app, pilot): + await pilot.press("2") + await pilot.pause() + app.screen.query_one("#manifest-output-dir", Input).focus() + await pilot.pause() + await pilot.press("escape") + await pilot.pause() + return _focus_id(app) + + assert _run(body, manifest) == "manifest-entries" + + +# --------------------------------------------------------------------------- # +# bindings: keys the footer advertises must reach their action +# --------------------------------------------------------------------------- # + +def test_preview_keys_are_ours_not_the_command_palette(tmp_path, monkeypatch): + """Textual claims ctrl+p for its command palette as a priority binding, which + outranks the focused widget — ENABLE_COMMAND_PALETTE=False gives it back.""" + manifest = _write_manifest(tmp_path) + calls = [] + monkeypatch.setattr(UploadPage, "action_preview_scripts", lambda self: calls.append(1)) + + async def body(app, pilot): + await pilot.press("5") + await pilot.pause() + await pilot.press("ctrl+p") + await pilot.pause() + await pilot.press("p") + await pilot.pause() + # no palette overlay was pushed over the workspace + return len(calls), [type(s).__name__ for s in app.screen_stack] + + count, stack = _run(body, manifest) + assert count == 2, "ctrl+p and p should both reach the page's preview action" + assert stack == ["Screen", "WorkspaceScreen"] + + +def test_enter_on_a_manifest_row_opens_the_editor(tmp_path, monkeypatch): + manifest = _write_manifest(tmp_path) + calls = [] + monkeypatch.setattr(ManifestPage, "action_edit_entry", lambda self: calls.append(1)) + + async def body(app, pilot): + await pilot.press("2") + await pilot.pause() + assert app.screen.query_one("#manifest-entries", DataTable).row_count == 1 + await pilot.press("enter") + await pilot.pause() + return len(calls) + + assert _run(body, manifest) == 1 + + +def test_enter_on_a_run_drills_into_its_files(tmp_path): + manifest = _write_manifest(tmp_path) + _fake_run(tmp_path, files=2) + + async def body(app, pilot): + await pilot.press("4") + await pilot.pause() + assert _focus_id(app) == "evidence-runs" + assert app.screen.query_one("#evidence-files", DataTable).row_count == 2 + await pilot.press("enter") + await pilot.pause() + return _focus_id(app) + + assert _run(body, manifest) == "evidence-files" + + +# --------------------------------------------------------------------------- # +# the Input trap: documented behaviour, asserted so it can't drift silently +# --------------------------------------------------------------------------- # + +def test_a_focused_field_swallows_the_global_keys(tmp_path): + """Every printable global (1-5, m, q, /) types into a focused Input instead + of firing. Left as-is deliberately — priority bindings would make the filter + boxes untypeable — which is why the footer advertises esc.""" + manifest = _write_manifest(tmp_path) + + async def body(app, pilot): + await pilot.press("2") + await pilot.pause() + field = app.screen.query_one("#manifest-output-dir", Input) + field.focus() + await pilot.pause() + await pilot.press("3", "q") + await pilot.pause() + return app.screen.query_one(TabbedContent).active, field.value, app.is_running + + tab, value, running = _run(body, manifest) + assert tab == "tab-manifest", "a global tab key fired from inside a text field" + assert "3" in value and "q" in value + assert running, "'q' quit the app from inside a text field" + + +def test_output_dir_survives_focus_and_commits_on_blur(tmp_path): + """select_on_focus off (the first keystroke no longer wipes the path), and an + edit that was never submitted is committed on blur instead of reverted.""" + manifest = _write_manifest(tmp_path) + + async def body(app, pilot): + await pilot.press("2") + await pilot.pause() + field = app.screen.query_one("#manifest-output-dir", Input) + original = field.value + field.focus() + await pilot.pause() + await pilot.press("x") + after_one_key = field.value + # leave the field without pressing enter + app.screen.query_one("#manifest-entries", DataTable).focus() + await pilot.pause() + committed = (app.manifest.get("run") or {}).get("output_dir") + return original, after_one_key, committed + + original, after_one_key, committed = _run(body, manifest) + assert original and original in after_one_key, "focus+keystroke replaced the whole path" + assert committed == after_one_key, "an unsubmitted edit was lost on blur" + + +# --------------------------------------------------------------------------- # +# safety: enter must not mean "yes, delete it" +# --------------------------------------------------------------------------- # + +def test_confirm_modal_enter_means_no(tmp_path): + manifest = _write_manifest(tmp_path) + + async def body(app, pilot): + results = [] + app.push_screen(ConfirmModal("Remove 'x' from the manifest?"), results.append) + await pilot.pause() + focused = _focus_id(app) + await pilot.press("enter") + await pilot.pause() + return focused, results + + focused, results = _run(body, manifest) + assert focused == "no" + assert results == [False] + + +def test_confirm_modal_y_still_confirms(tmp_path): + manifest = _write_manifest(tmp_path) + + async def body(app, pilot): + results = [] + app.push_screen(ConfirmModal("Remove 'x' from the manifest?"), results.append) + await pilot.pause() + await pilot.press("y") + await pilot.pause() + return results + + assert _run(body, manifest) == [True] diff --git a/tests/test_ver_timestamps.py b/tests/test_ver_timestamps.py new file mode 100644 index 0000000..e7e0599 --- /dev/null +++ b/tests/test_ver_timestamps.py @@ -0,0 +1,108 @@ +"""Every instant the Paramify VER reports emit is UTC, second precision, Z. + + 2026-07-30T09:00:00Z + +The reports mix three timestamp sources — values the fetcher generates, values +passed through from the Paramify API (which returns milliseconds), and report +bounds supplied as config (which may be a bare date). Without normalization one +document carried all three notations. These tests pin the single format. + +ver_common lives under fetchers/ and is loaded by path: fetchers are scripts the +runner exec's, not an importable package, so there is no `from fetchers...` +import to make. This mirrors how the runner puts _shared on sys.path. + +Run: ``pytest tests/test_ver_timestamps.py`` +""" + +from __future__ import annotations + +import importlib.util +import json +import re +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parent.parent +_VER_COMMON = REPO_ROOT / "fetchers" / "paramify" / "_shared" / "ver_common.py" + +# The one accepted shape. Anchored: a trailing offset or fractional seconds fails. +CANONICAL = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$") +# Anything date-shaped, so an off-format value is found rather than skipped. +LOOSE = re.compile(r"\d{4}-\d{2}-\d{2}[T ][\d:.]+(?:Z|[+-]\d{2}:\d{2})?") + + +def _load_ver_common(): + spec = importlib.util.spec_from_file_location("ver_common_under_test", _VER_COMMON) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +vc = _load_ver_common() + + +@pytest.mark.parametrize("raw,expected", [ + ("2026-02-01T00:00:00.000Z", "2026-02-01T00:00:00Z"), # API millisecond form + ("2026-02-01T00:00:00.123456Z", "2026-02-01T00:00:00Z"), # microseconds + ("2026-02-01T00:00:00Z", "2026-02-01T00:00:00Z"), # already canonical + ("2026-02-01T09:00:00+02:00", "2026-02-01T07:00:00Z"), # offset -> UTC + ("2026-02-01T00:00:00-05:00", "2026-02-01T05:00:00Z"), + ("2026-02-01", "2026-02-01T00:00:00Z"), # bare date + ("2026-02-01T00:00:00", "2026-02-01T00:00:00Z"), # naive == UTC +]) +def test_to_utc_z_normalizes(raw, expected): + assert vc.to_utc_z(raw) == expected + assert CANONICAL.match(vc.to_utc_z(raw)) + + +@pytest.mark.parametrize("raw", ["not a date", "", None, "2026-13-45"]) +def test_to_utc_z_passes_through_what_it_cannot_parse(raw): + """Better an off-format value than a silently dropped one — schema + verification is the right place for a malformed source value to surface.""" + assert vc.to_utc_z(raw) == raw + + +def test_report_period_bounds_date_only_end_covers_the_whole_day(): + """A date-only end means "through the end of that day" to the coverage + filter, so reporting its midnight would understate the period by a day.""" + assert vc.report_period_bounds("2026-01-01", "2026-06-30") == ( + "2026-01-01T00:00:00Z", "2026-06-30T23:59:59Z", + ) + + +def test_report_period_bounds_timestamped_end_is_echoed(): + assert vc.report_period_bounds("2026-01-01", "2026-07-30T15:23:28Z") == ( + "2026-01-01T00:00:00Z", "2026-07-30T15:23:28Z", + ) + + +def test_report_period_bounds_normalizes_both_ends(): + for bound in vc.report_period_bounds("2026-01-01T00:00:00.000Z", "2026-06-30T12:00:00+02:00"): + assert CANONICAL.match(bound), bound + + +def test_current_timestamp_is_canonical(): + assert CANONICAL.match(vc.current_timestamp()) + + +def test_vulnerability_detail_emits_only_canonical_timestamps(): + """The whole mapped object, from an issue whose every date is off-format — + including the free-text overdue explanation, which interpolates a dueDate.""" + issue = { + "id": "x", "poamId": "V-1", "status": "OPEN", "level": "HIGH", + "createdAt": "2026-02-01T00:00:00.000Z", + "evaluationDate": "2025-01-01T08:30:00.123Z", + "dueDate": "2020-07-01T00:00:00.000Z", # long past => overdue, so the + "description": "messy timestamps", # explanation is populated + "deviations": [], + } + detail = vc.map_vulnerability_detail(issue) + + assert detail["overdueStatus"]["isOverdue"] is True, "fixture must exercise the explanation" + # Serialize and scan: catches timestamps in free text (the overdue + # explanation) as well as in fields, without a bespoke tree walker. + found = LOOSE.findall(json.dumps(detail)) + assert found, "no timestamps found — the fixture is broken" + off_format = [ts for ts in found if not CANONICAL.match(ts)] + assert not off_format, f"off-format timestamps in {json.dumps(detail)}: {off_format}" diff --git a/uploaders/paramify_evidence/uploader.py b/uploaders/paramify_evidence/uploader.py index 2dcee6c..7a1a632 100644 --- a/uploaders/paramify_evidence/uploader.py +++ b/uploaders/paramify_evidence/uploader.py @@ -167,7 +167,11 @@ def resolve_evidence_set(metadata: Dict, overrides: Dict) -> Optional[Dict]: # Target fields preferred as the single identifying suffix in an artifact title. -_TITLE_KEYS = ("project_id", "name", "id", "region", "cluster", "host", "bucket", "account_id") +# program_name leads: where a target carries both a readable label and an opaque +# id (Paramify programs), the reviewer picking among artifacts needs the label. +# Fetchers whose id IS readable (gitlab's group/project) declare no program_name, +# so they fall through to project_id exactly as before. +_TITLE_KEYS = ("program_name", "project_id", "name", "id", "region", "cluster", "host", "bucket", "account_id") def build_artifact_meta(metadata: Dict, es_name: str) -> Dict: