Solana Token and Program Authority Watchdog with Kestra

id: solana-authority-watchdog
namespace: company.team
description: |
  Watch the Solana tokens and programs your product depends on. Reports
  changes to mint and freeze authorities, Token-2022 extension authorities
  (permanent delegate, transfer hook, fees), program upgrade authorities and
  program redeploys, plus sudden supply jumps. Each change is compared with
  an approved baseline kept in the Kestra KV store, and on-call approves it
  with a note.

triggers:
  - id: account_changed
    type: io.kestra.plugin.core.trigger.Webhook
    description: Point a Helius or QuickNode webhook for the watched mints and
      programs at this URL, so a change is checked within seconds.
    key: solana-watchdog-change-me
  - id: periodic_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Catches every change, and measures supply growth between checks.
    cron: "*/10 * * * *"

inputs:
  - id: mints
    type: ARRAY
    itemType: STRING
    defaults:
      - USDT=Es9vMFrzaCERmJfrF4H2FYD4KCoNkY11McCe8BenwNYB
      - PYUSD=2b1kV6DkPAnxd5ixfnxCpjxmKwqjjaYmCZfHsFu24GXo
    description: Token mints to watch, as `Label=address` (SPL Token or Token-2022).
      The defaults are mainnet mints.
  - id: programs
    type: ARRAY
    itemType: STRING
    defaults:
      - Raydium AMM v4=675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8
      - Orca Whirlpool=whirLbMiicVdio4qvUfM5KAg6Ct8VwpYzGff3uctyCc
    description: Programs to watch, as `Label=programId`. The defaults are mainnet programs.
  - id: supply_jump_percent
    type: FLOAT
    defaults: 5
    description: Alert when a mint's supply grows by more than this between two
      checks. An unexpected mint is how infinite-mint exploits show up.
  - id: remind_minutes
    type: INT
    defaults: 60
    description: Ask again after this long while a change is still not approved.
  - id: explorer_url
    type: STRING
    defaults: https://solscan.io
    description: Block explorer for links in Slack.
  - id: kestra_url
    type: STRING
    defaults: http://localhost:8080
    description: Base URL of your Kestra UI, used for the approval link.

variables:
  # KV keys hold one JSON document per watched account: approved state, observed state, last supply, when on-call was last asked.
  kv_prefix: solana_watch_
  # Pebble cannot concatenate lists; jq can.
  entries: "{{ {'mints': inputs.mints, 'programs': inputs.programs} |
    jq('[(.mints[] | {kind: \"mint\", entry: .}), (.programs[] | {kind:
    \"program\", entry: .})] | map(. + {label: (.entry | split(\"=\") | .[0] |
    gsub(\"^ +| +$\"; \"\")), address: (.entry | split(\"=\") | .[1] | gsub(\"
    \"; \"\"))})') | first | toJson }}"
  result: "{{ read(outputs.analyze.outputFiles['result.json']) }}"

tasks:
  - id: validate_inputs
    type: io.kestra.plugin.core.flow.If
    description: Entries must be `Label=address` with a base58 address; labels
      cannot contain commas.
    condition: "{{ {'mints': inputs.mints, 'programs': inputs.programs} |
      jq('[.mints[], .programs[] | select(test(\"^[^=,]+=
      *[1-9A-HJ-NP-Za-km-z]{32,44} *$\") | not)] | length') | first > 0 }}"
    then:
      - id: invalid_inputs
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: "These entries are not in the form Label=address: {{ {'mints':
          inputs.mints, 'programs': inputs.programs} | jq('[.mints[],
          .programs[] | select(test(\"^[^=,]+= *[1-9A-HJ-NP-Za-km-z]{32,44}
          *$\") | not)] | join(\", \")') | first }}"

  - id: read_accounts
    type: io.kestra.plugin.core.http.Request
    description: One call for every mint and program, decoded by the node
      (jsonParsed). Up to 100 accounts per call.
    uri: "{{ secret('SOLANA_RPC_URL') }}"
    method: POST
    contentType: application/json
    body: "{{ {'jsonrpc': '2.0', 'id': 1, 'method': 'getMultipleAccounts', 'params':
      [fromJson(render(vars.entries)) | jq('[.[].address]') | first,
      {'encoding': 'jsonParsed', 'commitment': 'confirmed'}]} | toJson }}"
    retry:
      type: exponential
      interval: PT2S
      maxInterval: PT30S
      maxAttempts: 3

  - id: accounts_read
    type: io.kestra.plugin.core.flow.If
    description: JSON-RPC errors arrive with HTTP 200.
    condition: "{{ (outputs.read_accounts.body | jq('.error.message // \"\"') |
      first) != '' }}"
    then:
      - id: rpc_refused
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: "getMultipleAccounts failed: {{ outputs.read_accounts.body |
          jq('.error.message') | first }}"

  - id: read_program_data
    type: io.kestra.plugin.core.http.Request
    description: |
      The upgrade authority and last deploy slot live in each program's ProgramData account.
      Only its 45-byte header is read (dataSlice), never the program binary.
    uri: "{{ secret('SOLANA_RPC_URL') }}"
    method: POST
    contentType: application/json
    body: "{{ {'jsonrpc': '2.0', 'id': 1, 'method': 'getMultipleAccounts', 'params':
      [outputs.read_accounts.body | jq('[.result.value[] | select(. != null) |
      .data.parsed? // {} | select(.type == \"program\") | .info.programData]')
      | first, {'encoding': 'base64', 'commitment': 'confirmed', 'dataSlice':
      {'offset': 0, 'length': 45}}]} | toJson }}"
    retry:
      type: exponential
      interval: PT2S
      maxInterval: PT30S
      maxAttempts: 3

  - id: analyze
    type: io.kestra.plugin.scripts.python.Script
    description: Build each account's state, compare it with the approved baseline
      from the KV store, and decide what to report, propose and save.
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    inputFiles:
      entries.json: "{{ render(vars.entries) }}"
      accounts.json: "{{ outputs.read_accounts.body }}"
      program_data.json: "{{ outputs.read_program_data.body }}"
      previous.json: >-
        { {% for e in fromJson(render(vars.entries)) %}"{{ e.address }}": {{
        kv(vars.kv_prefix ~ e.address, errorOnMissing=false) | toJson }}{% if
        not loop.last %}, {% endif %}{% endfor %} }
      settings.json: "{{ {'supply_jump_percent': inputs.supply_jump_percent,
        'remind_minutes': inputs.remind_minutes, 'kv_prefix': vars.kv_prefix} |
        toJson }}"
    outputFiles:
      - result.json
    script: |
      import base64, hashlib, json
      from datetime import datetime, timedelta, timezone

      entries = json.load(open("entries.json"))
      accounts = json.load(open("accounts.json"))["result"]["value"]
      program_data = {}
      pd_response = json.load(open("program_data.json"))
      if "error" in pd_response:
          raise SystemExit("reading ProgramData failed: " + pd_response["error"]["message"])
      previous = json.load(open("previous.json"))
      settings = json.load(open("settings.json"))
      now = datetime.now(timezone.utc)

      ALPHABET = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz"
      def base58(raw):
          n, out = int.from_bytes(raw, "big"), ""
          while n:
              n, r = divmod(n, 58)
              out = ALPHABET[r] + out
          return "1" * (len(raw) - len(raw.lstrip(b"\0"))) + out

      # ProgramData header: u32 type (3), u64 last deploy slot, option<pubkey> upgrade authority
      pd_addresses = [a["data"]["parsed"]["info"]["programData"] for a in accounts
                      if a and isinstance(a["data"], dict) and a["data"].get("parsed", {}).get("type") == "program"]
      for address, account in zip(pd_addresses, pd_response["result"]["value"]):
          raw = base64.b64decode(account["data"][0]) if account else b""
          if len(raw) == 45 and int.from_bytes(raw[:4], "little") == 3:
              program_data[address] = {
                  "last_deploy_slot": int.from_bytes(raw[4:12], "little"),
                  "upgrade_authority": base58(raw[13:45]) if raw[12] else "none (immutable)",
              }

      AUTHORITY_FIELDS = ("authority", "delegate", "programId", "closeAuthority")
      def mint_state(info, owner):
          state = {
              "program": "Token-2022" if owner.startswith("TokenzQd") else "SPL Token",
              "mint_authority": info.get("mintAuthority") or "none (fixed supply)",
              "freeze_authority": info.get("freezeAuthority") or "none",
          }
          for ext in info.get("extensions", []):
              name = ext["extension"]
              state["extension " + name] = "enabled"
              for key, value in (ext.get("state") or {}).items():
                  if key.endswith("Authority") or key in AUTHORITY_FIELDS:
                      state[f"{name}.{key}"] = value or "none"
          return state

      rows, updates, problems = [], [], []
      for entry, account in zip(entries, accounts):
          label, address, kind = entry["label"], entry["address"], entry["kind"]
          if account is None:
              problems.append(f"{label} ({address}): no account at this address on this cluster")
              continue
          parsed = account["data"].get("parsed", {}) if isinstance(account["data"], dict) else {}
          supply = decimals = None
          if kind == "mint":
              if parsed.get("type") != "mint":
                  problems.append(f"{label} ({address}): not a token mint (owner {account['owner']}); programs go under `programs`")
                  continue
              info = parsed["info"]
              state = mint_state(info, account["owner"])
              supply, decimals = int(info["supply"]), info["decimals"]
          else:
              if not account.get("executable"):
                  problems.append(f"{label} ({address}): not a program (owner {account['owner']}); token mints go under `mints`")
                  continue
              if parsed.get("type") == "program":
                  pd = program_data.get(parsed["info"]["programData"])
                  if pd is None:
                      problems.append(f"{label} ({address}): its ProgramData account could not be read")
                      continue
                  state = dict(pd)
              else:  # loaders without upgrades
                  state = {"loader": account["owner"], "upgrade_authority": "none (immutable)"}

          fingerprint = hashlib.sha256(json.dumps(state, sort_keys=True).encode()).hexdigest()[:10]
          prev = previous.get(address)
          first_seen = prev is None
          approved = state if first_seen else prev["approved"]
          changed_now = not first_seen and state != prev["observed"]
          unapproved = state != approved
          asked_at = None if first_seen else prev.get("asked_at")
          due = asked_at is None or now - datetime.fromisoformat(asked_at) >= timedelta(minutes=settings["remind_minutes"])
          propose = unapproved and (changed_now or due)
          if propose:
              asked_at = now.isoformat()
          if not unapproved:
              asked_at = None

          diff = [f"- {k}: {approved.get(k, 'not set')} → {state.get(k, 'not set')}"
                  for k in sorted(set(state) | set(approved)) if state.get(k) != approved.get(k)]
          changed_keys = {k for k in set(state) | set(approved) if state.get(k) != approved.get(k)}
          only_redeploy = unapproved and changed_keys == {"last_deploy_slot"}
          # an authority that was revoked (fixed supply, immutable program) removes a risk rather than adding one
          only_revoked = unapproved and all(str(state.get(k, "")).startswith("none") for k in changed_keys)

          supply_line = None
          if supply is not None and prev and prev.get("supply"):
              growth = (supply - prev["supply"]) * 100 / prev["supply"]
              if growth > settings["supply_jump_percent"]:
                  fmt = lambda v: f"{v / 10 ** decimals:,.{min(decimals, 2)}f}"
                  supply_line = f"supply grew {growth:.1f}% since the last check ({fmt(prev['supply'])} → {fmt(supply)})"

          rows.append({
              "label": label, "address": address, "kind": kind, "first_seen": first_seen,
              "newly_changed": changed_now and unapproved, "restored": changed_now and not unapproved,
              "unapproved": unapproved, "propose": propose, "fingerprint": fingerprint,
              "severity": None if not unapproved else "UPGRADE" if only_redeploy else "NOTICE" if only_revoked else "CRITICAL",
              "diff": "\n".join(diff),
              "summary": ", ".join(f"{k} {v}" for k, v in sorted(state.items())),
              "supply_jump": supply_line,
          })
          updates.append({"key": settings["kv_prefix"] + address, "value": {
              "label": label, "kind": kind, "approved": approved, "observed": state,
              "fingerprint": fingerprint, "asked_at": asked_at,
              "supply": supply, "decimals": decimals, "checked_at": now.isoformat(),
          }})

      json.dump({"rows": rows, "updates": updates, "problems": problems}, open("result.json", "w"))
      print(f"{len(rows)} accounts checked, {sum(r['unapproved'] for r in rows)} unapproved, {len(problems)} problems")

  - id: inputs_ok
    type: io.kestra.plugin.core.flow.If
    description: A typo, the wrong cluster or a mint listed as a program fails the
      run instead of becoming an empty baseline.
    condition: "{{ fromJson(render(vars.result)).problems | length > 0 }}"
    then:
      - id: bad_accounts
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: "{{ fromJson(render(vars.result)).problems | join('; ') }}"

  - id: save_state
    type: io.kestra.plugin.core.flow.Loop
    description: Save each account's state before notifying, so the next run
      compares against it.
    values: "{{ fromJson(render(vars.result)).updates | toJson }}"
    tasks:
      - id: save
        type: io.kestra.plugin.core.kv.Set
        key: "{{ fromJson(item.value).key }}"
        value: "{{ fromJson(item.value).value | toJson }}"
        kvType: JSON

  - id: report
    type: io.kestra.plugin.core.flow.If
    condition: "{{ fromJson(render(vars.result)).rows | jq('[.[] |
      select(.first_seen or .newly_changed or .restored or .supply_jump !=
      null)] | length') | first > 0 }}"
    then:
      - id: notify_changes
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          {% for x in fromJson(render(vars.result)).rows %}{% if x.newly_changed %}{{ x.severity }}: {{ x.label }} ({{ x.kind }} {{ x.address }}) changed:
          {{ x.diff }}
          {{ inputs.explorer_url }}/account/{{ x.address }}
          {% elseif x.restored %}{{ x.label }} ({{ x.address }}) is back to its approved state.
          {% elseif x.first_seen %}Now watching {{ x.label }} ({{ x.kind }} {{ x.address }}): {{ x.summary }}
          {% endif %}{% if x.supply_jump != null %}SUPPLY: {{ x.label }} {{ x.supply_jump }}. {{ inputs.explorer_url }}/token/{{ x.address }}
          {% endif %}{% endfor %}

  - id: propose
    type: io.kestra.plugin.core.flow.If
    condition: "{{ fromJson(render(vars.result)).rows | jq('[.[] | select(.propose)]
      | length') | first > 0 }}"
    then:
      - id: ask_on_call
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          Unapproved Solana account changes:
          {% for x in fromJson(render(vars.result)).rows %}{% if x.propose %}- {{ x.severity }} {{ x.label }} ({{ x.address }})
          {% endif %}{% endfor %}Decide within 4 hours: {{ inputs.kestra_url }}/ui/main/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}
          - Approve as new baseline: you checked the change (announcement, governance vote, your own deploy); the note is kept with the execution
          - Not yet: keep it unapproved; you will be asked again in {{ inputs.remind_minutes }} min

      - id: wait_for_decision
        type: io.kestra.plugin.core.flow.Pause
        description: No answer within 4 hours leaves the change unapproved; it is
          proposed again later.
        pauseDuration: PT4H
        behavior: RESUME
        onResume:
          - id: decision
            type: SELECT
            values:
              - Approve as new baseline
              - Not yet
            defaults: Not yet
          - id: only_accounts
            type: STRING
            required: false
            description: Optional comma-separated labels or addresses to approve. Empty
              means every proposed account.
          - id: note
            type: STRING
            required: false
            description: Why the change is expected, for example a link to the announcement.

      - id: approve
        type: io.kestra.plugin.core.flow.If
        condition: "{{ outputs.wait_for_decision.onResume.decision == 'Approve as new
          baseline' }}"
        then:
          - id: approve_each
            type: io.kestra.plugin.core.flow.Loop
            description: Approve exactly the state that was shown. The KV store is read
              again, so an account that changed after the proposal is not
              approved.
            values: >-
              {{ {'rows': fromJson(render(vars.result)).rows, 'only':
              outputs.wait_for_decision.onResume.only_accounts ?? ''} |
              jq('(.only | split(",") | map(gsub("^ +| +$"; "")) | map(select(.
              != ""))) as $o
                    | [.rows[] | select(.propose)
                       | select(($o | length) == 0 or (. as $r | any($o[]; ascii_downcase == ($r.label | ascii_downcase) or . == $r.address)))]')
              | first | toJson }}
            outputs:
              - id: result
                type: STRING
                value: >-
                  {{ fromJson(item.value).label }}: {% if outputs.approve_state
                  is defined %}approved as the new baseline{% else %}NOT
                  approved: it changed again after the proposal, or was already
                  handled{% endif %}
            tasks:
              - id: unchanged
                type: io.kestra.plugin.core.flow.If
                condition: "{{ (kv(vars.kv_prefix ~ fromJson(item.value).address,
                  errorOnMissing=false) | jq('.fingerprint // \"\"') | first) ==
                  fromJson(item.value).fingerprint }}"
                then:
                  - id: approve_state
                    type: io.kestra.plugin.core.kv.Set
                    key: "{{ vars.kv_prefix ~ fromJson(item.value).address }}"
                    value: "{{ kv(vars.kv_prefix ~ fromJson(item.value).address) | jq('.approved =
                      .observed | .asked_at = null') | first | toJson }}"
                    kvType: JSON

          - id: notify_approved
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            messageText: |
              Solana account changes reviewed{% if (outputs.wait_for_decision.onResume.note ?? '') != '' %} ("{{ outputs.wait_for_decision.onResume.note }}"){% endif %}:
              {% for r in outputs.approve_each.outputs ?? [] %}- {{ r.outputs.result }}
              {% else %}- nothing matched only_accounts ({{ outputs.wait_for_decision.onResume.only_accounts }}); nothing approved
              {% endfor %}

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: >-
      Solana authority watchdog failed in execution {{ execution.id }} {% if
      errorLogs() | length > 0 %}at `{{ errorLogs()[0]['taskId'] }}`: {{
      errorLogs()[0]['message'] | split(' \[\[') | first }}{% else %}without an
      error message{% endif %}. {{ inputs.kestra_url }}/ui/main/executions/{{
      flow.namespace }}/{{ flow.id }}/{{ execution.id }}

outputs:
  - id: accounts
    type: JSON
    description: Every watched mint and program with its state, whether it is
      approved, what changed, and any supply jump.
    value: "{{ outputs.analyze is defined ? fromJson(render(vars.result)).rows : []
      }}"

On Solana, whoever holds a token's mint authority can create unlimited supply. Whoever holds its freeze authority, or a Token-2022 permanent delegate, can freeze or move anyone's balance. And whoever holds a program's upgrade authority can replace the code behind it. Wallets, exchanges and DeFi apps depend on these keys staying where they are, yet a key handover or a redeploy usually goes unnoticed until something breaks. This blueprint keeps an approved baseline for every mint and program you rely on and turns each change into a reviewed decision.

How it works

account_changed (Webhook) runs a check when a Helius or QuickNode webhook sees activity on a watched account. periodic_check (Schedule) runs every 10 minutes and measures supply growth between checks.
read_accounts fetches every mint and program in one getMultipleAccounts call, decoded by the node (jsonParsed, confirmed commitment).
read_program_data reads only the 45-byte header of each program's ProgramData account (dataSlice), which holds the last deploy slot and the upgrade authority, and never the program binary.
analyze (Python, standard library only) builds each account's state:
- mints: token program, mint authority, freeze authority, and every Token-2022 extension with its authorities (permanent delegate, transfer hook, transfer fee, metadata, close authority…);
- programs: upgrade authority and last deploy slot.
It compares the state with the approved baseline that the KV store holds for that account, then classifies each change:
- CRITICAL when control moved;
- UPGRADE for a redeploy by the same authority;
- NOTICE when an authority was revoked (fixed supply, immutable program).
It also flags supply growth above supply_jump_percent.
save_state (Loop + kv.Set) stores each account's state before anyone is notified.
report posts new changes, restorations and supply jumps; propose asks on-call through a Pause (4 hours) to Approve as new baseline or Not yet, optionally for some accounts only, with a note.
approve_each (Loop) re-reads the KV store and approves an account only if its state is still the one that was shown.

Error handling

Approvals never cover more than was shown. Each proposal carries a fingerprint of the state. If the account changed again while the approval waited, it is "NOT approved: it changed again after the proposal" and the new change is reported on its own. An only_accounts answer that matches nothing approves nothing and says so.
No alert storms. A change is reported once, proposed again only after remind_minutes, and reported again when the account returns to its approved state. Supply jumps are alerts, not proposals.
No silent empty baselines:
- an address with no account ("no account at this address on this cluster", usually a typo or the wrong cluster) fails the run;
- a mint listed under programs, or a program listed under mints, fails with a hint;
- malformed entries fail before any RPC call.
JSON-RPC errors arrive with HTTP 200. accounts_read fails with the node's message.
Clear failures. RPC calls retry with backoff, and the errors alert names the task and cause and links the execution.

What this blueprint teaches about Kestra

The KV store as a small state database: kv() reads the baselines into a script, and kv.Set in a Loop writes them back. No external database is needed.
jq inside Pebble to combine list inputs and build JSON-RPC requests.
Reading only what you need from a large on-chain account (dataSlice) and decoding it in a dependency-free Python script on the Process task runner.
A Pause whose approval is bound to a fingerprint, so a late answer cannot approve something newer.

Prerequisites

A Solana RPC endpoint: any provider, or the public endpoint for light use. Use the cluster where your tokens live.
A Slack incoming webhook.
Python 3 on the Kestra worker (included in the Kestra image) for the Process task runner.

Secrets

SOLANA_RPC_URL: Solana JSON-RPC endpoint.
SLACK_WEBHOOK_URL: Slack incoming webhook.

Inputs

mints (ARRAY): Label=mintAddress entries, SPL Token or Token-2022. The defaults watch USDT and PYUSD on mainnet.
programs (ARRAY): Label=programId entries. The defaults watch Raydium AMM v4 and Orca Whirlpool.
supply_jump_percent (FLOAT, default 5): supply growth between checks that triggers an alert.
remind_minutes (INT, default 60): how often an unapproved change is proposed again.
explorer_url, kestra_url (STRING): links in Slack.

Outputs

accounts (JSON): each mint and program with its state, whether it is approved, what changed, and any supply jump.

Quick start

Add the secrets and run the flow with the default inputs against mainnet. Slack lists each account with its authorities; PYUSD shows its Token-2022 extensions, including the permanent delegate.
Replace mints and programs with what your product holds, routes through or integrates with. Up to 100 accounts per flow.
Point a Helius or QuickNode webhook for these accounts at account_changed, and change its key.
When an issuer or protocol announces a change, approve it with a link to the announcement as the note.
Adjust supply_jump_percent to the normal issuance rate of each token you watch; split fast-growing tokens into their own copy of the flow if needed.