Gate Releases on Dependency License Audit

id: pip-dependency-license-audit
namespace: company.team
description: |
  Install a Python project's dependencies, list their licenses, and alert when
  anything lands on the denylist (AGPL, GPL by default).

triggers:
  - id: pre_release_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Call from the release pipeline; the license gate runs before
      artifacts ship.
    key: license-audit
  - id: weekly_dependency_scan
    type: io.kestra.plugin.core.trigger.Schedule
    description: Weekly scan catches new transitive deps with surprise licenses;
      shipped disabled.
    cron: "0 2 * * 4"
    disabled: true

inputs:
  - id: repo_url
    type: STRING
    defaults: "https://github.com/owner/repo.git"
    description: Repo with requirements.txt or pyproject.toml.
  - id: denied_licenses
    type: STRING
    defaults: "GPL,AGPL,LGPL"
    description: Comma-separated license substrings that fail the gate.

tasks:
  - id: audit_licenses
    type: io.kestra.plugin.scripts.shell.Commands
    description: Clone, install deps, then pip-licenses --format=json; count hits
      against the denylist via the stdout outputs protocol. A broken audit
      reports worst case.
    containerImage: python:3.12-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    commands:
      - |
        apt-get update -qq && apt-get install -y -qq git >/dev/null 2>&1
        git clone --depth 1 --quiet "{{ inputs.repo_url }}" app
        cd app
        (pip install --quiet -r requirements.txt || pip install --quiet .) 2>/dev/null || true
        pip install --quiet pip-licenses
        pip-licenses --format=json > licenses.json
        cat > audit.py <<'PYEOF'
        import os, json
        denied = [d.strip().lower() for d in os.environ.get("DENIED", "").split(",") if d.strip()]
        try:
            pkgs = json.load(open("licenses.json"))
        except Exception:
            pkgs = []
        hits = [p["Name"] for p in pkgs if any(d in (p.get("License") or "").lower() for d in denied)]
        print(f"{len(pkgs)} packages, {len(hits)} with denied licenses")
        print("::" + json.dumps({"outputs": {"total": len(pkgs), "denied_hits": len(hits), "flagged": ",".join(hits)}}) + "::")
        PYEOF
        python3 audit.py
    env:
      DENIED: "{{ inputs.denied_licenses }}"

  - id: denied_found
    type: io.kestra.plugin.core.flow.If
    description: Any denied license is an alert; otherwise log the clean pass.
    condition: "{{ outputs.audit_licenses.vars.denied_hits > 0 }}"
    then:
      - id: alert_denied
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Name the flagged packages so the replacement search starts there.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": ":warning: License audit on {{ inputs.repo_url }}: {{ outputs.audit_licenses.vars.denied_hits }} package(s) match the denylist ({{ inputs.denied_licenses }}): {{ outputs.audit_licenses.vars.flagged }}"
          }
    else:
      - id: log_clean
        type: io.kestra.plugin.core.log.Log
        description: Record the pass for the compliance trail.
        message: "License audit clean across {{ outputs.audit_licenses.vars.total }}
          packages."

errors:
  - id: alert_audit_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert when the audit fails - no result must ever read as compliant.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "License audit FAILED in flow {{ flow.id }} (execution {{ execution.id }}). Check the repo and install logs."
      }

outputs:
  - id: audit_result
    type: JSON
    description: 'Audit counts, e.g. {"total": 41, "denied_hits": 0}'
    value: '{{ {"total": outputs.audit_licenses.vars.total, "denied_hits":
      outputs.audit_licenses.vars.denied_hits} | toJson }}'