id: dependency-license-flip-radar
namespace: company.platform
description: |
Weekday check of a dependency watchlist for relicensing. Each package's pinned
version is compared with its latest release using license data from deps.dev.
A move into source-available or network-copyleft terms (BUSL, SSPL, Elastic
License, AGPL and similar) is reported to Slack once, together with the first
version, in publish order, where the license changed.
inputs:
- id: packages
type: ARRAY
itemType: STRING
displayName: Dependency watchlist
description: Each entry is system:name@version. Systems are npm, pypi, go,
maven, cargo, nuget and rubygems. Maven names are group:artifact, and npm
scoped names keep their @scope. A webhook call can replace this list with
a packages array in its JSON body.
defaults:
- npm:express@4.18.2
- pypi:requests@2.31.0
- go:github.com/hashicorp/terraform@v1.5.7
- maven:org.elasticsearch:elasticsearch@7.10.2
- id: restricted_licenses
type: ARRAY
itemType: STRING
displayName: License fragments to treat as restrictive
description: Case-insensitive fragments matched against each license string
deps.dev returns, so SPDX ids and plain names are both caught. Edit this
to match your own legal policy.
defaults:
- busl
- business source
- sspl
- server side public
- elastic license
- elastic-2
- functional source
- fsl-
- polyform
- commons clause
- agpl
- affero
- cc-by-nc
- noncommercial
- non-commercial
- redis source available
- id: dry_run
type: BOOL
displayName: Dry run
description: Log the Slack alert instead of sending it, and leave the alert
state untouched so the first live run still reports everything. Leave this
on until the findings look right.
defaults: true
variables:
baseline_key: license_flip_radar_alerted
triggers:
- id: weekdays
type: io.kestra.plugin.core.trigger.Schedule
description: Relicensing is rare and announced on the maintainer's schedule, not
yours, so the watchlist is rechecked every weekday at 08:00.
cron: "0 8 * * 1-5"
- id: dependency_change
type: io.kestra.plugin.core.trigger.Webhook
description: >
Lets CI audit new dependencies as soon as a lockfile changes. POST a JSON
body such as {"packages": ["npm:left-pad@1.3.0"]}. Anyone holding the key
can start a run, so keep it in a secret.
key: "{{ secret('LICENSE_RADAR_WEBHOOK_KEY') }}"
tasks:
- id: read_alert_state
type: io.kestra.plugin.core.kv.Get
description: The license set already reported for each package, so the same flip
is not announced every weekday. Missing on the first run, which is not an
error.
key: "{{ render(vars.baseline_key) }}"
errorOnMissing: false
- id: audit_licenses
type: io.kestra.plugin.scripts.python.Script
description: >
Looks each package up on the deps.dev v3 API, compares the license of the
pinned version with the default (latest) version, and classifies the
change. CRITICAL means every latest license is restrictive and the pin had
none. WARNING means a restrictive option appears next to a permissive one,
for example in a dual license. Anything else that changed is INFO. For
CRITICAL and WARNING it bisects the version history to name the first
version, in publish order, whose license differs from the pin. A package
deps.dev cannot find, or has no license data for, is reported as such and
never treated as clean. If no package can be resolved at all the task
fails.
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
containerImage: python:3.12-slim
env:
RESTRICTED: "{{ inputs.restricted_licenses | toJson }}"
inputFiles:
packages.json: "{{ (trigger.body.packages ?? inputs.packages) | toJson }}"
state.json: "{{ (outputs.read_alert_state.value ?? '') | toJson }}"
script: |
import json
import os
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
SYSTEMS = {"npm", "pypi", "go", "maven", "cargo", "nuget", "rubygems"}
SPEC = re.compile(r"^([A-Za-z]+):(.+)@([^@\s]+)$")
RESTRICTED = [str(t).strip().lower() for t in json.loads(os.environ.get("RESTRICTED", "[]")) if str(t).strip()]
BASE = "https://api.deps.dev/v3/systems/{}/packages/{}"
def fetch(url):
request = urllib.request.Request(url, headers={
"User-Agent": "kestra-license-flip-radar", "Accept": "application/json"})
for attempt in range(1, 4):
try:
with urllib.request.urlopen(request, timeout=30) as response:
return json.load(response)
except urllib.error.HTTPError as error:
if error.code == 404:
return None
if (error.code != 429 and error.code < 500) or attempt == 3:
raise
except urllib.error.URLError:
if attempt == 3:
raise
time.sleep(2 * attempt)
def quote(value):
return urllib.parse.quote(value, safe="")
def read_state():
try:
text = open("state.json", encoding="utf-8").read().strip()
data = json.loads(text) if text else {}
if isinstance(data, str):
data = json.loads(data)
except (OSError, ValueError):
data = {}
alerted = data.get("alerted") if isinstance(data, dict) else None
return dict(alerted) if isinstance(alerted, dict) else {}
def mentions_restricted(entry):
lowered = entry.lower()
return any(token in lowered for token in RESTRICTED)
def entry_restricted(entry):
# SPDX: AND binds tighter than OR. An OR expression is restrictive only if every option is.
options = re.split(r"\s+or\s+", entry, flags=re.IGNORECASE)
return all(any(mentions_restricted(part) for part in re.split(r"\s+and\s+", option, flags=re.IGNORECASE))
for option in options)
def all_restricted(licenses):
return bool(licenses) and all(entry_restricted(e) for e in licenses)
def any_mention(licenses):
return any(mentions_restricted(e) for e in licenses)
def licenses_of(system, name, version):
data = fetch(BASE.format(system, quote(name)) + "/versions/" + quote(version))
if data is None:
return None
return sorted({str(x) for x in data.get("licenses") or []})
def first_flip(system, name, versions, pinned, default, pinned_licenses):
dated = sorted((v for v in versions if v["published"]), key=lambda v: v["published"])
names = [v["version"] for v in dated]
if pinned not in names or default not in names:
return None
lo, hi = names.index(pinned) + 1, names.index(default)
calls = 0
while lo < hi and calls < 16:
mid = (lo + hi) // 2
calls += 1
found = licenses_of(system, name, names[mid])
if found and found != pinned_licenses:
hi = mid
else:
lo = mid + 1
if lo > hi:
return None
return {"version": names[lo], "published": dated[names.index(names[lo])]["published"][:10]}
specs = json.load(open("packages.json", encoding="utf-8"))
previous = read_state()
next_state = dict(previous)
rows, alerts = [], []
counts = {"critical": 0, "warning": 0, "info": 0, "unchanged": 0, "unknown": 0, "unresolved": 0}
resolved = 0
for spec in specs:
match = SPEC.match(str(spec).strip())
if not match or match.group(1).lower() not in SYSTEMS:
counts["unresolved"] += 1
rows.append("| {} | - | - | UNRESOLVED | not system:name@version with a supported system |".format(spec))
continue
system, name, pinned = match.group(1).lower(), match.group(2), match.group(3)
key = "{}:{}".format(system, name)
try:
package = fetch(BASE.format(system, quote(name)))
if package is None:
raise LookupError("package not found on deps.dev")
versions = [{"version": v["versionKey"]["version"], "published": v.get("publishedAt") or "",
"default": bool(v.get("isDefault"))} for v in package.get("versions", [])]
default = next((v["version"] for v in versions if v["default"]), None)
if default is None and versions:
default = sorted(versions, key=lambda v: v["published"])[-1]["version"]
if default is None:
raise LookupError("no versions listed")
old = licenses_of(system, name, pinned)
if old is None:
raise LookupError("pinned version {} not found".format(pinned))
new = licenses_of(system, name, default)
except (LookupError, urllib.error.URLError, urllib.error.HTTPError) as error:
counts["unresolved"] += 1
rows.append("| {} | {} | - | UNRESOLVED | {} |".format(key, pinned, error))
continue
resolved += 1
if not old or not new:
counts["unknown"] += 1
next_state.pop(key, None)
rows.append("| {} | {} | {} | UNKNOWN | deps.dev has no license data for {} |".format(
key, pinned, default, "the pinned version" if not old else "the latest version"))
continue
if old == new:
counts["unchanged"] += 1
next_state.pop(key, None)
rows.append("| {} | {} | {} | UNCHANGED | {} |".format(key, pinned, default, ", ".join(new)))
continue
if all_restricted(new) and not any_mention(old):
severity = "CRITICAL"
elif any_mention(new) and not any_mention(old):
severity = "WARNING"
else:
severity = "INFO"
counts[severity.lower()] += 1
change = "{} -> {}".format(", ".join(old), ", ".join(new))
flip_text = ""
if severity in ("CRITICAL", "WARNING"):
flip = first_flip(system, name, versions, pinned, default, old)
flip_text = "first different license at {} (published {})".format(flip["version"], flip["published"]) if flip else "first flipped version could not be determined"
fingerprint = "|".join(new)
next_state[key] = fingerprint
if previous.get(key) != fingerprint:
marker = ":red_circle:" if severity == "CRITICAL" else ":warning:"
alerts.append("{} *{}* `{}` pinned {} is {}; latest {} is {}; {}. https://deps.dev/{}/{}".format(
marker, severity, key, pinned, ", ".join(old), default, ", ".join(new), flip_text, system, quote(name)))
else:
next_state.pop(key, None)
rows.append("| {} | {} | {} | {} | {}{} |".format(
key, pinned, default, severity, change, "; " + flip_text if flip_text else ""))
if specs and not resolved:
sys.exit("None of the {} package(s) could be resolved on deps.dev. Not reporting a clean check.".format(len(specs)))
shown = alerts[:15] + (["...and {} more".format(len(alerts) - 15)] if len(alerts) > 15 else [])
alert_text = ""
if alerts:
alert_text = ":rotating_light: *{} dependency license change(s) need review*\n{}".format(
len(alerts), "\n".join(shown))
summary = "\n".join([
"{} package(s): {} critical, {} warning, {} info, {} unchanged, {} unknown, {} unresolved. {} new alert(s).".format(
len(specs), counts["critical"], counts["warning"], counts["info"], counts["unchanged"],
counts["unknown"], counts["unresolved"], len(alerts)),
"| package | pinned | latest | severity | detail |",
"| --- | --- | --- | --- | --- |",
] + rows)
print(summary)
print("::" + json.dumps({"outputs": {
"package_count": len(specs),
"critical_count": counts["critical"],
"warning_count": counts["warning"],
"info_count": counts["info"],
"unresolved_count": counts["unresolved"],
"has_alerts": len(alerts) > 0,
"alert_text": alert_text,
"summary": summary,
"state_json": json.dumps({"alerted": next_state}, sort_keys=True),
}}) + "::")
- id: report
type: io.kestra.plugin.core.log.Log
description: The full table, so a run with no new alert still shows what was
checked and why nothing was raised.
message: "{{ outputs.audit_licenses.vars.summary }}"
- id: alert
type: io.kestra.plugin.core.flow.If
description: Only a restrictive license that has not been reported yet pages
anyone. Everything else stays in the log.
condition: "{{ outputs.audit_licenses.vars.has_alerts }}"
then:
- id: delivery_mode
type: io.kestra.plugin.core.flow.If
condition: "{{ inputs.dry_run }}"
then:
- id: would_send
type: io.kestra.plugin.core.log.Log
message: |
DRY RUN would post this to Slack:
{{ outputs.audit_licenses.vars.alert_text }}
else:
- id: notify_slack
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: One message for the whole run, not one per package.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{{ {'text': outputs.audit_licenses.vars.alert_text} | toJson }}
- id: record_alert_state
type: io.kestra.plugin.core.kv.Set
description: Last task on purpose. The state only advances after the alert was
delivered, so a failed Slack call repeats the alert next run instead of
swallowing it. Dry runs never write it.
runIf: "{{ not inputs.dry_run }}"
key: "{{ render(vars.baseline_key) }}"
kvType: JSON
overwrite: true
value: "{{ outputs.audit_licenses.vars.state_json }}"
errors:
- id: state_untouched
type: io.kestra.plugin.core.log.Log
level: ERROR
message: "License flip check failed in execution {{ execution.id }}. The alert
state under {{ render(vars.baseline_key) }} was not advanced, so the next
run repeats any alert that was not delivered. Do not read this run as a
clean check."
outputs:
- id: package_count
type: INT
description: Watchlist entries checked, e.g. 4.
value: "{{ outputs.audit_licenses.vars.package_count }}"
- id: critical_count
type: INT
description: Packages whose latest license is entirely restrictive and whose pin
was not, e.g. 1.
value: "{{ outputs.audit_licenses.vars.critical_count }}"
- id: warning_count
type: INT
description: Packages that gained a restrictive option beside a permissive one, e.g. 0.
value: "{{ outputs.audit_licenses.vars.warning_count }}"
- id: unresolved_count
type: INT
description: Entries deps.dev could not resolve or that were malformed, e.g. 0.
value: "{{ outputs.audit_licenses.vars.unresolved_count }}"
