# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
"""Generate a CVE 5.x JSON record from an airflow-s/airflow-s issue.

This script reads a tracking issue via ``gh issue view``, parses the
standard template fields in the issue body, and emits a JSON document
that can be pasted straight into the Vulnogram ``#source`` tab of the ASF
CVE tool at

    https://cveprocess.apache.org/cve5/<CVE-ID>#source

The output shape is modelled on a real CVE record exported from
Vulnogram so that a paste-and-save round-trip does not mangle the
record. In particular, the script produces:

* ``affected[]`` entries with ``vendor``, ``product``, ``collectionURL``
  (PyPI), ``packageName`` (``apache-airflow``) and a semver range;
* a ``descriptions[]`` entry with both a plain ``value`` and an HTML
  ``supportingMedia`` alternative (Vulnogram's WYSIWYG mode expects
  both);
* ``metrics[].other`` with ``type`` = *"Textual description of severity"*
  and ``content.text`` = the severity word;
* ``problemTypes[].descriptions[]`` with ``cweId``, ``description`` and
  ``type`` = *"CWE"*;
* ``credits[]`` entries from the *Reporter credited as* field
  (``type: "finder"`` by default; ``type: "tool"`` when the credit
  matches the bot/AI policy in
  ``tools/cve-tool-vulnogram/bot-credits-policy.md`` — Dependabot, Renovate,
  Snyk, scanners, ``*-bot`` / ``*[bot]`` handles, etc.) and the
  *Remediation developer* field (``type: "remediation developer"``)
  — both newline-separated, with the ``Full Name, Affiliation``
  pattern preserved as one credit; the ``--remediation-developer``
  CLI flag still works as an additional / override mechanism on top
  of the body field;
* ``references[]`` with automatic ``tags`` (``patch`` for
  ``github.com/.../pull/...`` URLs, ``vendor-advisory`` for
  ``lists.apache.org``/``security.apache.org``). URLs from the
  tracking-issue's *Public advisory URL* body field are picked up
  automatically and tagged as ``vendor-advisory``; URLs from the
  *Security mailing list thread* field are **deliberately not** exported
  because that field holds the internal ``security@airflow.apache.org``
  thread URL, which 404s for anyone outside the security team;
* ``providerMetadata.orgId`` and ``cveMetadata.assignerOrgId`` set to
  the ASF org id;
* an ``x_generator`` annotation so the origin of the JSON is visible.

The output is **deterministic**: the same issue body and the same CLI
flags always produce the same JSON bytes. Keys are sorted, references
are sorted alphabetically, credits preserve the order of first
occurrence, and no timestamps or machine-dependent data are included.

Usage::

    # Read the issue from GitHub and write the JSON to stdout:
    uv run --project tools/cve-tool-vulnogram/generate-cve-json generate-cve-json 232

    # Write to a file and show the Vulnogram paste URL:
    uv run --project tools/cve-tool-vulnogram/generate-cve-json generate-cve-json 232 \\
        --output /tmp/CVE-2026-40913.json

    # Attach an explicit "remediation developer" credit:
    uv run --project tools/cve-tool-vulnogram/generate-cve-json generate-cve-json 232 \\
        --remediation-developer "Kevin Yang"

    # Test mode: read a prepared issue body from stdin so you can
    # iterate on the parser without touching the network:
    cat /tmp/issue232-body.md | \\
        uv run --project tools/cve-tool-vulnogram/generate-cve-json generate-cve-json \\
            --stdin --cve-id CVE-2026-40913 \\
            --title "Apache Airflow: ..."

The runtime uses only the Python standard library; the ``pyproject.toml``
declares no runtime dependencies, so ``uv run`` resolves the environment
in milliseconds. Dev tooling (pytest, ruff, mypy) lives in the ``dev``
dependency group.
"""

from __future__ import annotations

import argparse
import contextlib
import html
import json
import os
import re
import subprocess
import sys
import tempfile

# -----------------------------------------------------------------------------
# Configuration loading.
# -----------------------------------------------------------------------------
#
# All project-specific values (vendor, top-level product/package, project
# display map, package-name regex, CNA org id, generator tag, …) are read
# at module load time from a TOML config file the adopting project ships
# in its tracker repo at:
#
#   <project-config>/tools/cve-tool-vulnogram/cve-json-config.toml
#
# (where `<project-config>` is the adopting project's
# `.apache-magpie-overrides/` directory, per the apache/magpie
# placeholder convention).
#
# Resolution order for the config path:
#   1. The `--config <path>` CLI flag (or the `config_path=` argument to
#      `_load_config()`).
#   2. The `CVE_JSON_CONFIG` environment variable.
#   3. `<cwd>/.apache-magpie-overrides/tools/cve-tool-vulnogram/cve-json-config.toml`.
#      Back-compat: if only the pre-rename `tools/vulnogram/` copy exists at
#      this default location, it is used with a deprecation warning.
#
# Schema: see the README in this package.
import tomllib
import urllib.parse
from pathlib import Path

_DEFAULT_CONFIG_RELPATH = ".apache-magpie-overrides/tools/cve-tool-vulnogram/cve-json-config.toml"
# Pre-rename location (`tools/vulnogram` was renamed to `tools/cve-tool-vulnogram`).
# Used as a fallback for adopters who upgraded across the rename — see `_load_config`.
_LEGACY_CONFIG_RELPATH = ".apache-magpie-overrides/tools/vulnogram/cve-json-config.toml"
_CONFIG_PATH_ENV = "CVE_JSON_CONFIG"

# Cached, lazily-loaded config. `_populate_constants()` reads this and
# fills the module globals below; tests that want to use a different
# config can call `_set_config_path(path)` before calling any
# generator function.
_CONFIG: dict | None = None


def _resolve_config_path(config_path: Path | str | None = None) -> tuple[Path, bool]:
    """Resolve the config file path following the documented order.

    Returns ``(path, is_default)`` — ``is_default`` is True only when the path
    came from the default location (no ``--config``, no ``$CVE_JSON_CONFIG``),
    the single case the legacy-path fallback in ``_load_config`` applies to.
    Keeping the precedence decision here means the fallback keys off one source
    of truth instead of re-deriving it.
    """
    if config_path is not None:
        return Path(config_path), False
    env = os.environ.get(_CONFIG_PATH_ENV)
    if env:
        return Path(env), False
    return Path.cwd() / _DEFAULT_CONFIG_RELPATH, True


def _load_config(config_path: Path | str | None = None) -> dict:
    """Load TOML config, raising ``FileNotFoundError`` with usage hints
    if it cannot be located.
    """
    path, is_default = _resolve_config_path(config_path)
    if not path.exists():
        # Back-compat: `tools/vulnogram` was renamed to `tools/cve-tool-vulnogram`.
        # An adopter who upgraded across the rename still keeps their committed
        # config at the pre-rename path. When resolving from the default location
        # (no `--config`, no `$CVE_JSON_CONFIG`) and only the legacy copy exists,
        # use it and warn — so the upgrade doesn't hard-break mid-run.
        legacy = Path.cwd() / _LEGACY_CONFIG_RELPATH
        if is_default and legacy.is_file():
            print(
                f"generate-cve-json: WARNING reading config from the legacy path "
                f"{_LEGACY_CONFIG_RELPATH}. The tool dir was renamed "
                f"`tools/vulnogram` -> `tools/cve-tool-vulnogram`; move the file to "
                f"{_DEFAULT_CONFIG_RELPATH} to silence this warning.",
                file=sys.stderr,
            )
            path = legacy
        else:
            raise FileNotFoundError(
                f"generate-cve-json: config file not found at {path}.\n"
                f"  Pass --config <path>, set ${_CONFIG_PATH_ENV}, or place a config\n"
                f"  at {_DEFAULT_CONFIG_RELPATH} relative to the cwd.\n"
                f"  Schema: see the generate-cve-json package README."
            )
    return tomllib.loads(path.read_text(encoding="utf-8"))


def _set_config_path(config_path: Path | str) -> None:
    """Set / override the config path (used by `--config` CLI and tests)."""
    global _CONFIG
    _CONFIG = _load_config(config_path)
    _populate_constants()


def _populate_constants() -> None:
    """(Re)populate module globals from the loaded config."""
    global _CONFIG
    if _CONFIG is None:
        _CONFIG = _load_config()
    cfg = _CONFIG

    global DEFAULT_REPO, DEFAULT_VENDOR, DEFAULT_PRODUCT
    global DEFAULT_PACKAGE_NAME, DEFAULT_COLLECTION_URL
    global DEFAULT_ASF_ORG_ID, GENERATOR_TAG, SKILL_SOURCE_URL
    global PROJECT_DISPLAY_MAP, PACKAGE_RE
    global TOP_LEVEL_NAME, TOP_LEVEL_PRODUCT, PROJECT_PRODUCT_TEMPLATE
    global CNA_PRIVATE_PROJECT_URL, CNA_PRIVATE_OWNER, CNA_PRIVATE_USERS_LIST
    global TITLE_STRIP_RE, TRACKER_FILTER_TOKEN
    global RELEASE_VOTE_GATING, RC_VOTING_LABEL, FORWARD_STATE_LABELS

    DEFAULT_REPO = cfg["meta"]["tracker_repo"]
    DEFAULT_VENDOR = cfg["product"]["vendor"]
    DEFAULT_PRODUCT = cfg["product"]["default_product"]
    DEFAULT_PACKAGE_NAME = cfg["product"]["default_package_name"]
    DEFAULT_COLLECTION_URL = cfg["product"]["default_collection_url"]
    DEFAULT_ASF_ORG_ID = cfg["cna"]["org_id"]
    GENERATOR_TAG = cfg["meta"]["generator_tag"]
    SKILL_SOURCE_URL = cfg["meta"].get(
        "skill_source_url", f"https://github.com/{cfg['meta']['tracker_repo']}"
    )
    PROJECT_DISPLAY_MAP = dict(cfg["packages"]["project_display_map"])
    PACKAGE_RE = re.compile(cfg["packages"]["package_pattern"])
    TOP_LEVEL_NAME = cfg["packages"]["top_level_name"]
    TOP_LEVEL_PRODUCT = cfg["packages"]["top_level_product"]
    PROJECT_PRODUCT_TEMPLATE = cfg["packages"]["project_product_template"]

    # Per-project CVE 5.x `CNA_private` envelope fields. These end up
    # in every CVE record this tool generates for the project, and
    # used to be hardcoded to Airflow values — now config-driven so
    # adopters can populate them with their project's URLs / lists.
    cna_private_cfg = cfg.get("cna_private", {})
    CNA_PRIVATE_PROJECT_URL = cna_private_cfg.get("project_url", "")
    CNA_PRIVATE_OWNER = cna_private_cfg.get("owner", "")
    CNA_PRIVATE_USERS_LIST = cna_private_cfg.get("users_list", "")

    # Workflow gating around the DRAFT → REVIEW transition. The ASF CVE
    # tool's REVIEW state means "this record is about to be published"
    # — so for ASF projects with a release-vote cadence, advancing to
    # REVIEW only makes sense once an RC is being voted on. Non-ASF
    # adopters (which often don't have a separate release-vote step
    # before publishing the advisory) keep the original behavior: a
    # fully-populated record auto-advances to REVIEW.
    #
    # Opt in by setting `[workflow].release_vote_gating = true` and the
    # tracker label name in `[workflow].rc_voting_label`. The sync skill
    # is responsible for detecting active [VOTE] threads on the project's
    # dev mailing list (e.g. `dev@<project>.apache.org`) and proposing
    # the label add/remove; the generator only reads the label.
    workflow_cfg = cfg.get("workflow", {})
    RELEASE_VOTE_GATING = bool(workflow_cfg.get("release_vote_gating", False))
    RC_VOTING_LABEL = workflow_cfg.get("rc_voting_label", "rc voting")
    # Forward-state labels — when the tracker carries any of these the
    # release has already shipped (the vote, if there was one, passed) and
    # the rc-voting gate is moot. Treat them as "vote completed", which
    # means a ready CNA stays at REVIEW (or advances to PUBLIC on
    # vendor-advisory) regardless of whether the `rc voting` label is
    # still on the tracker. Without this, removing `rc voting` at the
    # `pr merged → fix released` transition (per the sync skill) walks
    # the state back to DRAFT — the wrong direction for a record that's
    # about to be published.
    FORWARD_STATE_LABELS = set(
        workflow_cfg.get(
            "forward_state_labels",
            ["fix released", "announced - emails sent", "announced", "vendor-advisory ready"],
        )
    )

    # Title-strip regex for `resolve_title` — built from the configured
    # top-level product so the *"<vendor>: <product>:"* prefix is
    # stripped without hardcoding a specific project's name.
    _stripped_product = re.escape(TOP_LEVEL_PRODUCT.strip())
    TITLE_STRIP_RE = re.compile(
        rf"^\s*{_stripped_product}\s*[:\-–—]?\s*",  # noqa: RUF001 — en-/em-dash deliberate
        re.IGNORECASE,
    )

    # Substring used by `build_references` to filter out tracker URLs
    # before serialising. We use the tracker repo's owner segment
    # (the part before `/`) — distinctive enough for any URL that
    # points at the tracker (issues, PRs, raw blobs all carry it)
    # without false positives on URLs that mention the project name
    # in a path component.
    TRACKER_FILTER_TOKEN = DEFAULT_REPO.split("/", 1)[0]


# Module-level constants — populated by `_populate_constants()`. The
# initial values are placeholders; the call below populates them at
# import time. Tests can override by calling `_set_config_path()`
# before importing any generator function.
DEFAULT_REPO: str = ""
DEFAULT_VENDOR: str = ""
DEFAULT_PRODUCT: str = ""
DEFAULT_PACKAGE_NAME: str = ""
DEFAULT_COLLECTION_URL: str = ""
DEFAULT_ASF_ORG_ID: str = ""
GENERATOR_TAG: str = ""
SKILL_SOURCE_URL: str = ""
PROJECT_DISPLAY_MAP: dict[str, str] = {}
PACKAGE_RE: re.Pattern[str] = re.compile("")
TOP_LEVEL_NAME: str = ""
TOP_LEVEL_PRODUCT: str = ""
PROJECT_PRODUCT_TEMPLATE: str = ""
CNA_PRIVATE_PROJECT_URL: str = ""
CNA_PRIVATE_OWNER: str = ""
CNA_PRIVATE_USERS_LIST: str = ""
TITLE_STRIP_RE: re.Pattern[str] = re.compile("")
TRACKER_FILTER_TOKEN: str = ""
RELEASE_VOTE_GATING: bool = False
RC_VOTING_LABEL: str = ""
FORWARD_STATE_LABELS: set[str] = set()

# CVE 5.x convention values that are not project-specific.
DEFAULT_CREDIT_TYPE = "finder"
TOOL_CREDIT_TYPE = "tool"
DEFAULT_LANG = "en"
DEFAULT_DISCOVERY = "UNKNOWN"

# Bot / AI / automation detection for the *Reporter credited as*
# field. When a credit row matches any of these rules, the CVE 5.x
# ``credits[]`` entry is emitted with ``type: "tool"`` instead of
# ``type: "finder"``. The matching rules mirror
# ``tools/cve-tool-vulnogram/bot-credits-policy.md`` (single source of truth);
# update both together. Detection is intentionally broad — false
# positives are cheap (the user can edit the JSON afterwards or set
# the credit name to a clearer human-style string), false negatives
# put a bot handle into the ``finder`` credit class in a public CVE
# record, which is what this policy exists to prevent.
#
# Known bot / automation names matched as a case-insensitive
# whole-word against the credit string (e.g. ``"Dependabot"``,
# ``"discovered by Automated Scanner v3"``).
BOT_CREDIT_KNOWN_NAMES: tuple[str, ...] = (
    "dependabot",
    "renovate",
    "snyk-bot",
    "snyk",
    "copilot",
    "ghsa-probot",
    "github-actions",
    "mend-bot",
    "mend",
    "whitesource",
    "sonatype-lift",
    "lift-bot",
    "codecov",
    "mergify",
    "mergifyio",
    "allcontributors",
    "fossabot",
    "imgbotapp",
    "pre-commit-ci",
    "claude",
    "chatgpt",
    "gpt-bot",
    "gpt",
    "anthropic",
    "openai",
    "automated",
    "automation",
    "scanner",
    "auto-scanner",
    "vulnerability-scanner",
    "security-scanner",
    "sast",
    "dast",
)

# Handle-shaped patterns matched as case-insensitive regexes against
# the credit string. ``\b`` anchors keep them from firing on
# unrelated human names (``Joe Bot`` does not match ``*-bot`` because
# the space breaks the word boundary, but ``joe-bot`` does).
BOT_CREDIT_PATTERNS: tuple[re.Pattern[str], ...] = (
    re.compile(r"\b[\w]*-bot\b", re.IGNORECASE),
    re.compile(r"\b[\w]+bot\b", re.IGNORECASE),
    re.compile(r"\b[\w]*-ai\b", re.IGNORECASE),
    re.compile(r"\b[\w-]+\.ai\b", re.IGNORECASE),
    re.compile(r"\b[\w]*-agent\b", re.IGNORECASE),
    re.compile(r"\b[\w]*-gpt\b", re.IGNORECASE),
    re.compile(r"\b[\w]*scanner[\w]*\b", re.IGNORECASE),
    re.compile(r"\b[\w]*automat[\w]*\b", re.IGNORECASE),
)


def is_bot_credit(name: str) -> bool:
    """Return True if *name* matches the bot/AI credit policy.

    Matches any of three rules:

    1. Literal GitHub ``[bot]`` suffix — handle ends in ``[bot]``
       (e.g. ``dependabot[bot]``).
    2. Known-bot / automation-name list — case-insensitive whole-word
       occurrence of any of ``BOT_CREDIT_KNOWN_NAMES`` in the credit
       string.
    3. Suffix / contains-pattern handles — any of
       ``BOT_CREDIT_PATTERNS`` matches.

    See ``tools/cve-tool-vulnogram/bot-credits-policy.md`` for the canonical
    rule set and rationale.
    """
    cleaned = name.strip()
    if not cleaned:
        return False
    if cleaned.endswith("[bot]"):
        return True
    lowered = cleaned.lower()
    for known in BOT_CREDIT_KNOWN_NAMES:
        if re.search(rf"\b{re.escape(known)}\b", lowered):
            return True
    return any(pattern.search(cleaned) for pattern in BOT_CREDIT_PATTERNS)


# Populate at import time. If the config is not present, defer the
# error to first actual use so test harnesses can call
# `_set_config_path()` before exercising the module.
with contextlib.suppress(FileNotFoundError):
    _populate_constants()

NO_RESPONSE = "_No response_"

# Sentinel token for an as-yet-unreleased upper bound in
# ``Affected versions`` entries. Used predominantly for projects
# trackers where the wave milestone (date-based, e.g. ``Projects
# 2026-04-21``) does not reveal which exact project package version
# will ship the fix. The token is stripped before parsing and the
# resulting CVE 5.x ``versions[]`` entry omits ``lessThan`` —
# downstream consumers (Vulnogram, cve.org) read that as
# *"affected from <low> onwards, no fix released yet"*. Once the
# release ships and the version is known the sync skill replaces the
# token with the real upper bound (``< X.Y.Z``).
NEXT_VERSION_TOKEN_RE = re.compile(r"<\s*NEXT\s+VERSION\b", re.IGNORECASE)

# Map a CVE ``affected[].collectionURL`` to the canonical project-page
# URL template used to render a clickable per-package link in the
# attachment table. The PyPI ``collectionURL`` is the legacy
# ``https://pypi.python.org`` host, but the canonical project page lives
# on ``pypi.org/project/<name>/``; the table renders that.
# Only collection URLs Airflow actually emits today need entries here —
# unknown URLs fall through to ``None`` and the attachment table
# renders ``—`` for them.
COLLECTION_URL_TO_PROJECT_URL_TEMPLATE: dict[str, str] = {
    "https://pypi.python.org": "https://pypi.org/project/{package}/",
    "https://pypi.org": "https://pypi.org/project/{package}/",
}

# HTML-comment marker prefix used to identify the single "CVE JSON
# attachment" comment on a tracking issue, so ``--attach`` can update
# the existing comment in place instead of posting a duplicate on every
# re-run. The full marker includes the CVE id and a version tag.
ATTACHMENT_MARKER_BEGIN_PREFIX = "<!-- generate-cve-json:"
ATTACHMENT_MARKER_END_PREFIX = "<!-- generate-cve-json:end"
ATTACHMENT_MARKER_VERSION = "v1"
# Back-compat alias — the `v1` comment-based attachments used a single
# marker. New body-embedded attachments use a begin/end pair.
ATTACHMENT_MARKER_PREFIX = ATTACHMENT_MARKER_BEGIN_PREFIX


# -----------------------------------------------------------------------------
# Input: fetch the issue body from GitHub (or read from stdin in test mode).
# -----------------------------------------------------------------------------


def fetch_issue(issue_number: int | str, repo: str) -> tuple[str, str, list[str]]:
    """Return ``(title, body, labels)`` for the given issue.

    Calls ``gh issue view <N> --repo <repo> --json title,body,labels``.
    The ``labels`` list contains label names (strings); the gh JSON
    objects' ``.name`` field is unwrapped here so callers can do a
    simple ``"rc voting" in labels`` check. Raises a ``RuntimeError``
    with the stderr output if ``gh`` is unavailable or the issue does
    not exist.
    """
    try:
        result = subprocess.run(
            [
                "gh",
                "issue",
                "view",
                str(issue_number),
                "--repo",
                repo,
                "--json",
                "title,body,labels",
            ],
            check=True,
            capture_output=True,
            text=True,
        )
    except FileNotFoundError as exc:
        raise RuntimeError(
            "`gh` CLI not found on PATH. Install it from https://cli.github.com/ "
            "or run with --stdin to bypass the network call."
        ) from exc
    except subprocess.CalledProcessError as exc:
        raise RuntimeError(f"`gh issue view {issue_number} --repo {repo}` failed:\n{exc.stderr}") from exc

    data = json.loads(result.stdout)
    labels = [
        label.get("name", "")
        for label in (data.get("labels") or [])
        if isinstance(label, dict) and label.get("name")
    ]
    return data.get("title", ""), data.get("body", ""), labels


# -----------------------------------------------------------------------------
# Parse the issue body into individual template fields.
# -----------------------------------------------------------------------------


def extract_field(body: str, heading: str) -> str:
    """Return the text under ``### <heading>`` up to the next ``###``.

    Trailing whitespace is stripped. The literal placeholder
    ``_No response_`` is normalised to an empty string so callers only
    have to check truthiness.
    """
    pattern = rf"^###\s+{re.escape(heading)}\s*\n+(?P<value>.*?)(?=\n###\s|\Z)"
    match = re.search(pattern, body, re.MULTILINE | re.DOTALL)
    if not match:
        return ""
    value = match.group("value").strip()
    if value == NO_RESPONSE:
        return ""
    return value


def parse_credits_from_field(value: str) -> list[str]:
    """Split the ``Reporter credited as`` field into a list of credit
    names.

    Credits are split on **newlines only**, deliberately *not* on commas:
    the common ``Full Name, Affiliation`` pattern (for example
    ``Jed Cunningham, Astronomer``) is one credit, not two. If you need
    to credit multiple people, put each one on its own line. Bullets
    (``-``, ``*``, ``+``, ``1.``) are stripped. Blank lines are ignored.
    De-duplicated preserving order of first occurrence.
    """
    if not value:
        return []
    names: list[str] = []
    seen: set[str] = set()
    for raw_line in value.splitlines():
        line = raw_line.strip()
        if not line:
            continue
        line = re.sub(r"^(?:[-*+]\s+|\d+[.)]\s+)", "", line)
        line = line.strip(" `\t")
        if line and line not in seen:
            seen.add(line)
            names.append(line)
    return names


def combine_remediation_developers(field_value: str, cli_overrides: list[str]) -> list[str]:
    """Merge body-field developers with CLI overrides; preserve order, drop dupes.

    The body's *Remediation developer* field is the source of truth
    in the normal flow (auto-populated by the sync skill from the
    PR's author). The ``--remediation-developer`` CLI flag is
    additive: any name passed there is appended after the body
    entries unless it already appears in the body. Duplicates are
    dropped silently so a triager can pass ``--remediation-developer
    "Alice"`` without worrying about whether Alice is already in the
    body.
    """
    combined: list[str] = []
    seen: set[str] = set()
    for name in [*parse_credits_from_field(field_value), *cli_overrides]:
        cleaned = name.strip()
        if cleaned and cleaned not in seen:
            combined.append(cleaned)
            seen.add(cleaned)
    return combined


def parse_url_list(value: str) -> list[str]:
    """Return every URL mentioned in ``value`` in order of first
    occurrence.

    Handles bullet lists, plain-text lines and markdown-link syntax
    (``[text](url)``) uniformly. Non-URL tokens are ignored, which lets
    a reviewer annotate the field with notes without breaking parsing.
    """
    if not value:
        return []
    urls: list[str] = []
    seen: set[str] = set()
    for match in re.finditer(r"https?://[^\s)>\]]+", value):
        url = match.group(0).rstrip(").,;>")
        if url not in seen:
            seen.add(url)
            urls.append(url)
    return urls


def parse_cve_id(value: str) -> str:
    """Extract the first ``CVE-YYYY-NNNN+`` token from ``value``."""
    match = re.search(r"CVE-\d{4}-\d{4,}", value)
    return match.group(0) if match else ""


def parse_cwe(value: str) -> tuple[str, str]:
    """Return ``(cweId, human-readable description)`` from a CWE field.

    Accepts ``CWE-285``, ``CWE-285: Improper Authorization``,
    ``CWE-285 (Improper Authorization)``, and free-form text. If no
    ``CWE-\\d+`` token is found, ``cweId`` is an empty string and the
    full value is returned as the description.

    When the title following the id is wrapped in a single layer of
    parentheses or brackets (e.g. ``CWE-285 (Improper Authorization)``
    — the form most projects' CWE pickers serialise to), the outer
    wrapper is stripped before the description is assembled. Without
    the strip the result reads ``"CWE-285: (Improper Authorization)"``
    — both a colon *and* parens, which the ASF CVE-CNA reviewers flag
    as cluttered. The accepted shapes are ``"CWE-285: Title"`` (colon
    separator, no wrapper) or ``"CWE-285 Title"`` (no separator); the
    parser emits the colon-separated form.

    Only the **outer** wrapper is stripped: ``CWE-285 (Foo) Bar``
    stays ``"CWE-285: (Foo) Bar"`` because the parens are not the
    outermost shape.
    """
    value = value.strip()
    id_match = re.search(r"CWE-\d+", value)
    cwe_id = id_match.group(0) if id_match else ""
    if cwe_id:
        # Use "CWE-285: Improper Authorization" as the description when
        # a title follows the id; otherwise just the id.
        title_match = re.match(r"^CWE-\d+\s*[:\-]?\s*(?P<rest>.*)", value)
        rest = (title_match.group("rest") if title_match else "").strip()
        # Strip a single layer of outer wrapping parens / brackets.
        # The inner content must contain no unbalanced wrappers; this
        # avoids stripping ``(Foo) Bar (Baz)`` to ``Foo) Bar (Baz``.
        wrapper_match = re.match(r"^\((?P<paren>[^()]+)\)$|^\[(?P<bracket>[^\[\]]+)\]$", rest)
        if wrapper_match:
            rest = (wrapper_match.group("paren") or wrapper_match.group("bracket")).strip()
        description = f"{cwe_id}: {rest}" if rest else cwe_id
    else:
        description = value
    return cwe_id, description


def parse_affected_versions(value: str, version_start_override: str | None) -> list[dict]:
    """Turn the ``Affected versions`` field into a CVE ``versions``
    array.

    Supports a few common shapes:

    * ``< 3.2.2`` / ``<3.2.2`` → ``{version: "0", lessThan: "3.2.2", ...}``
    * ``>= 2.0.0, < 3.2.2`` → ``{version: "2.0.0", lessThan: "3.2.2", ...}``
    * ``<= 3.2.1`` / ``<=3.2.1`` → ``{version: "0", lessThanOrEqual: "3.2.1", ...}``
    * A bare version like ``3.1.5`` → ``{version: "3.1.5", status: "affected"}``
    * A bare lower bound like ``>=2.0.0`` → ``{version: "2.0.0",
      status: "affected"}`` (no upper bound — useful for trackers
      whose fix-shipped version is not yet known).

    The ``< NEXT VERSION`` sentinel signals "fix not yet released,
    upper bound unknown" — used predominantly on project trackers
    where the wave milestone is date-based and the package version
    that will carry the fix is decided by the release manager during
    the wave. The token is stripped before further parsing; the
    resulting entry has no ``lessThan``. Once the release ships, the
    sync skill replaces ``< NEXT VERSION`` with the actual ``< X.Y.Z``
    upper bound and the next regen produces a fully-bounded entry.
    See the ``NEXT VERSION`` rule in
    ``.claude/skills/security-issue-sync/SKILL.md`` for the lifecycle.

    ``--version-start`` overrides the low bound unconditionally when set,
    because the issue body rarely specifies it and a reviewer usually
    knows the first affected release by heart.

    **Strict mode** — unparsable inputs raise ``ValueError``. The
    prior fallback that emitted ``{"version": <raw string>}`` produced
    invalid CVE 5.x records: the schema's ``version`` field is a
    literal version, never a range expression. Forcing the parser to
    fail loud catches reviewer-typo / parenthetical-comment inputs
    before they ship as malformed JSON to Vulnogram (for example,
    ``>= 3.0.0 (reporter verified...)`` falling through and
    serialising the parenthetical clause into the ``version`` field).

    **Bare lower bounds without an upper bound** emit a warning to
    stderr unless the ``< NEXT VERSION`` sentinel was used. The
    resulting record claims the bare version is the *sole* affected
    release, which is almost always wrong; the warning catches the
    case without blocking emission, since a small number of
    workflows legitimately want bare-version entries (single-point-
    fix records where only one release train is involved).
    """
    raw_value_for_diagnostics = (value or "").strip()
    cleaned = value.strip().strip("`").strip() if value else ""
    low_bound = (version_start_override or "").strip() or "0"

    has_next_version_sentinel = bool(NEXT_VERSION_TOKEN_RE.search(cleaned))
    if has_next_version_sentinel:
        cleaned = NEXT_VERSION_TOKEN_RE.sub("", cleaned).strip().rstrip(",").strip()

    if not cleaned:
        return [
            {
                "status": "affected",
                "version": low_bound,
                "versionType": "semver",
            }
        ]

    range_match = re.search(
        r">=?\s*(?P<low>[0-9][0-9A-Za-z.\-+]*)\s*,?\s*<\s*(?P<high>[0-9][0-9A-Za-z.\-+]*)",
        cleaned,
    )
    if range_match:
        return [
            {
                "status": "affected",
                "version": version_start_override or range_match.group("low"),
                "lessThan": range_match.group("high"),
                "versionType": "semver",
            }
        ]

    less_than_or_equal_match = re.match(r"<=\s*(?P<high>[0-9][0-9A-Za-z.\-+]*)", cleaned)
    if less_than_or_equal_match:
        return [
            {
                "status": "affected",
                "version": low_bound,
                "lessThanOrEqual": less_than_or_equal_match.group("high"),
                "versionType": "semver",
            }
        ]

    less_than_match = re.match(r"<\s*(?P<high>[0-9][0-9A-Za-z.\-+]*)", cleaned)
    if less_than_match:
        return [
            {
                "status": "affected",
                "version": low_bound,
                "lessThan": less_than_match.group("high"),
                "versionType": "semver",
            }
        ]

    single_match = re.match(r"^(?P<v>[0-9][0-9A-Za-z.\-+]*)$", cleaned)
    if single_match:
        return [
            {
                "status": "affected",
                "version": single_match.group("v"),
                "versionType": "semver",
            }
        ]

    ge_only_match = re.match(r">=?\s*(?P<low>[0-9][0-9A-Za-z.\-+]*)\s*$", cleaned)
    if ge_only_match:
        # Warn when the input was a bare lower bound AND no `< NEXT
        # VERSION` sentinel was given. The emitted entry has only a
        # `version` field (no `lessThan`), which CVE 5.x readers
        # interpret as "this version alone is affected" — almost
        # always misleading for a CVE that affects a continuous
        # range. The sentinel signals "fix not yet released, upper
        # bound unknown" and is the right shape for in-flight
        # trackers; without it, the reviewer should add an explicit
        # upper bound.
        if not has_next_version_sentinel:
            print(
                f"warning: `Affected versions` value {raw_value_for_diagnostics!r} "
                f"has a bare lower bound (no upper bound, no `NEXT VERSION` "
                f"sentinel). The emitted record will claim "
                f"{ge_only_match.group('low')!r} alone is affected. Add a "
                f"`< X.Y.Z` upper bound or the `< NEXT VERSION` sentinel "
                f"to silence this warning.",
                file=sys.stderr,
            )
        return [
            {
                "status": "affected",
                "version": version_start_override or ge_only_match.group("low"),
                "versionType": "semver",
            }
        ]

    # Fall-through: nothing matched. Refuse to emit a record whose
    # `version` field is a range expression — that JSON would be
    # rejected by Vulnogram's schema validation anyway, and silently
    # emitting it has shipped malformed records to reviewers (CVE-
    # 2026-46763 — `>= 3.0.0` ended up as a literal `version` string
    # because the parser couldn't strip a trailing parenthetical).
    raise ValueError(
        f"Could not parse `Affected versions` value {raw_value_for_diagnostics!r}. "
        f"Supported shapes: `< X.Y.Z`, `<= X.Y.Z`, `>= X.Y.Z`, "
        f"`>= X.Y.Z, < A.B.C`, and bare `X.Y.Z`. Strip parenthetical "
        f"comments, stray backticks, and markdown wrappers before "
        f"re-running. The `< NEXT VERSION` sentinel may be combined "
        f"with `>= X.Y.Z` when the fix-shipped version is not yet "
        f"known."
    )


def normalise_severity(value: str) -> str:
    """Return the severity as a lower-case word
    (``none`` / ``low`` / ``moderate`` / ``medium`` / ``high`` / ``important`` /
    ``critical``) or the
    original text if it doesn't match the standard set."""
    lowered = value.strip().lower()
    if lowered in {"none", "low", "moderate", "medium", "high", "important", "critical"}:
        return lowered
    return value.strip()


def to_html(text: str) -> str:
    """Convert a plain-text description to the minimal HTML shape that
    Vulnogram's WYSIWYG mode stores in ``supportingMedia``.

    The output is HTML-escaped and paragraph breaks (``\\n\\n``) are
    rendered as ``<br><br>``, single newlines as ``<br>``. No inline
    styles or span wrappers are emitted -- those are editor cruft that
    appears after a save round-trip and is not needed on paste-in.
    """
    escaped = html.escape(text)
    escaped = escaped.replace("\r\n", "\n").replace("\n\n", "<br><br>")
    escaped = escaped.replace("\n", "<br>")
    return escaped


# -----------------------------------------------------------------------------
# Reference tagging.
# -----------------------------------------------------------------------------


def classify_reference(url: str) -> list[str]:
    """Return the CVE reference ``tags`` for ``url``.

    * ``github.com/.../pull/N`` and ``github.com/.../commit/<sha>`` →
      ``["patch"]``.
    * ``lists.apache.org/...`` and ``security.apache.org/...`` →
      ``["vendor-advisory"]``.
    * ``cve.org/CVERecord?id=...`` and ``nvd.nist.gov/vuln/detail/...`` →
      ``["related"]`` (links to other CVE records on the public CVE
      databases — used for incomplete-fix / sibling-CVE cross-references
      per ASF Security's request).
    * Anything else → no tags (empty list).
    """
    if re.search(r"github\.com/[^/]+/[^/]+/(pull|commit)/", url):
        return ["patch"]
    # Match the host exactly, not a substring — `if "lists.apache.org"
    # in url` would also flag e.g. `https://evil.com/?q=lists.apache.org`,
    # which CodeQL flags as `py/incomplete-url-substring-sanitization`.
    try:
        host = (urllib.parse.urlparse(url).hostname or "").lower()
    except ValueError:
        return []
    if host in ("lists.apache.org", "security.apache.org"):
        return ["vendor-advisory"]
    if host in ("cve.org", "www.cve.org", "nvd.nist.gov"):
        return ["related"]
    return []


# Match a complete CVE-YYYY-NNNNN identifier with word boundaries so
# substrings inside larger tokens (e.g. ``CVE-2026-12345-extra``) do
# not match.
_CVE_ID_RE = re.compile(r"\bCVE-\d{4}-\d{4,7}\b")


def extract_related_cve_ids(text: str, current_cve_id: str | None = None) -> list[str]:
    """Extract distinct CVE identifiers cited in ``text``, in order of
    first appearance.

    ``current_cve_id`` is excluded from the result so the generator
    never emits a self-reference. The check is case-insensitive.

    Typical inputs:

    * The tracker's *Short public summary for publish* body field —
      where Gate #3 (incomplete-fix cross-CVE clause) places a prior
      CVE identifier the current CVE is a follow-up to.
    * The tracker's *Security mailing list thread* field — when the
      report references a prior CVE in its body for context.

    Output is a list (not a set) to preserve first-appearance order
    so the emitted references list is deterministic across runs.
    """
    seen: set[str] = set()
    ordered: list[str] = []
    current_upper = (current_cve_id or "").upper()
    for match in _CVE_ID_RE.finditer(text):
        cve_id = match.group(0).upper()
        if cve_id == current_upper:
            continue
        if cve_id in seen:
            continue
        seen.add(cve_id)
        ordered.append(cve_id)
    return ordered


def related_cve_url(cve_id: str) -> str:
    """Return the public ``cve.org`` record URL for a CVE identifier.

    Format matches ASF Security's preference per Arnout Engelen's
    2026-05-29 review comment on CVE-2026-49298: ``https://cve.org/
    CVERecord?id=<CVE-ID>``.
    """
    return f"https://www.cve.org/CVERecord?id={cve_id}"


def build_references(
    mailing_list_field: str,
    pr_field: str,
    extra_urls: list[str] | None = None,
) -> list[dict]:
    """Collect every URL from the relevant fields and emit a sorted,
    de-duplicated CVE ``references`` list with automatic tags.

    The ASF CVE tool URL (``cveprocess.apache.org``) and any
    ``airflow-s`` URLs are filtered out before serialising -- those are
    private tracker links and must not appear in a public CVE record.

    The ``mailing_list_field`` argument is **deliberately ignored** for
    URL extraction. The tracking-issue's "Security mailing list thread"
    field is an internal reference that, in our flow, almost always
    points at a non-publicly archived ``security@airflow.apache.org``
    thread -- those URLs 404 for anyone outside the security team.
    The field is kept in the function signature for call-site
    compatibility; callers that have a real public advisory URL to
    include must pass it via ``extra_urls`` (typically the
    ``--advisory-url`` CLI flag). See the "CVE references must never
    point at non-public mailing-list threads" section of ``AGENTS.md``
    for the full rationale.
    """
    _ = mailing_list_field  # intentionally unused; see docstring
    gathered: list[str] = list(parse_url_list(pr_field))
    if extra_urls:
        gathered.extend(extra_urls)
    filtered = [
        url for url in gathered if "cveprocess.apache.org" not in url and TRACKER_FILTER_TOKEN not in url
    ]
    # De-duplicate preserving order, then sort alphabetically.
    seen: set[str] = set()
    ordered: list[str] = []
    for url in filtered:
        if url not in seen:
            seen.add(url)
            ordered.append(url)
    ordered.sort()
    references: list[dict] = []
    for url in ordered:
        entry: dict = {"url": url}
        tags = classify_reference(url)
        if tags:
            entry["tags"] = tags
        references.append(entry)
    return references


# -----------------------------------------------------------------------------
# Build the CVE 5.x CNA container dict.
# -----------------------------------------------------------------------------


def _product_for_package(
    package_name: str,
    *,
    product_overrides: dict[str, str] | None = None,
) -> str:
    """Return the display product name for a project package directory name.

    The mapping is driven by the loaded config (`<project-config>/tools/
    vulnogram/cve-json-config.toml`):

    * `package_name == TOP_LEVEL_NAME` → `TOP_LEVEL_PRODUCT` (e.g.
      `apache-airflow` → `Apache Airflow`).
    * `package_name` matches `PACKAGE_RE` and the regex captures a
      non-empty named `project` group →
      `PROJECT_PRODUCT_TEMPLATE.format(display=...)`, where `display`
      is `PROJECT_DISPLAY_MAP[<project>]` when the directory name is
      known, or a title-cased dash-split fallback otherwise
      (`foo-bar` → `Foo Bar`). The `project` group encodes the
      project's subpackage convention (`-project-<dir>`,
      `-providers-<dir>`, …) — whatever the regex declares — so the
      mapping works for any subpackage prefix the project ships.
    * `product_overrides` lets callers shadow either source by package
      name — used by the `--product-for` CLI flag for unknown
      subpackages or acronyms that don't round-trip through `title()`.

    Unknown `package_name` (neither `TOP_LEVEL_NAME` nor a project
    subpackage) is returned unchanged — callers that pass a
    non-project package name are opting into bring-your-own product
    naming.
    """
    if not TOP_LEVEL_NAME:
        _populate_constants()
    overrides = product_overrides or {}
    if package_name in overrides:
        return overrides[package_name]
    if package_name == TOP_LEVEL_NAME:
        return TOP_LEVEL_PRODUCT
    # Anchor at both ends: a partial match (`match`) can capture an
    # arbitrary suffix into the `project` group for non-standard
    # package strings, producing a fabricated product name. If the
    # package name does not fully match the configured pattern we
    # fall through and treat it as opaque.
    match = PACKAGE_RE.fullmatch(package_name)
    if match is not None:
        project_dir = (match.groupdict().get("project") or "").strip()
        if project_dir:
            if project_dir in PROJECT_DISPLAY_MAP:
                display = PROJECT_DISPLAY_MAP[project_dir]
            else:
                display = " ".join(part.title() for part in project_dir.split("-") if part)
            return PROJECT_PRODUCT_TEMPLATE.format(display=display)
    return package_name


def _split_affected_lines(affected_versions_value: str) -> list[tuple[str | None, str]]:
    """Split the *Affected versions* body field into per-package entries.

    Each non-empty line is inspected for a leading Airflow package
    directory name (``apache-airflow`` or
    ``apache-airflow-project-<dir>``); when detected, the line splits
    into ``(package_name, version_expression)``. Lines without a
    recognisable package prefix yield ``(None, <whole line>)`` so the
    caller can fall back to the default product/package. Bullets
    (``-``, ``*``, ``+``, ``1.``) are stripped so a reviewer can write
    the field as a bulleted list.

    When the input is empty, returns ``[(None, "")]`` so callers emit a
    single placeholder ``affected[]`` entry (the historical behaviour
    for an empty field).
    """
    if not affected_versions_value.strip():
        return [(None, "")]
    entries: list[tuple[str | None, str]] = []
    for raw_line in affected_versions_value.splitlines():
        line = raw_line.strip()
        if not line:
            continue
        line = re.sub(r"^(?:[-*+]\s+|\d+[.)]\s+)", "", line).strip()
        if not line:
            continue
        if not PACKAGE_RE.pattern:
            _populate_constants()
        match = PACKAGE_RE.match(line)
        if match:
            package = match.group("package")
            rest = (match.group("rest") or "").strip()
            entries.append((package, rest))
        else:
            entries.append((None, line))
    if not entries:
        return [(None, "")]
    return entries


def build_affected(
    affected_versions_value: str,
    *,
    vendor: str,
    product: str,
    package_name: str,
    collection_url: str,
    version_start: str | None,
    product_overrides: dict[str, str] | None = None,
) -> list[dict]:
    """Return one ``affected[]`` entry per detected package.

    The *Affected versions* body field is parsed line by line. Each line
    that starts with an Airflow package directory name
    (``apache-airflow`` or ``apache-airflow-project-<dir>``) creates
    an entry with that package's derived product / packageName. Lines
    without a recognisable prefix — or an empty field — fall back to
    the explicit ``product`` / ``package_name`` arguments, which is the
    historical single-entry shape.

    ``product_overrides`` is consulted *before*
    :data:`PROJECT_DISPLAY_MAP` and lets callers shadow the
    resolved product name for any specific package. It is wired to the
    ``--product-for PACKAGE=PRODUCT`` CLI flag.
    """
    entries: list[dict] = []
    for detected_package, version_expr in _split_affected_lines(affected_versions_value):
        if detected_package is not None:
            entry_package = detected_package
            entry_product = _product_for_package(
                detected_package,
                product_overrides=product_overrides,
            )
            version_source = version_expr
        else:
            entry_package = package_name
            entry_product = product
            version_source = version_expr or affected_versions_value
        entries.append(
            {
                "collectionURL": collection_url,
                "defaultStatus": "unaffected",
                "packageName": entry_package,
                "product": entry_product,
                "vendor": vendor,
                "versions": parse_affected_versions(version_source, version_start),
            }
        )
    return entries


def build_descriptions(plain_text: str) -> list[dict]:
    if not plain_text:
        return []
    return [
        {
            "lang": DEFAULT_LANG,
            "supportingMedia": [
                {
                    "base64": False,
                    "type": "text/html",
                    "value": to_html(plain_text),
                }
            ],
            "value": plain_text,
        }
    ]


def build_problem_types(cwe_value: str) -> list[dict]:
    if not cwe_value:
        return []
    cwe_id, description = parse_cwe(cwe_value)
    desc: dict = {
        "lang": DEFAULT_LANG,
        "description": description,
    }
    if cwe_id:
        desc["cweId"] = cwe_id
        desc["type"] = "CWE"
    return [{"descriptions": [desc]}]


def build_metrics(severity_value: str) -> list[dict]:
    if not severity_value:
        return []
    return [
        {
            "other": {
                "content": {"text": normalise_severity(severity_value)},
                "type": "Textual description of severity",
            }
        }
    ]


def build_credits(
    credited_as_value: str,
    *,
    remediation_developers: list[str],
) -> list[dict]:
    credits: list[dict] = []
    for name in parse_credits_from_field(credited_as_value):
        credit_type = TOOL_CREDIT_TYPE if is_bot_credit(name) else DEFAULT_CREDIT_TYPE
        credits.append({"lang": DEFAULT_LANG, "type": credit_type, "value": name})
    for name in remediation_developers:
        cleaned = name.strip()
        if cleaned:
            credits.append(
                {
                    "lang": DEFAULT_LANG,
                    "type": "remediation developer",
                    "value": cleaned,
                }
            )
    return credits


def build_cna_container(
    *,
    title: str,
    description: str,
    affected_versions_value: str,
    cwe_value: str,
    severity_value: str,
    credits_value: str,
    mailing_list_value: str,
    pr_value: str,
    vendor: str,
    product: str,
    package_name: str,
    collection_url: str,
    org_id: str,
    version_start: str | None,
    discovery: str,
    remediation_developers: list[str],
    advisory_urls: list[str] | None = None,
    product_overrides: dict[str, str] | None = None,
    current_cve_id: str | None = None,
) -> dict:
    # Sibling-CVE cross-references — extract every distinct CVE-YYYY-NNNNN
    # mentioned in the description (the short public summary) and emit a
    # ``cve.org/CVERecord?id=<id>`` reference for each, tagged ``related``
    # by :func:`classify_reference`. This satisfies ASF Security's request
    # (Arnout Engelen, 2026-05-29 review on CVE-2026-49298) that incomplete-
    # fix follow-ups carry a structured ``references[]`` link back to the
    # prior CVE. The current record's own CVE ID is excluded so the
    # generator never emits a self-reference.
    related_cve_urls = [related_cve_url(cid) for cid in extract_related_cve_ids(description, current_cve_id)]
    extra_urls = list(advisory_urls or []) + related_cve_urls
    cna: dict = {
        "affected": build_affected(
            affected_versions_value,
            vendor=vendor,
            product=product,
            package_name=package_name,
            collection_url=collection_url,
            version_start=version_start,
            product_overrides=product_overrides,
        ),
        "credits": build_credits(credits_value, remediation_developers=remediation_developers),
        "descriptions": build_descriptions(description),
        "metrics": build_metrics(severity_value),
        "problemTypes": build_problem_types(cwe_value),
        "providerMetadata": {"orgId": org_id},
        "references": build_references(mailing_list_value, pr_value, extra_urls=extra_urls),
        "source": {"discovery": discovery},
        "title": title,
        "x_generator": {"engine": GENERATOR_TAG},
    }
    return cna


def _has_vendor_advisory_reference(cna: dict) -> bool:
    """Return ``True`` when the CNA container has at least one
    ``references[]`` entry tagged ``vendor-advisory``.

    :func:`classify_reference` tags ``lists.apache.org/...`` and
    ``security.apache.org/...`` URLs as ``vendor-advisory``
    automatically. When the tracking issue's *"Public advisory URL"*
    body field has been populated with the archived advisory URL —
    i.e. the advisory has actually shipped to users@ / announce@ and
    been archived on the public ponymail archive — that URL flows
    into ``references[]`` and flips this check to ``True``.

    That is the cue the Vulnogram workflow state should advance from
    ``REVIEW`` to ``PUBLIC``. The result lands in
    ``CNA_private.state`` in the emitted envelope
    (:func:`wrap_cve_record`); ``cveMetadata.state`` is the standard
    CVE 5.x schema field and always holds ``"PUBLISHED"``.
    """
    references = cna.get("references") or []
    for ref in references:
        tags = ref.get("tags") or []
        if "vendor-advisory" in tags:
            return True
    return False


def _is_cna_ready_for_review(cna: dict, cve_id: str) -> bool:
    """Return ``True`` when the CNA container has everything a release
    manager needs for the ASF CVE tool's ``REVIEW`` state.

    The fields checked here correspond to the minimum a release manager
    needs to send the public advisory: a CVE ID, a title, a non-empty
    description, at least one affected-versions entry, a CWE in
    ``problemTypes``, a severity in ``metrics``, at least one credit,
    and at least one reference (the fix PR). If any of those is
    missing, the record is a ``DRAFT`` -- still useful as the
    paste-ready scaffold, but not yet review-ready.
    """
    if not cve_id:
        return False
    if not cna.get("title") or not str(cna.get("title")).strip():
        return False
    descriptions = cna.get("descriptions") or []
    if not any((d.get("value") or "").strip() for d in descriptions):
        return False
    affected = cna.get("affected") or []
    if not any(a.get("versions") for a in affected):
        return False
    problem_types = cna.get("problemTypes") or []
    has_cwe = any(
        (desc.get("cweId") or "").strip() for pt in problem_types for desc in (pt.get("descriptions") or [])
    )
    if not has_cwe:
        return False
    metrics = cna.get("metrics") or []
    has_severity = False
    for metric in metrics:
        other = metric.get("other") or {}
        content = other.get("content") or {}
        severity_text = (content.get("text") or "").strip().lower()
        if severity_text and severity_text not in {"unknown", "_no response_"}:
            has_severity = True
            break
    if not has_severity:
        return False
    credits = cna.get("credits") or []
    if not any(c.get("value") for c in credits):
        return False
    references = cna.get("references") or []
    return any(r.get("url") for r in references)


def compute_cna_private_state(
    cna: dict,
    cve_id: str,
    *,
    release_vote_in_progress: bool | None = None,
) -> str:
    """Return the ``CNA_private.state`` value (DRAFT / REVIEW / PUBLIC).

    State machine:

    * ``DRAFT`` — the CNA is incomplete (some required field missing)
      *or* the caller signalled "release-vote gating is on but no vote
      is happening" (``release_vote_in_progress=False``).
    * ``REVIEW`` — the CNA is fully populated (``_is_cna_ready_for_review``
      passes), no public advisory has shipped yet, AND either
      release-vote gating is disabled (``release_vote_in_progress=None``,
      the default — legacy behaviour for non-ASF adopters) or the
      caller signalled that an RC is being voted
      (``release_vote_in_progress=True``).
    * ``PUBLIC`` — the CNA is review-ready *and* the public advisory
      has shipped (a ``vendor-advisory`` reference is present).

    The ``release_vote_in_progress`` parameter is tri-state:

    * ``None`` (default): release-vote gating is *off* — a ready CNA
      advances to REVIEW immediately. This preserves the original
      generator behaviour and is the right default for non-ASF
      adopters that don't have a separate release-vote step before
      publishing an advisory.
    * ``True``: release-vote gating is on and the vote is in progress
      — a ready CNA advances to REVIEW. ASF projects opt into this
      via ``[workflow].release_vote_gating = true`` plus either the
      ``rc voting`` tracker label or a ``--review`` CLI override.
    * ``False``: release-vote gating is on but no vote is happening
      — a ready CNA stays at DRAFT. This is the new "don't auto-
      advance to REVIEW outside the vote window" behaviour requested
      for ASF projects, where REVIEW carries the semantics
      "release manager is about to publish".

    Centralises the decision that :func:`wrap_cve_record` makes so
    other code paths (the issue-body attachment table) can display the
    same state without duplicating the logic.
    """
    if not _is_cna_ready_for_review(cna, cve_id):
        return "DRAFT"
    if _has_vendor_advisory_reference(cna):
        return "PUBLIC"
    if release_vote_in_progress is False:
        return "DRAFT"
    return "REVIEW"


def compute_package_url(collection_url: str, package_name: str) -> str | None:
    """Return the canonical project-page URL for a package.

    Maps the CVE ``affected[].collectionURL`` (e.g. legacy
    ``https://pypi.python.org``) to the public project page on the
    same registry (``https://pypi.org/project/<package>/``). Returns
    ``None`` when the collection URL has no mapping (e.g. the
    Helm-chart entries that point at ``airflow.apache.org`` rather
    than a package registry); the attachment table renders ``—`` in
    that case.
    """
    if not collection_url or not package_name:
        return None
    template = COLLECTION_URL_TO_PROJECT_URL_TEMPLATE.get(collection_url.rstrip("/"))
    if template is None:
        return None
    return template.format(package=package_name)


def format_version_range(versions: list[dict]) -> str:
    """Render ``affected[].versions[]`` as a human-readable range string.

    Inverse of :func:`parse_affected_versions`: takes the structured
    CVE 5.x version entries and reconstructs the ``>= X, < Y`` /
    ``<= X`` / bare-version expressions a reviewer can recognise at a
    glance. Multiple entries (rare) are joined with ``; ``.
    """
    if not versions:
        return ""
    parts: list[str] = []
    for entry in versions:
        version = (entry.get("version") or "").strip()
        less_than = (entry.get("lessThan") or "").strip()
        less_than_or_equal = (entry.get("lessThanOrEqual") or "").strip()
        if less_than:
            parts.append(f">= {version}, < {less_than}" if version and version != "0" else f"< {less_than}")
        elif less_than_or_equal:
            parts.append(
                f">= {version}, <= {less_than_or_equal}"
                if version and version != "0"
                else f"<= {less_than_or_equal}"
            )
        elif version:
            parts.append(version)
    return "; ".join(parts)


def wrap_cve_record(
    cna: dict,
    *,
    cve_id: str,
    org_id: str,
    release_vote_in_progress: bool | None = None,
) -> dict:
    """Wrap the ``cna`` container in a CVE 5.x record envelope.

    The envelope matches the real Vulnogram export shape: ``dataType``,
    ``dataVersion``, ``containers.cna`` and a ``cveMetadata`` block
    containing ``cveId``, ``assignerOrgId``, ``serial`` and ``state``.

    The envelope carries two different ``state`` fields at different
    levels. They track different things and must not be confused:

    - ``CNA_private.state`` is the **ASF Vulnogram workflow state** —
      an internal-to-the-CNA scratchpad that describes where the
      record is in the tool's own review workflow
      (``DRAFT`` → ``REVIEW`` → ``PUBLIC``). It is what the ASF CVE
      tool UI displays and what a release manager flips as they
      shepherd the record from staging to publication. Its value is
      derived here from the readiness of the CNA *and* whether the
      public advisory has actually shipped:

      * ``"DRAFT"`` when any required field is missing. The JSON is
        a paste-ready scaffold; Vulnogram will not let a reviewer
        move it further without the missing fields.
      * ``"REVIEW"`` when every field a release manager needs is
        present (CVE ID, title, description, affected versions,
        CWE, non-``Unknown`` severity, at least one credit, at
        least one reference), **no public advisory URL has been
        captured yet**, *and* the caller signalled that an RC is
        being voted (``release_vote_in_progress=True`` —
        typically because the tracker carries the configured
        ``rc voting`` label). This is the state the RM pastes at
        Step 13 when sending the advisory email. Without the
        vote signal the record stays at ``DRAFT`` even when fully
        populated; this avoids auto-advancing records that are
        not yet at the *send-the-advisory* moment.
      * ``"PUBLIC"`` when the CNA is review-ready **and** at least
        one ``references[]`` entry is tagged ``vendor-advisory`` —
        which is the case as soon as the tracking issue's *"Public
        advisory URL"* body field is populated with the archived
        users@airflow.apache.org thread URL. At that point the
        advisory is out, the CVE record is ready to go public, and
        the next paste into Vulnogram should land the record at
        PUBLIC. See Steps 14 and 15 of the handling process in
        ``README.md``. ``CNA_private.emailed`` is also flipped to
        ``"yes"`` at this state so Vulnogram's UI reflects that the
        advisory email has been sent.

    - ``cveMetadata.state`` is the **CVE 5.x schema state** per
      `the CVE-List schema
      <https://github.com/CVEProject/cve-schema>`_. The only valid
      values are ``"PUBLISHED"`` and ``"REJECTED"``; any submitted
      record is by definition published (the CVE List's earlier
      ``"RESERVED"`` state applies to an allocated-but-unsubmitted
      CVE ID, not to a record envelope). The generator therefore
      hard-codes this to ``"PUBLISHED"`` — the intent-statement
      that cve.org reads when the record eventually flows through
      Vulnogram.
    """
    cna_private_state = compute_cna_private_state(
        cna, cve_id, release_vote_in_progress=release_vote_in_progress
    )
    record: dict = {
        "CNA_private": {
            "emailed": "yes" if cna_private_state == "PUBLIC" else None,
            "projecturl": CNA_PRIVATE_PROJECT_URL,
            "owner": CNA_PRIVATE_OWNER,
            "userslist": CNA_PRIVATE_USERS_LIST,
            "state": cna_private_state,
            "todo": [],
            "type": "unsure",
        },
        "containers": {"cna": cna},
        "dataType": "CVE_RECORD",
        "dataVersion": "5.1",
    }
    cve_metadata: dict = {
        "assignerOrgId": org_id,
        "serial": 1,
        "state": "PUBLISHED",
    }
    if cve_id:
        cve_metadata["cveId"] = cve_id
    record["cveMetadata"] = cve_metadata
    return record


# -----------------------------------------------------------------------------
# Attach / replace the JSON as an issue comment on the tracking issue.
#
# GitHub issues have no native "attachment" API for automation (only the
# drag-and-drop `user-attachments` uploads in the web UI, which aren't
# scriptable). The next best thing is to post an issue comment that
# contains the full JSON in a fenced code block, with a stable HTML
# marker comment so re-runs can find the previous attachment comment and
# edit it in place instead of posting a duplicate every time.
# -----------------------------------------------------------------------------


def _attachment_marker(cve_id: str) -> str:
    """Return the begin marker. Kept for back-compat with older call sites."""
    return _attachment_marker_begin(cve_id)


def _attachment_marker_begin(cve_id: str) -> str:
    return (
        f"{ATTACHMENT_MARKER_BEGIN_PREFIX} cve={cve_id or 'UNKNOWN'} version={ATTACHMENT_MARKER_VERSION} -->"
    )


def _attachment_marker_end(cve_id: str) -> str:
    return f"{ATTACHMENT_MARKER_END_PREFIX} cve={cve_id or 'UNKNOWN'} version={ATTACHMENT_MARKER_VERSION} -->"


def _escape_table_cell(value: str) -> str:
    """Escape characters that would break a markdown table cell."""
    if not value:
        return ""
    return value.replace("|", "\\|").replace("\n", " ").strip()


def _build_affected_table(affected: list[dict]) -> str:
    """Return the markdown table summarising ``affected[]`` for the
    attachment block.

    One row per package (so multi-product CVEs surface every entry the
    JSON will land in Vulnogram). Columns: package name, product,
    versions (rendered back to ``>= X, < Y`` shape via
    :func:`format_version_range`), and a clickable PyPI project-page
    URL derived from ``collectionURL`` + ``packageName`` so a reader
    can jump straight to the package without reconstructing the URL.
    """
    if not affected:
        return ""
    header = "| # | Package | Product | Versions | PyPI URL |\n| :-- | :-- | :-- | :-- | :-- |\n"
    rows: list[str] = []
    for index, entry in enumerate(affected, start=1):
        package = (entry.get("packageName") or "").strip()
        product = (entry.get("product") or "").strip()
        versions = format_version_range(entry.get("versions") or [])
        package_url = compute_package_url(entry.get("collectionURL") or "", package)
        package_cell = f"`{_escape_table_cell(package)}`" if package else "—"
        product_cell = _escape_table_cell(product) or "—"
        versions_cell = f"`{_escape_table_cell(versions)}`" if versions else "—"
        url_cell = f"<{package_url}>" if package_url else "—"
        rows.append(f"| {index} | {package_cell} | {product_cell} | {versions_cell} | {url_cell} |")
    return header + "\n".join(rows) + "\n"


def _build_attachment_body(
    *,
    cve_id: str,
    json_text: str,
    cna: dict,
    cna_private_state: str | None,
) -> str:
    """Return the markdown fragment embedded into the issue body.

    The fragment starts with a ``<!-- generate-cve-json: cve=… -->``
    begin marker and ends with a matching ``:end …`` marker, so
    subsequent re-runs can locate and replace this exact block without
    guessing at boundaries. The JSON itself is wrapped in a
    ``<details>`` disclosure so a long payload does not bloat the
    issue view, and placed inside a four-backtick fenced block so any
    three-backtick tokens inside the JSON (unlikely, but defensive)
    don't break the markdown parse.

    The summary table above the JSON surfaces the CVE id, title, the
    Vulnogram workflow state (``cna_private_state``), credit /
    reference counts, and a per-package sub-table so a reader can see
    *what* the JSON will land in Vulnogram without expanding the
    payload.
    """
    begin_marker = _attachment_marker_begin(cve_id)
    end_marker = _attachment_marker_end(cve_id)
    byte_count = len(json_text.encode("utf-8"))
    cve_heading = "CVE JSON — paste-ready"
    if cve_id:
        cve_link = f"[`{cve_id}`](https://cveprocess.apache.org/cve5/{cve_id}#source)"
        cve_heading = f"CVE JSON — paste-ready for {cve_link}"
    script_link = f"[`generate_cve_json.py`]({SKILL_SOURCE_URL})"
    title = (cna.get("title") or "").strip()
    affected = cna.get("affected") or []
    credit_count = len(cna.get("credits") or [])
    reference_count = len(cna.get("references") or [])
    title_cell = _escape_table_cell(title) or "—"
    state_cell = f"`{cna_private_state}`" if cna_private_state else "— (`--no-envelope`)"
    package_names = [_escape_table_cell((entry.get("packageName") or "").strip()) for entry in affected]
    package_names = [name for name in package_names if name]
    packages_cell = ", ".join(f"`{name}`" for name in package_names) if package_names else "—"
    metric_table = (
        f"| Metric | Value |\n"
        f"| :-- | :-- |\n"
        f"| CVE ID | `{cve_id or 'UNKNOWN'}` |\n"
        f"| Title | {title_cell} |\n"
        f"| Vulnogram state | {state_cell} |\n"
        f"| Affected packages | {packages_cell} |\n"
        f"| Credits | {credit_count} |\n"
        f"| References | {reference_count} |\n"
        f"| Size | {byte_count} bytes |\n"
    )
    package_block = ""
    affected_table = _build_affected_table(affected)
    if affected_table:
        package_block = f"\n**Packages this JSON covers:**\n\n{affected_table}"
    return (
        f"{begin_marker}\n\n"
        f"## {cve_heading}\n\n"
        f"Generated by {script_link} from the current state of this "
        f"issue body. The JSON is **deterministic**: re-running the "
        f"skill after updating the body (for example once the reporter "
        f"confirms a credit form, or the affected version range is "
        f"narrowed) produces a byte-for-byte identical payload.\n\n"
        f"**Paste into Vulnogram** via the `#source` tab at the CVE tool "
        f"URL above and click save; the form view reflects the new "
        f"values.\n\n"
        f"{metric_table}"
        f"{package_block}\n"
        f"<details>\n"
        f"<summary>CVE 5.x JSON ({byte_count} bytes) — click to "
        f"expand</summary>\n\n"
        f"````json\n"
        f"{json_text}\n"
        f"````\n\n"
        f"</details>\n\n"
        f"{end_marker}\n"
    )


def _gh_api_json(
    args: list[str],
    *,
    body_payload: dict | None = None,
) -> dict | list:
    """Run ``gh api`` and return the parsed JSON stdout.

    When ``body_payload`` is supplied, it is serialised to a temp file
    and passed via ``--input`` so ``gh api`` uses it as the request
    body (this is how ``gh api`` wants long JSON bodies for ``POST`` /
    ``PATCH`` calls -- ``-f`` URL-encodes and mangles newlines).
    """
    tmp_path: str | None = None
    try:
        if body_payload is not None:
            with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as handle:
                json.dump(body_payload, handle)
                tmp_path = handle.name
            args = [*args, "--input", tmp_path]
        result = subprocess.run(
            ["gh", "api", *args],
            check=True,
            capture_output=True,
            text=True,
        )
    except FileNotFoundError as exc:
        raise RuntimeError("`gh` CLI not found on PATH. Install it from https://cli.github.com/.") from exc
    except subprocess.CalledProcessError as exc:
        raise RuntimeError(f"`gh api {' '.join(args)}` failed:\n{exc.stderr.strip()}") from exc
    finally:
        if tmp_path is not None:
            with contextlib.suppress(OSError):
                os.unlink(tmp_path)

    if not result.stdout.strip():
        return {}
    return json.loads(result.stdout)


def _fetch_issue(repo: str, issue_number: str) -> dict:
    """Return the full issue dict from the REST API, including ``body`` and ``html_url``."""
    response = _gh_api_json([f"repos/{repo}/issues/{issue_number}"])
    if not isinstance(response, dict):
        raise RuntimeError(f"Unexpected response shape for `gh api repos/{repo}/issues/{issue_number}`")
    return response


def _splice_attachment_into_body(issue_body: str, attachment: str, cve_id: str) -> str:
    """Return ``issue_body`` with the CVE-JSON attachment spliced in place.

    Behaviour, in order of preference:

    - If the body already contains a ``generate-cve-json`` begin/end marker
      pair for this CVE, replace the block between them with the new
      attachment.
    - Otherwise, if the body contains a ``### CVE tool link`` field,
      append the attachment *after* that field's value, i.e. right at
      the end of the template.
    - Otherwise, append the attachment at the very end of the body.

    The body is returned with a trailing newline.
    """
    begin_marker = _attachment_marker_begin(cve_id)
    end_marker = _attachment_marker_end(cve_id)

    # Regex that matches begin marker ... end marker, non-greedy, across lines.
    # The marker text itself contains characters that need escaping.
    begin_escaped = re.escape(begin_marker)
    end_escaped = re.escape(end_marker)
    pattern = rf"{begin_escaped}.*?{end_escaped}\n?"
    if re.search(pattern, issue_body, re.DOTALL):
        new_body = re.sub(pattern, attachment.rstrip("\n") + "\n", issue_body, count=1, flags=re.DOTALL)
        return new_body.rstrip() + "\n"

    # Also tolerate a legacy single-marker attachment that was embedded
    # without the matching end marker (shouldn't happen for our content,
    # but defensive): drop from the begin marker to end-of-body and
    # re-append.
    legacy_pattern = rf"{begin_escaped}.*\Z"
    if re.search(legacy_pattern, issue_body, re.DOTALL):
        stripped = re.sub(legacy_pattern, "", issue_body, count=1, flags=re.DOTALL).rstrip()
        return f"{stripped}\n\n{attachment.rstrip()}\n"

    # No existing block. Try to locate the CVE-tool-link field so we can
    # append after the whole template rather than breaking it in half.
    # The CVE-tool-link field is the last one in the standard template,
    # so appending after it naturally puts the attachment at the end.
    cve_tool_link_pattern = r"^###\s+CVE tool link\s*\n+.*?(?=\n###\s|\Z)"
    match = re.search(cve_tool_link_pattern, issue_body, re.MULTILINE | re.DOTALL)
    if match is not None:
        insertion_point = match.end()
        prefix = issue_body[:insertion_point].rstrip()
        suffix = issue_body[insertion_point:].rstrip()
        middle = f"\n\n{attachment.rstrip()}\n"
        if suffix:
            return f"{prefix}{middle}\n\n{suffix}\n"
        return f"{prefix}{middle}"

    return issue_body.rstrip() + f"\n\n{attachment.rstrip()}\n"


def attach_to_issue(
    *,
    issue_number: str,
    repo: str,
    cve_id: str,
    json_text: str,
    cna: dict,
    cna_private_state: str | None,
) -> tuple[str, bool]:
    """Embed the CVE-JSON attachment in the issue body; update idempotently.

    The attachment used to be posted as a separate issue comment, but
    GitHub has no pin-comment API, so the comment ended up buried below
    every status-change comment and newcomers had no idea it existed.
    Embedding in the issue body puts it above the entire comment
    timeline permanently without needing any pin mechanism.

    Returns ``(html_url, was_update)`` where ``html_url`` is a stable
    anchor into the issue body at the CVE-JSON section heading (GitHub
    auto-generates heading anchors from the ``## CVE JSON — paste-ready
    …`` text), and ``was_update`` is ``True`` when an existing
    attachment was replaced in place, ``False`` when the attachment was
    appended to the body for the first time.
    """
    attachment = _build_attachment_body(
        cve_id=cve_id,
        json_text=json_text,
        cna=cna,
        cna_private_state=cna_private_state,
    )
    issue = _fetch_issue(repo, issue_number)
    current_body = issue.get("body") or ""
    had_existing = _attachment_marker_begin(cve_id) in current_body
    new_body = _splice_attachment_into_body(current_body, attachment, cve_id)

    if new_body.rstrip() == current_body.rstrip():
        # Byte-for-byte identical — skip the PATCH to avoid a no-op update
        # timestamp on the issue.
        html_url = issue.get("html_url", "")
        return (html_url, had_existing)

    _gh_api_json(
        [
            "-X",
            "PATCH",
            f"repos/{repo}/issues/{issue_number}",
        ],
        body_payload={"body": new_body},
    )
    html_url = issue.get("html_url", "")
    # Point the returned URL at the CVE-JSON heading anchor inside the
    # body. GitHub renders the H2 as `#cve-json--paste-ready-for-cve-...`.
    if cve_id:
        anchor_slug = f"cve-json--paste-ready-for-{cve_id.lower()}"
        html_url = f"{html_url}#{anchor_slug}" if html_url else ""
    return (html_url, had_existing)


# -----------------------------------------------------------------------------
# Entry point.
# -----------------------------------------------------------------------------


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=(
            "Generate a CVE 5.x JSON record from a tracker issue, ready "
            "to paste into the Vulnogram #source tab of the ASF CVE tool. "
            "Project-specific defaults (vendor, product, package map, "
            "CNA org id, …) come from a TOML config file the adopting "
            "project ships at `<project-config>/tools/cve-tool-vulnogram/"
            "cve-json-config.toml`."
        ),
    )
    parser.add_argument(
        "--config",
        type=Path,
        default=None,
        help=(
            f"Path to the TOML config file. Defaults to "
            f"$CVE_JSON_CONFIG or {_DEFAULT_CONFIG_RELPATH} relative to cwd."
        ),
    )
    parser.add_argument(
        "issue",
        nargs="?",
        help="Issue number in the tracker repo (omit when --stdin is used).",
    )
    parser.add_argument(
        "--repo",
        default=None,
        help="GitHub repo holding the tracking issue (default from config: tracker_repo).",
    )
    parser.add_argument(
        "--output",
        type=Path,
        help=("Write the resulting JSON to this path. When omitted the JSON is printed to stdout."),
    )
    parser.add_argument(
        "--stdin",
        action="store_true",
        help=(
            "Read the issue body from stdin instead of calling `gh`. Useful "
            "for offline iteration and for the skill's test mode."
        ),
    )
    parser.add_argument(
        "--cve-id",
        help=("Override the CVE ID. Defaults to the id parsed from the issue's `CVE tool link` field."),
    )
    parser.add_argument(
        "--title",
        help=(
            "Override the CVE title. Defaults to the GitHub issue title "
            "with an `Apache Airflow: ` prefix when not already present."
        ),
    )
    parser.add_argument(
        "--vendor",
        default=None,
        help="CVE vendor name (default from config: product.vendor).",
    )
    parser.add_argument(
        "--product",
        default=None,
        help="CVE product name (default from config: product.default_product).",
    )
    parser.add_argument(
        "--package-name",
        default=None,
        help="Package name (default from config: product.default_package_name).",
    )
    parser.add_argument(
        "--collection-url",
        default=None,
        help="Package collection URL (default from config: product.default_collection_url).",
    )
    parser.add_argument(
        "--product-for",
        action="append",
        default=[],
        metavar="PACKAGE=PRODUCT",
        help=(
            "Override the CVE product name for a specific packageName, "
            "e.g. `--product-for apache-airflow-project-foo='Apache "
            "Airflow Foo Project'`. Useful when a project is not yet "
            "in the built-in display-name map, or when the default "
            "title-cased fallback needs a different casing. Repeat the "
            "flag to override multiple packages."
        ),
    )
    parser.add_argument(
        "--org-id",
        default=None,
        help="CNA assigner org id (default from config: cna.org_id).",
    )
    parser.add_argument(
        "--version-start",
        help=(
            "Override the version range start (the value put in "
            "`affected[].versions[].version`). Defaults to the lower bound "
            "parsed from the Affected versions field when it uses "
            "`>= X, < Y` syntax, otherwise `0`."
        ),
    )
    parser.add_argument(
        "--discovery",
        default=DEFAULT_DISCOVERY,
        help=(
            f"Value for `source.discovery` (default: {DEFAULT_DISCOVERY!r}). "
            "Valid CVE 5.x values include UNKNOWN, INTERNAL, EXTERNAL, USER."
        ),
    )
    parser.add_argument(
        "--remediation-developer",
        action="append",
        default=[],
        metavar="NAME",
        help=(
            "Name to add to `credits[]` with type 'remediation developer'. "
            "In normal use, the developer name lives in the issue body's "
            "*Remediation developer* field (auto-populated by the "
            "`security-issue-sync` skill from the linked PR's author). "
            "This flag is an additional / override mechanism: names "
            "passed here are appended to whatever the body already "
            "specifies, with duplicates silently dropped. Repeat the "
            "flag to credit multiple people."
        ),
    )
    parser.add_argument(
        "--advisory-url",
        action="append",
        default=[],
        metavar="URL",
        help=(
            "Public advisory URL to add to `references[]` as a "
            "`vendor-advisory` reference. Repeat the flag to add multiple "
            "URLs. Any URL already present in the tracking-issue's "
            "'Public advisory URL' body field is picked up automatically, "
            "so in the normal flow the release manager populates that "
            "field once the advisory is archived on "
            "`users@airflow.apache.org` / `announce@apache.org` and this "
            "flag is only needed for ad-hoc overrides. The script never "
            "pulls URLs from the 'Security mailing list thread' field — "
            "that field is the internal `security@airflow.apache.org` "
            "thread, which 404s for anyone outside the security team. See "
            "the 'CVE references must never point at non-public mailing-"
            "list threads' section of `AGENTS.md` for the full rationale."
        ),
    )
    parser.add_argument(
        "--no-envelope",
        action="store_true",
        help=(
            "Emit only the `cna` container instead of the full CVE 5.x "
            "record. Use this if Vulnogram's #source tab is in 'inner block "
            "only' mode."
        ),
    )
    state_override = parser.add_mutually_exclusive_group()
    state_override.add_argument(
        "--review",
        action="store_true",
        help=(
            "Force `CNA_private.state = REVIEW` for this run (overrides "
            "the tracker-label signal). Use when nudging a record forward "
            "by hand — for example when the release manager has cut an RC "
            "but the `rc voting` label is not yet on the tracker. Requires "
            "the CNA to be review-ready (all fields populated); otherwise "
            "the state stays DRAFT regardless."
        ),
    )
    state_override.add_argument(
        "--draft",
        action="store_true",
        help=(
            "Force `CNA_private.state = DRAFT` for this run (overrides "
            "the tracker-label signal). Use to walk a record back to "
            "DRAFT when an RC vote failed or was cancelled and the "
            "`rc voting` label is still on the tracker."
        ),
    )
    parser.add_argument(
        "--attach",
        action="store_true",
        help=(
            "After generating the JSON, embed it at the end of the "
            "tracking issue's **body** (wrapped in a collapsible "
            "`<details>` block). The attachment lives *in the body*, "
            "not as a separate comment, so it stays above every "
            "status-change comment in the timeline — effectively "
            "pinned without needing any pin mechanism. If a previous "
            "attachment from this script already exists in the body "
            "(matched by a `<!-- generate-cve-json: cve=... -->` begin "
            "marker), the block between the begin/end markers is "
            "replaced in place — no duplicates. Requires the positional "
            "issue argument (cannot be combined with --stdin)."
        ),
    )
    return parser.parse_args(argv)


def resolve_title(
    issue_title: str,
    summary: str,
    override: str | None,
) -> str:
    if override:
        title = override.strip()
    else:
        title = issue_title.strip()
        if not title:
            title = summary.split(".", 1)[0].strip() if summary else ""
    # The CNA container is already scoped to the project, so the
    # repeated "<vendor>: <product>:" prefix on the title is noise.
    # Strip it (along with any trailing punctuation/whitespace) if
    # present. The strip pattern is built from the configured
    # `top_level_product` at config-load time — see `TITLE_STRIP_RE`
    # in `_populate_constants`.
    title = TITLE_STRIP_RE.sub("", title)
    return title


def emit_json(obj: dict, output: Path | None) -> str:
    text = json.dumps(obj, indent=4, sort_keys=True, ensure_ascii=False)
    if output:
        output.parent.mkdir(parents=True, exist_ok=True)
        output.write_text(text + "\n", encoding="utf-8")
    return text


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)

    # Load config (CLI flag overrides env var overrides default path).
    try:
        if args.config is not None:
            _set_config_path(args.config)
        else:
            _populate_constants()
    except FileNotFoundError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2

    # Apply config-derived defaults to args that defaulted to None.
    if args.repo is None:
        args.repo = DEFAULT_REPO
    if args.vendor is None:
        args.vendor = DEFAULT_VENDOR
    if args.product is None:
        args.product = DEFAULT_PRODUCT
    if args.package_name is None:
        args.package_name = DEFAULT_PACKAGE_NAME
    if args.collection_url is None:
        args.collection_url = DEFAULT_COLLECTION_URL
    if args.org_id is None:
        args.org_id = DEFAULT_ASF_ORG_ID

    if args.attach and args.stdin:
        print(
            "error: --attach cannot be combined with --stdin (there is no real issue to attach to)",
            file=sys.stderr,
        )
        return 2
    if args.attach and not args.issue:
        print(
            "error: --attach requires the positional issue argument",
            file=sys.stderr,
        )
        return 2

    issue_labels: list[str] = []
    if args.stdin:
        body = sys.stdin.read()
        issue_title = args.title or ""
    else:
        if not args.issue:
            print(
                "error: issue number is required unless --stdin is used",
                file=sys.stderr,
            )
            return 2
        try:
            issue_title, body, issue_labels = fetch_issue(args.issue, args.repo)
        except RuntimeError as exc:
            print(f"error: {exc}", file=sys.stderr)
            return 1

    # CNA_private.state = REVIEW is tri-state. When release-vote gating
    # is disabled (the default, [workflow].release_vote_gating = false),
    # a fully-populated CNA advances to REVIEW immediately — the legacy
    # behaviour kept for non-ASF adopters that don't have a separate
    # release-vote step before publishing the advisory.
    #
    # When release-vote gating is enabled (ASF adopters opt in), the
    # signal is computed from one of:
    #   - explicit CLI overrides (--review / --draft);
    #   - the presence of RC_VOTING_LABEL on the tracker;
    #   - default to False (no vote happening).
    # In --stdin mode we have no labels to inspect, so the gated default
    # is False — pass --review explicitly to opt into REVIEW from a
    # piped body.
    release_vote_in_progress: bool | None
    if args.review:
        release_vote_in_progress = True
    elif args.draft:
        release_vote_in_progress = False
    elif RELEASE_VOTE_GATING:
        # `rc voting` is the *forward* signal that a vote is happening.
        # Once the release has shipped, the vote-gating is moot — the
        # release is out, the advisory is going to be written next. Any
        # forward-state label on the tracker (`fix released`,
        # `announced - emails sent`, `announced`, `vendor-advisory ready`,
        # plus any project-specific extensions in
        # `[workflow].forward_state_labels`) is the "vote completed"
        # signal. Without this OR, removing `rc voting` at the
        # `pr merged → fix released` transition (per the sync skill's
        # convention) would compute `release_vote_in_progress=False` and
        # walk the embedded `CNA_private.state` back from REVIEW to
        # DRAFT — the wrong direction for a record that's about to
        # publish.
        release_vote_in_progress = RC_VOTING_LABEL in issue_labels or bool(
            FORWARD_STATE_LABELS & set(issue_labels)
        )
    else:
        release_vote_in_progress = None  # legacy: ready ⇒ REVIEW

    summary = extract_field(body, "Short public summary for publish")
    affected_field = extract_field(body, "Affected versions")
    mailing_list = extract_field(body, "Security mailing list thread")
    public_advisory_field = extract_field(body, "Public advisory URL")
    credited_as = extract_field(body, "Reporter credited as")
    pr_field = extract_field(body, "PR with the fix")
    remediation_developer_field = extract_field(body, "Remediation developer")
    cwe = extract_field(body, "CWE")
    severity = extract_field(body, "Severity")
    cve_tool = extract_field(body, "CVE tool link")

    cve_id = args.cve_id or parse_cve_id(cve_tool)
    title = resolve_title(issue_title, summary, args.title)
    description = summary or title

    # Advisory URLs come from two sources: the body's "Public advisory URL"
    # field (populated by the release manager / security-issue-sync skill
    # once the advisory is archived on users@airflow.apache.org) and any
    # --advisory-url CLI overrides. Both are forwarded to the references
    # builder; de-duplication happens there.
    body_advisory_urls = parse_url_list(public_advisory_field)
    combined_advisory_urls = [*body_advisory_urls, *args.advisory_url]

    combined_remediation_developers = combine_remediation_developers(
        remediation_developer_field,
        args.remediation_developer,
    )

    product_overrides: dict[str, str] = {}
    for item in args.product_for:
        if "=" not in item:
            print(
                f"error: --product-for expects PACKAGE=PRODUCT, got {item!r}",
                file=sys.stderr,
            )
            return 2
        pkg, _, display = item.partition("=")
        pkg = pkg.strip()
        display = display.strip()
        if not pkg or not display:
            print(
                f"error: --product-for expects non-empty PACKAGE and PRODUCT, got {item!r}",
                file=sys.stderr,
            )
            return 2
        product_overrides[pkg] = display

    try:
        cna = build_cna_container(
            title=title,
            description=description,
            affected_versions_value=affected_field,
            cwe_value=cwe,
            severity_value=severity,
            credits_value=credited_as,
            mailing_list_value=mailing_list,
            pr_value=pr_field,
            vendor=args.vendor,
            product=args.product,
            package_name=args.package_name,
            collection_url=args.collection_url,
            org_id=args.org_id,
            version_start=args.version_start,
            discovery=args.discovery,
            remediation_developers=combined_remediation_developers,
            advisory_urls=combined_advisory_urls,
            product_overrides=product_overrides,
            current_cve_id=cve_id,
        )
    except ValueError as exc:
        # parse_affected_versions (and any future fail-loud parser
        # called from build_cna_container) raises ValueError on
        # unparsable input. Surface it as a user-facing error and
        # exit non-zero rather than crashing with a traceback.
        print(f"error: {exc}", file=sys.stderr)
        return 2

    if args.no_envelope:
        payload: dict = cna
    else:
        payload = wrap_cve_record(
            cna,
            cve_id=cve_id,
            org_id=args.org_id,
            release_vote_in_progress=release_vote_in_progress,
        )

    text = emit_json(payload, args.output)
    if args.output is None:
        sys.stdout.write(text + "\n")
    else:
        print(f"Wrote {len(text.encode('utf-8'))} bytes to {args.output}")
        if cve_id:
            print(
                "Open the Vulnogram #source tab and paste the file:\n"
                f"  https://cveprocess.apache.org/cve5/{cve_id}#source"
            )

    if args.attach:
        try:
            attachment_url, was_update = attach_to_issue(
                issue_number=args.issue,
                repo=args.repo,
                cve_id=cve_id,
                json_text=text,
                cna=cna,
                cna_private_state=(
                    None
                    if args.no_envelope
                    else compute_cna_private_state(
                        cna, cve_id, release_vote_in_progress=release_vote_in_progress
                    )
                ),
            )
        except RuntimeError as exc:
            print(f"error: {exc}", file=sys.stderr)
            return 1
        action = "Replaced" if was_update else "Embedded"
        print(f"{action} CVE JSON in issue body on {args.repo}#{args.issue}")
        if attachment_url:
            print(f"  {attachment_url}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())
