PyPI Publish Provenance Gate with Kestra

id: pypi-publish-provenance-gate
namespace: company.team
description: |
  Verify a published GitHub release against the PyPI project record: the tag
  matches the served version, the sdist and the wheel are attached, no upload
  is yanked, and attestation metadata is checked when PyPI exposes it. A failed
  check opens a GitHub issue, alerts Slack, and fails the run so the gate is
  visibly red.

triggers:
  - id: release_published
    type: io.kestra.plugin.core.trigger.Webhook
    description: GitHub posts the release payload here. The key segment guards the
      URL, so treat it as a secret and not as the only thing standing between a
      stranger and a run.
    key: "{{ secret('RELEASE_WEBHOOK_KEY') }}"

  - id: daily_reverify
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled. Enable it for a daily re-check of the same
      package, which is when yanked or quietly replaced uploads usually surface.
    cron: "0 6 * * *"
    disabled: true

inputs:
  - id: package
    type: STRING
    defaults: your-package
    description: PyPI project name to verify, normalized to its canonical form
      before it is used in the project API request.
  - id: repository
    type: STRING
    defaults: your-org/your-repo
    description: GitHub repository that receives the gate issue when a webhook
      payload does not carry one, which is the case for scheduled and manual
      runs.
  - id: release_tag
    type: STRING
    defaults: latest
    description: Tag verified on scheduled and manual runs. Keep latest to check
      whatever PyPI currently serves, while webhook runs always use the tag from
      the payload.

tasks:
  # --- 1. parse the release payload ---
  - id: parse_release
    type: io.kestra.plugin.scripts.python.Script
    description: Pull the release tag, the repository, and the package name out of
      the webhook payload, falling back to the flow inputs so scheduled and
      manual runs take the same path.
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    env:
      RELEASE_PAYLOAD: "{{ trigger.body ?? '' }}"
      INPUT_TAG: "{{ inputs.release_tag }}"
      INPUT_REPOSITORY: "{{ inputs.repository }}"
      PACKAGE: "{{ inputs.package }}"
    script: |
      import json
      import os
      import re

      def as_object(value):
          if isinstance(value, dict):
              return value
          if isinstance(value, str):
              try:
                  decoded = json.loads(value)
              except ValueError:
                  return {}
              if isinstance(decoded, str):
                  try:
                      decoded = json.loads(decoded)
                  except ValueError:
                      return {}
              return decoded if isinstance(decoded, dict) else {}
          return {}

      data = as_object(os.environ.get("RELEASE_PAYLOAD", "") or "")
      release = as_object(data.get("release"))
      payload_repo = as_object(data.get("repository"))

      payload_tag = str(release.get("tag_name") or "").strip()
      input_tag = (os.environ.get("INPUT_TAG", "") or "").strip()
      tag = payload_tag or ("" if input_tag.lower() == "latest" else input_tag)
      if payload_tag:
          source = "webhook"
      elif tag:
          source = "input"
      else:
          source = "latest"

      repository = str(payload_repo.get("full_name") or "").strip()
      if not repository:
          repository = (os.environ.get("INPUT_REPOSITORY", "") or "").strip()

      package_raw = (os.environ.get("PACKAGE", "") or "").strip()
      if not package_raw and repository:
          package_raw = repository.split("/")[-1]
      package = re.sub(r"[-_.]+", "-", package_raw).lower()

      payload = {
          "tag": tag,
          "package": package,
          "repository": repository,
          "source": source,
          "release_url": str(release.get("html_url") or "").strip(),
          "prerelease": bool(release.get("prerelease") or False),
          "payload_parsed": bool(data),
      }
      print("parsed tag=%s package=%s repository=%s source=%s" % (
          payload["tag"] or "<latest>", payload["package"], payload["repository"], payload["source"]))
      print("::" + json.dumps({"outputs": payload}) + "::")

  # --- 2. read what PyPI actually serves ---
  - id: fetch_project_json
    type: io.kestra.plugin.core.http.Request
    description: GET the public PyPI project API without any credentials, so the
      gate compares against the record PyPI serves rather than against the
      pipeline log that claims to have uploaded.
    uri: "https://pypi.org/pypi/{{ outputs.parse_release.vars.package }}/json"
    method: GET
    headers:
      Accept: "application/json"
    options:
      allowFailed: true
      timeout:
        connectTimeout: PT10S
        readIdleTimeout: PT30S

  # --- 3. provenance verdict ---
  - id: assess_provenance
    type: io.kestra.plugin.scripts.python.Script
    description: Compare the release tag with the served version, require an sdist
      and a wheel, reject yanked uploads, and record uploader or attestation
      metadata whenever PyPI exposes it. Emits verdict and reasons through the
      outputs protocol.
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    env:
      RELEASE_TAG: "{{ outputs.parse_release.vars.tag }}"
      PACKAGE: "{{ outputs.parse_release.vars.package }}"
      REPOSITORY: "{{ outputs.parse_release.vars.repository }}"
      PYPI_CODE: "{{ outputs.fetch_project_json.code ?? 0 }}"
    inputFiles:
      project.json: "{{ outputs.fetch_project_json.body ?? '' }}"
    script: |
      import json
      import os

      def norm_version(value):
          text = (value or "").strip()
          if len(text) > 1 and text[0] in "vV" and text[1].isdigit():
              text = text[1:]
          return text.strip().lower()

      tag = (os.environ.get("RELEASE_TAG", "") or "").strip()
      package = (os.environ.get("PACKAGE", "") or "").strip()
      repository = (os.environ.get("REPOSITORY", "") or "").strip()
      try:
          http_code = int(os.environ.get("PYPI_CODE", "0") or 0)
      except ValueError:
          http_code = 0

      raw = ""
      try:
          with open("project.json") as handle:
              raw = handle.read()
      except OSError:
          raw = ""

      project = None
      if raw.strip():
          try:
              project = json.loads(raw)
          except ValueError:
              project = None

      checks = {}
      reasons = []
      latest = ""
      files = []
      sdist_files = []
      wheel_files = []
      yanked_files = []
      attestation_files = []
      uploader_files = []

      if http_code != 200 or not isinstance(project, dict):
          checks["pypi_record"] = False
          reasons.append("PyPI project API returned HTTP %s for %s instead of a JSON project record" % (
              http_code, package or "the configured package"))
      else:
          checks["pypi_record"] = True
          info = project.get("info") or {}
          latest = str(info.get("version") or "").strip()
          urls = project.get("urls")
          if not isinstance(urls, list):
              urls = []
          files = [f for f in urls if isinstance(f, dict)]
          if not files and latest and isinstance(project.get("releases"), dict):
              fallback = project["releases"].get(latest)
              if isinstance(fallback, list):
                  files = [f for f in fallback if isinstance(f, dict)]
          label = latest or "the latest version"

          if not tag:
              checks["tag_matches"] = bool(latest)
              tag = latest
              reasons.append("no release tag was supplied, so the gate verified the latest published version %s" % label)
          else:
              checks["tag_matches"] = bool(latest) and norm_version(tag) == norm_version(latest)
              if checks["tag_matches"]:
                  reasons.append("release tag %s matches the PyPI version %s" % (tag, latest))
              else:
                  reasons.append("release tag %s does not match the latest PyPI version %s" % (
                      tag, latest or "<unknown>"))

          sdist_files = [f for f in files
                         if f.get("packagetype") == "sdist"
                         or str(f.get("filename") or "").endswith((".tar.gz", ".tar.bz2", ".zip"))]
          checks["sdist_present"] = bool(sdist_files)
          if sdist_files:
              reasons.append("sdist archive present for %s" % label)
          else:
              reasons.append("no sdist archive is attached to %s" % label)

          wheel_files = [f for f in files if str(f.get("filename") or "").lower().endswith(".whl")]
          checks["wheel_present"] = bool(wheel_files)
          if wheel_files:
              reasons.append("wheel present for %s" % label)
          else:
              reasons.append("no wheel is attached to %s" % label)

          yanked_files = [f for f in files if f.get("yanked")]
          checks["uploads_not_yanked"] = not yanked_files
          if yanked_files:
              names = ", ".join(str(f.get("filename") or "unnamed") for f in yanked_files[:3])
              reasons.append("%d upload(s) yanked from %s (%s)" % (len(yanked_files), label, names))
          else:
              reasons.append("no upload on %s is yanked" % label)

          attestation_files = [f for f in files
                               if isinstance(f.get("attestations"), list) and f.get("attestations")]
          uploader_files = [f for f in files if f.get("uploader")]
          if attestation_files:
              checks["provenance_metadata"] = True
              reasons.append("attestation metadata present on %d of %d file(s) for %s" % (
                  len(attestation_files), len(files), label))
          elif uploader_files:
              checks["provenance_metadata"] = False
              reasons.append("uploader metadata is published but no attestation is attached to any file of %s" % label)
          else:
              checks["provenance_metadata"] = True
              reasons.append("PyPI exposes no uploader or attestation metadata for %s, so the check is recorded without blocking" % label)

          if not files:
              reasons.append("the project record lists no files for %s" % label)

      verdict = "pass" if all(checks.values()) else "fail"
      reasons_text = "; ".join(reasons) if reasons else "all provenance checks passed"

      outputs = {
          "verdict": verdict,
          "reasons": reasons,
          "reasons_text": reasons_text,
          "package": package,
          "repository": repository,
          "tag": tag,
          "pypi_version": latest,
          "file_count": len(files),
          "sdist_count": len(sdist_files),
          "wheel_count": len(wheel_files),
          "yanked_count": len(yanked_files),
          "attestation_count": len(attestation_files),
          "checks_passed": sum(1 for value in checks.values() if value),
          "checks_total": len(checks),
      }
      print("%s: %s" % (verdict.upper(), reasons_text))
      print("::" + json.dumps({"outputs": outputs}) + "::")

  # --- 4. gate the release ---
  - id: gate
    type: io.kestra.plugin.core.flow.If
    description: Any failed check blocks the release. The failing path opens a
      GitHub issue, posts the same reasons to Slack, and fails the execution so
      a waiting pipeline turns red.
    condition: "{{ outputs.assess_provenance.vars.verdict != 'pass' }}"
    then:
      - id: open_release_issue
        type: io.kestra.plugin.github.issues.Create
        description: Open an issue carrying the tag, the PyPI version, and every reason
          the gate recorded, so the mismatch is tracked where the release is
          managed.
        jwtToken: "{{ secret('GITHUB_TOKEN') }}"
        repository: "{{ outputs.parse_release.vars.repository }}"
        title: "PyPI provenance gate failed for {{ outputs.parse_release.vars.package }}
          {{ outputs.assess_provenance.vars.tag }}"
        body: |
          ## PyPI provenance gate failed

          **Package:** {{ outputs.parse_release.vars.package }}
          **Release tag:** {{ outputs.assess_provenance.vars.tag }}
          **PyPI latest version:** {{ outputs.assess_provenance.vars.pypi_version }}
          **GitHub prerelease:** {{ outputs.parse_release.vars.prerelease }}
          **Checks passed:** {{ outputs.assess_provenance.vars.checks_passed }} of {{ outputs.assess_provenance.vars.checks_total }}
          **Release:** {{ outputs.parse_release.vars.release_url }}

          {{ outputs.assess_provenance.vars.reasons_text }}

          Execution {{ execution.id }}. Fix the publish or the tag, then re-run the gate.

      - id: alert_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Post the failed checks to Slack so release owners see the block
          without opening Kestra.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "PyPI provenance gate failed for {{ outputs.parse_release.vars.package }} (tag {{ outputs.assess_provenance.vars.tag }}, PyPI {{ outputs.assess_provenance.vars.pypi_version }}): {{ outputs.assess_provenance.vars.reasons_text }}. Execution {{ execution.id }}."
          }

      - id: fail_gate
        type: io.kestra.plugin.core.execution.Fail
        description: Fail the execution with the recorded reasons as the error message,
          so the gate surfaces red instead of quietly completing.
        errorMessage: "Provenance gate failed for {{ outputs.parse_release.vars.package
          }} tag {{ outputs.assess_provenance.vars.tag }}: {{
          outputs.assess_provenance.vars.reasons_text }}"
    else:
      - id: log_pass
        type: io.kestra.plugin.core.log.Log
        description: Record the passing verdict so the execution history doubles as
          evidence that the published artifacts were checked.
        message: "Provenance gate passed for {{ outputs.parse_release.vars.package }}
          tag {{ outputs.assess_provenance.vars.tag }} (PyPI {{
          outputs.assess_provenance.vars.pypi_version }}, {{
          outputs.assess_provenance.vars.file_count }} file(s), source {{
          outputs.parse_release.vars.source }})."

errors:
  - id: slack_alert
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert when the gate itself cannot run, because an errored check and
      a verified release look identical unless they are worded apart.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "FAILED {{ flow.id }} run {{ execution.id }}: the PyPI provenance gate could not complete, so the release stays unverified. Check GITHUB_TOKEN, SLACK_WEBHOOK_URL, and worker network access."
      }

outputs:
  - id: verdict
    type: STRING
    description: 'Gate result, for example pass or fail.'
    value: "{{ outputs.assess_provenance.vars.verdict }}"
  - id: reasons
    type: STRING
    description: Every check the gate ran, joined with semicolons.
    value: "{{ outputs.assess_provenance.vars.reasons_text }}"
  - id: release_tag
    type: STRING
    description: Release tag that was verified, filled with the served PyPI version
      when the run supplied no tag.
    value: "{{ outputs.assess_provenance.vars.tag }}"
  - id: pypi_version
    type: STRING
    description: Version PyPI served for the package at check time.
    value: "{{ outputs.assess_provenance.vars.pypi_version }}"
  - id: file_count
    type: INT
    description: Number of files attached to that version on PyPI, for example 3.
    value: "{{ outputs.assess_provenance.vars.file_count }}"

A release that looks green in GitHub can still be wrong on PyPI. The tag can name a version that was never served, the wheel can be missing, an upload can be yanked an hour after the announcement, or the attestation metadata a supply chain review expects never appeared. This blueprint turns those facts into a gate that reads PyPI itself instead of trusting the pipeline log that claims to have uploaded. A GitHub release webhook starts the run, a Python step reads the payload, an unauthenticated request pulls the public project record, and a second Python step compares the tag with the version PyPI serves, requires an sdist and a wheel, rejects yanked uploads, and records uploader or attestation metadata whenever the API exposes it. Any failed check opens an issue on the repository, posts the same reasons to Slack, and fails the execution, so a waiting deploy pipeline stops instead of shipping.

How it works

The release_published trigger (io.kestra.plugin.core.trigger.Webhook) exposes a keyed URL for GitHub release events, and daily_reverify (io.kestra.plugin.core.trigger.Schedule, cron 0 6 * * *, shipped disabled: true) is there for a re-check after the fact.
parse_release (io.kestra.plugin.scripts.python.Script on the io.kestra.plugin.core.runner.Process runner) reads tag_name, repository.full_name, and the package name out of the payload with input fallbacks, normalizes the package to its canonical PyPI form, and emits tag, package, repository, source, and release_url through Kestra's outputs protocol. A payload that is missing or malformed falls back to the inputs instead of raising.
fetch_project_json (io.kestra.plugin.core.http.Request) issues an unauthenticated GET https://pypi.org/pypi/<package>/json with options.allowFailed: true, so an unknown project becomes a status code the verdict can reason about instead of an infrastructure error.
assess_provenance (io.kestra.plugin.scripts.python.Script) reads the project record and compares the release tag with the served version, stripping a leading v on both sides, then requires an sdist and a wheel, rejects any yanked upload, and records uploader or attestation metadata when PyPI publishes it. It emits verdict, reasons, the version, and the file counts through the outputs protocol.
gate (io.kestra.plugin.core.flow.If) branches on verdict != pass. The failing path opens an issue with io.kestra.plugin.github.issues.Create, posts the same reasons through io.kestra.plugin.slack.notifications.SlackIncomingWebhook, and ends with io.kestra.plugin.core.execution.Fail; the passing path logs the evidence for the execution history.
The errors block alerts Slack when the gate itself cannot run, since a broken gate must never be mistaken for a verified release.

What you get

A verdict that answers one question with PyPI's own record: does the published tag, the served version, and the attached file set agree.
Typed flow outputs (verdict, reasons, release_tag, pypi_version, file_count) that a parent release pipeline can branch on without scraping logs.
Escalation in three places at once, a GitHub issue on the repository, a Slack message with the failed checks, and a red execution for anything watching run status.
A dated record of every check in Kestra's execution history, so you can open last month's release and see exactly what the gate decided and why.
A disabled daily schedule you can enable to re-verify artifacts after the fact, which is when yanks and replaced files usually show up.
No PyPI credentials anywhere in the flow, because the project API it reads is public.

Who it's for

Release engineers who publish to PyPI from GitHub releases and want a check that does not trust their own pipeline.
Security and platform teams answering supply chain questions about what actually serves for a given tag.
Maintainers of shared packages who need a red run and a tracked issue the moment an upload goes missing, gets yanked, or loses its provenance metadata.

Why orchestrate this with Kestra

A webhook in CI can call a PyPI API and post a message, but the hard part is what happens next: who records the verdict, who re-checks tomorrow, and what turns the deploy red when the record changes after the fact. Kestra makes the gate a first class workflow. The event trigger and the schedule live in the same file, every check is a logged step whose inputs and outputs are preserved, the branch that blocks the release is declarative rather than buried in a shell script, and the error handler keeps alerting honest when the gate itself cannot run. The execution history doubles as the audit trail: for any past release you can open the run and see the tag, the version PyPI served, the file counts, and the reasons behind the decision. Replaying a webhook or widening the schedule to hourly is configuration, not another script to deploy somewhere.

Prerequisites

A Kestra worker with Python 3 available to the process runner, or switch the two Python tasks to a container task runner.
A GitHub repository where the provided token can open issues, classic repo scope or fine-grained Issues write permission.
A Slack incoming webhook for block notices and failure alerts.
A GitHub repository webhook pointing at the trigger URL and configured to send release events.
The PyPI project name to verify. No PyPI account or token is needed, since the project API used here is public.

Secrets

GITHUB_TOKEN: token used to open the gate issue when a check fails.
SLACK_WEBHOOK_URL: Slack incoming webhook URL used for block notices and error alerts.
RELEASE_WEBHOOK_KEY: key segment guarding the webhook URL. Anyone holding it can start a run, so treat it as a secret.

Quick start

Add the three secrets to the company.team namespace.
Set package to your PyPI project name and repository to the GitHub repository that should receive gate issues.
Point a GitHub webhook at /api/v1/{tenant}/executions/webhook/company.team/pypi-publish-provenance-gate/{RELEASE_WEBHOOK_KEY} and restrict it to release events. A payload without a release tag simply asks the gate to verify the latest version PyPI serves.
Publish a GitHub release whose tag matches PyPI and confirm the run ends with verdict: pass. Then push a tag PyPI does not know, or yank an upload, and confirm the issue, the Slack alert, and the red execution.
For the daily re-check, set daily_reverify.disabled to false and leave release_tag at latest.

How to extend

Cover several packages by calling this flow once per package from a parent flow with io.kestra.plugin.core.flow.Subflow and collecting each verdict output.
Make the gate blocking in your pipeline by branching on the verdict output of the parent call instead of parsing logs.
Check a second index, such as an internal mirror, with another io.kestra.plugin.core.http.Request and a third verdict task that compares the two records.
Persist the full project record next to the verdict with io.kestra.plugin.core.storage.Write, then link the storage URI from the GitHub issue so auditors can inspect the raw evidence.
Add a PEP 440 aware comparison (for example local version segments) by extending the version normalization helper in assess_provenance.

Gate PyPI Publishing on Release Artifact Provenance and Uploads

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Gate PyPI Publishing on Release Artifact Provenance and Uploads

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

Pulsar Topic Activity Sentinel

Apache Pulsar multi-tenant backlog quarantine and priority event router

Run Python script in a Docker container based on Google Artifact Registry

Push code to Git at regular intervals

Qdrant Vector Index Health and Memory Optimizer

Pull a container image from Amazon ECR registry and run a Python script

More Core Blueprints

Run a Python script and generate outputs, metrics and files specified with a variable

Parametrized flow with custom validators to ensure correct integer value

Interactively purge logs, executions, and internal storage files to free up disk space

Add multiple Redis keys in parallel from JSON input

Purge execution data including logs, metrics and outputs on a schedule

Create a Python subflow, acting like an abstracted component

New to Kestra?