fix(security): confine MCP credential paths; publish only announced records

plugins/magpie-security-tools/tools/cve-tool-vulnogram/oauth-api/src/vulnogram_api/record_publish.py

@@ -37,6 +37,12 @@
 3. POSTs the modified document back via the same path
    :class:`vulnogram_api.client.update_record` already uses.
 
+The script refuses the move unless the record carries the evidence the
+gate stands for: a ``references[]`` entry tagged ``vendor-advisory`` (the
+archived advisory URL, which ``generate-cve-json`` tags that way once the
+tracker's *Public advisory URL* is set). A record without one has not
+been announced, so clearing it for ``cve.org`` would disclose early.
+
 If the record is not currently in ``REVIEW`` state, the script
 refuses the transition with a non-zero exit (a sync that re-runs
 the publish on a tracker that already published is a bug, not an
@@ -65,6 +71,9 @@
 from vulnogram_api.credentials import Session, locate_session
 
 CVE_ID_RE = re.compile(r"^CVE-\d{4}-\d{4,7}$")
+#: The only states a record may be published from: a DRAFT (or anything
+#: earlier) has not been through review.
+PUBLISHABLE_STATES = frozenset({"REVIEW", "READY"})
 
 
 class UnexpectedRecordState(VulnogramAPIError):
@@ -98,10 +107,11 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
         "--allow-state",
         action="append",
         default=None,
+        choices=sorted(PUBLISHABLE_STATES),
         help=(
-            "Accepted current states. Default: REVIEW only. Pass multiple "
-            "times to allow more (e.g. --allow-state REVIEW --allow-state READY). "
-            "Use sparingly — refusing unexpected states is the safety net."
+            "Accepted current states. Default: REVIEW only. Pass both to also "
+            "accept a record moved to READY by hand (--allow-state REVIEW "
+            "--allow-state READY). Nothing earlier than REVIEW can be published."
         ),
     )
     ap.add_argument(
@@ -112,6 +122,22 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     return ap.parse_args(argv)
 
 
+def _has_vendor_advisory(body: dict[str, object]) -> bool:
+    """Whether the CVE document lists a reference tagged ``vendor-advisory``."""
+    containers = body.get("containers")
+    cna = containers.get("cna") if isinstance(containers, dict) else None
+    references = cna.get("references") if isinstance(cna, dict) else None
+    if not isinstance(references, list):
+        return False
+    return any(
+        isinstance(ref, dict)
+        and isinstance(ref.get("url"), str)
+        and ref["url"]
+        and "vendor-advisory" in (ref.get("tags") or [])
+        for ref in references
+    )
+
+
 def _state(document: dict[str, object]) -> str | None:
     cna = record_body(document).get("CNA_private")
     if not isinstance(cna, dict):
@@ -167,6 +193,15 @@ def main(argv: list[str] | None = None) -> int:
             file=sys.stderr,
         )
         return 3
+    if not _has_vendor_advisory(body):
+        print(
+            f"✗ {args.cve_id} has no references[] entry tagged 'vendor-advisory' "
+            f"(the archived advisory URL). Publishing it would clear an "
+            f"unannounced record for cve.org; set the tracker's Public advisory "
+            f"URL and regenerate the CVE JSON first.",
+            file=sys.stderr,
+        )
+        return 4
     body["CNA_private"]["state"] = "PUBLIC"
 
     if args.dry_run:

plugins/magpie-security-tools/tools/gmail/oauth-draft/src/oauth_draft/mcp_server.py

@@ -66,6 +66,7 @@
 from __future__ import annotations
 
 import contextlib
+import os
 import pathlib
 from collections.abc import Iterator
 from typing import Any
@@ -75,6 +76,7 @@
 from oauth_draft import create_draft as _cd
 from oauth_draft import setup_creds as _setup
 from oauth_draft.credentials import (
+    DEFAULT_CREDENTIALS_DIR,
     DEFAULT_CREDENTIALS_PATH,
     Credentials,
     locate_credentials,
@@ -84,6 +86,41 @@
 mcp = MCPServer("gmail-plaintext")
 
 
+def _vetted_credentials_path(credentials_path: str | None) -> str | None:
+    """A caller-supplied credentials path, confined to the operator's credential directories.
+
+    This server runs outside the agent's sandbox, and the path is a tool
+    argument, so whoever drives the agent chooses it. Unconfined,
+    `setup_credentials` would write the refresh token wherever a sandboxed
+    command can read it, and the other tools would load a planted file. So a
+    path is accepted only as a `.json` file directly in the default directory
+    (~/.config/apache-magpie/) or in the directory of $GMAIL_OAUTH_CREDENTIALS,
+    which comes from this server's own configuration; and never through a
+    symlink, at the file or on the way to it. `None` keeps the default lookup.
+    """
+    if credentials_path is None:
+        return None
+    path = pathlib.Path(credentials_path).expanduser()
+    if not path.is_absolute() or path.suffix != ".json":
+        raise ValueError(f"credentials_path must be an absolute path to a .json file: {credentials_path}")
+    allowed = {DEFAULT_CREDENTIALS_DIR}
+    env = os.environ.get("GMAIL_OAUTH_CREDENTIALS")
+    if env:
+        allowed.add(pathlib.Path(env).expanduser().parent)
+    parent = path.parent
+    real_parent = pathlib.Path(os.path.realpath(parent))
+    if (
+        parent != real_parent
+        or path.is_symlink()
+        or not any(real_parent == pathlib.Path(os.path.realpath(d)) for d in allowed)
+    ):
+        raise ValueError(
+            f"credentials_path must sit directly in {', '.join(sorted(str(d) for d in allowed))} "
+            f"(no symlinks): {credentials_path}"
+        )
+    return str(path)
+
+
 @contextlib.contextmanager
 def _as_tool_error() -> Iterator[None]:
     """Convert the CLI helpers' ``SystemExit`` into a normal tool error.
@@ -110,6 +147,7 @@ def _create_draft_impl(
     credentials_path: str | None,
 ) -> dict[str, str]:
     """Shared implementation, kept plain so the tests can call it directly."""
+    credentials_path = _vetted_credentials_path(credentials_path)
     with _as_tool_error():
         creds = Credentials.load(locate_credentials(credentials_path))
         assert creds.from_address is not None  # require_from_address=True default
@@ -214,10 +252,13 @@ def setup_credentials(
             $GMAIL_FROM, then `git config user.email`.
         credentials_path: Where to write the credentials JSON. Defaults to
             $GMAIL_OAUTH_CREDENTIALS or ~/.config/apache-magpie/gmail-oauth.json.
+            Only a .json file directly in ~/.config/apache-magpie/ or in the
+            directory of $GMAIL_OAUTH_CREDENTIALS is accepted.
 
     Returns:
         ``{credentials_path, from_address, scopes}``.
     """
+    credentials_path = _vetted_credentials_path(credentials_path)
     resolved_from = from_address or _setup.detect_from_address()
     if not resolved_from:
         raise ValueError(
@@ -260,11 +301,14 @@ def check_auth(credentials_path: str | None = None) -> dict[str, str]:
     Args:
         credentials_path: Override the credentials JSON path. Defaults to
             $GMAIL_OAUTH_CREDENTIALS or ~/.config/apache-magpie/gmail-oauth.json.
+            Only a .json file directly in ~/.config/apache-magpie/ or in the
+            directory of $GMAIL_OAUTH_CREDENTIALS is accepted.
 
     Returns:
         ``{status: "ok", credentials_path, from_address}`` when auth works;
         raises with a helpful message when credentials are missing or invalid.
     """
+    credentials_path = _vetted_credentials_path(credentials_path)
     with _as_tool_error():
         path = locate_credentials(credentials_path)
         creds = Credentials.load(path, require_from_address=False)

plugins/magpie-security/skills/issue-sync/apply-and-push.md

@@ -157,8 +157,8 @@ Use:
 - **Vulnogram state transition (`REVIEW → PUBLIC`):** invoke the
   [`vulnogram-api-record-publish`](../../../../tools/cve-tool-vulnogram/oauth-api/README.md)
   CLI to flip the record's `CNA_private.state` over the OAuth API.
-  It refuses the transition unless the current state is `REVIEW`;
-  widen with `--allow-state` only when explicitly justified (e.g. a record already moved to `READY` manually):
+  It refuses the transition unless the current state is `REVIEW` and the record already lists the archived advisory URL as a `vendor-advisory` reference (re-push the CVE JSON after capturing the URL, before this step);
+  `--allow-state READY` also accepts a record moved to `READY` manually, and no earlier state can be published:
 
   ```bash
   uv run --project <framework>/tools/cve-tool-vulnogram/oauth-api \
@@ -170,7 +170,7 @@ Use:
   Outside that trigger, the state transition stays manual.
 
   Idempotent: re-running on a record already in `PUBLIC` exits 0 with an informational message.
-  Exit codes match `record-update` (2 = token expired, 3 = unexpected state, 5 = save failed, 6 = other API error, 7 = unexpected envelope).
+  Exit codes match `record-update` (2 = token expired, 3 = unexpected state, 5 = save failed, 6 = other API error, 7 = unexpected envelope), plus 4 = no `vendor-advisory` reference.
   On a non-zero exit, the combined apply stops and the failure surfaces in the recap;
   the partial state (URL captured, labels flipped, JSON re-pushed, tracker NOT yet closed) is the recovery starting point for the next sync.
 - **Advisory short-summary extraction:** when the *Advisory archived on `<users-list>`* combined apply fires (Step 2b row),

tools/cve-tool-vulnogram/oauth-api/src/vulnogram_api/record_publish.py

@@ -37,6 +37,12 @@
 3. POSTs the modified document back via the same path
    :class:`vulnogram_api.client.update_record` already uses.
 
+The script refuses the move unless the record carries the evidence the
+gate stands for: a ``references[]`` entry tagged ``vendor-advisory`` (the
+archived advisory URL, which ``generate-cve-json`` tags that way once the
+tracker's *Public advisory URL* is set). A record without one has not
+been announced, so clearing it for ``cve.org`` would disclose early.
+
 If the record is not currently in ``REVIEW`` state, the script
 refuses the transition with a non-zero exit (a sync that re-runs
 the publish on a tracker that already published is a bug, not an
@@ -65,6 +71,9 @@
 from vulnogram_api.credentials import Session, locate_session
 
 CVE_ID_RE = re.compile(r"^CVE-\d{4}-\d{4,7}$")
+#: The only states a record may be published from: a DRAFT (or anything
+#: earlier) has not been through review.
+PUBLISHABLE_STATES = frozenset({"REVIEW", "READY"})
 
 
 class UnexpectedRecordState(VulnogramAPIError):
@@ -98,10 +107,11 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
         "--allow-state",
         action="append",
         default=None,
+        choices=sorted(PUBLISHABLE_STATES),
         help=(
-            "Accepted current states. Default: REVIEW only. Pass multiple "
-            "times to allow more (e.g. --allow-state REVIEW --allow-state READY). "
-            "Use sparingly — refusing unexpected states is the safety net."
+            "Accepted current states. Default: REVIEW only. Pass both to also "
+            "accept a record moved to READY by hand (--allow-state REVIEW "
+            "--allow-state READY). Nothing earlier than REVIEW can be published."
         ),
     )
     ap.add_argument(
@@ -112,6 +122,22 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     return ap.parse_args(argv)
 
 
+def _has_vendor_advisory(body: dict[str, object]) -> bool:
+    """Whether the CVE document lists a reference tagged ``vendor-advisory``."""
+    containers = body.get("containers")
+    cna = containers.get("cna") if isinstance(containers, dict) else None
+    references = cna.get("references") if isinstance(cna, dict) else None
+    if not isinstance(references, list):
+        return False
+    return any(
+        isinstance(ref, dict)
+        and isinstance(ref.get("url"), str)
+        and ref["url"]
+        and "vendor-advisory" in (ref.get("tags") or [])
+        for ref in references
+    )
+
+
 def _state(document: dict[str, object]) -> str | None:
     cna = record_body(document).get("CNA_private")
     if not isinstance(cna, dict):
@@ -167,6 +193,15 @@ def main(argv: list[str] | None = None) -> int:
             file=sys.stderr,
         )
         return 3
+    if not _has_vendor_advisory(body):
+        print(
+            f"✗ {args.cve_id} has no references[] entry tagged 'vendor-advisory' "
+            f"(the archived advisory URL). Publishing it would clear an "
+            f"unannounced record for cve.org; set the tracker's Public advisory "
+            f"URL and regenerate the CVE JSON first.",
+            file=sys.stderr,
+        )
+        return 4
     body["CNA_private"]["state"] = "PUBLIC"
 
     if args.dry_run:

tools/cve-tool-vulnogram/oauth-api/tests/test_record_publish.py

@@ -36,13 +36,24 @@ def _write_session(path):
     return path
 
 
-def _wrapped(state, **extra):
+ADVISORY = {
+    "cna": {
+        "references": [
+            {"url": "https://lists.apache.org/thread/abc", "tags": ["vendor-advisory"]},
+        ]
+    }
+}
+
+
+def _wrapped(state, advisory=True, **extra):
     """A record shaped the way Vulnogram's read API returns it: the CVE
     document (``CNA_private``, ``cveMetadata``, …) nested under a ``body``
     envelope. Publishing must read and write ``CNA_private`` under ``body``,
     not at the top level — mocking the unwrapped shape here is what let the
     body-envelope bug ship."""
     body = {"CNA_private": {"state": state}, **extra}
+    if advisory:
+        body.setdefault("containers", ADVISORY)
     return {"comments": [], "files": [], "body": body, "author": "x", "__v": 1}
 
 
@@ -163,3 +174,40 @@ def _raise_expired(*a, **kw):
     assert rc == 2
     err = capsys.readouterr().err
     assert "session expired" in err
+
+
+def test_a_record_without_a_vendor_advisory_is_not_published(tmp_path, monkeypatch, capsys):
+    """The gate is "the advisory has shipped": no archived advisory URL, no publish."""
+    _write_session(tmp_path / "session.json")
+    monkeypatch.setenv("VULNOGRAM_SESSION", str(tmp_path / "session.json"))
+    posted = []
+    monkeypatch.setattr(record_publish, "get_record", lambda *a, **kw: _wrapped("REVIEW", advisory=False))
+
+    def _fake_update(*a, **kw):
+        posted.append(a)
+        return {"type": "saved"}
+
+    monkeypatch.setattr(record_publish, "update_record", _fake_update)
+    rc = record_publish.main(["--cve-id", "CVE-2026-12345"])
+    assert rc == 4
+    assert posted == []
+    assert "vendor-advisory" in capsys.readouterr().err
+
+
+def test_a_patch_reference_is_not_an_advisory(tmp_path, monkeypatch):
+    _write_session(tmp_path / "session.json")
+    monkeypatch.setenv("VULNOGRAM_SESSION", str(tmp_path / "session.json"))
+    patch_only = {"cna": {"references": [{"url": "https://github.com/a/b/pull/1", "tags": ["patch"]}]}}
+    monkeypatch.setattr(
+        record_publish,
+        "get_record",
+        lambda *a, **kw: _wrapped("REVIEW", advisory=False, containers=patch_only),
+    )
+    assert record_publish.main(["--cve-id", "CVE-2026-12345", "--dry-run"]) == 4
+
+
+@pytest.mark.parametrize("state", ["DRAFT", "PUBLIC", "REJECTED"])
+def test_allow_state_cannot_widen_past_review(state):
+    with pytest.raises(SystemExit) as excinfo:
+        record_publish.parse_args(["--cve-id", "CVE-2026-12345", "--allow-state", state])
+    assert excinfo.value.code == 2