SFTP PGP Decrypt, Signature Verification and Replay Check

id: sftp-pgp-decrypt-and-verify
namespace: company.team
description: |
  Pick up PGP-encrypted files that a partner drops on an SFTP server, decrypt them and verify
  that they were signed by that partner. Accepted files stay in Kestra's internal storage and are
  exposed as flow outputs; unsigned, forged, modified (other than line endings) or replayed files are moved to quarantine
  and the execution ends in WARNING.

concurrency:
  limit: 1

variables:
  inbox_dir: /upload/inbox/
  received_dir: /upload/received/
  accepted_dir: /upload/accepted/
  quarantine_dir: /upload/quarantine/
  allowed_signer: partner@example.com
  replay_window: P90D

tasks:
  - id: each_file
    type: io.kestra.plugin.core.flow.Loop
    description: One iteration per file picked up by the trigger, one at a time, so
      a rejected file never blocks the valid ones.
    values: "{{ trigger.files ?? [] }}"
    concurrencyLimit: 1
    outputs:
      - id: file
        type: STRING
        value: "{{ fromJson(item.value).name }}"
      - id: status
        type: STRING
        value: "{{ (outputs.mark_accepted.values.status ?? '') == 'accepted' ?
          'accepted' : ((outputs.previous.values.seen ?? '') != '' ? 'duplicate'
          : ((outputs.decrypt_and_verify.uri ?? '') == '' ? 'rejected' :
          'error')) }}"
      - id: reason
        type: STRING
        value: "{{ (outputs.mark_accepted.values.status ?? '') == 'accepted' ?
          'signature and integrity verified' : ((outputs.previous.values.seen ??
          '') != '' ? 'same content already accepted: ' ~
          outputs.previous.values.seen : ((outputs.decrypt_and_verify.uri ?? '')
          == '' ? 'decrypt_and_verify failed: not signed by the partner key,
          modified, or not decryptable (a wrong PGP_PRIVATE_KEY or passphrase
          also ends here); the log of this iteration has the exact message' :
          'verified, but a later step failed; see the logs of this iteration'))
          }}"
      - id: sha256
        type: STRING
        value: "{{ outputs.fingerprint.values.sha256 ?? '' }}"
      - id: uri
        type: STRING
        value: "{{ (outputs.mark_accepted.values.status ?? '') == 'accepted' ?
          outputs.decrypt_and_verify.uri : '' }}"
      - id: encrypted_uri
        type: STRING
        value: "{{ fromJson(item.value).path }}"
      - id: archived_as
        type: STRING
        value: "{{ (outputs.mark_accepted.values.status ?? '') == 'accepted' ?
          vars.accepted_dir ~ execution.id ~ '_' ~ fromJson(item.value).name :
          '' }}"
      - id: quarantined_as
        type: STRING
        value: "{{ (outputs.quarantine.to ?? '') != '' ? vars.quarantine_dir ~
          execution.id ~ '_' ~ fromJson(item.value).name : '' }}"
    tasks:
      - id: intake
        type: io.kestra.plugin.core.flow.AllowFailure
        description: Decrypt, verify and de-duplicate one file. If any step fails, the
          errors branch quarantines the original and the loop moves on.
        tasks:
          - id: decrypt_and_verify
            type: io.kestra.plugin.crypto.openpgp.Decrypt
            description: Fails unless the message decrypts with your private key, passes its
              integrity check and carries a valid document signature from the
              partner key whose user ID is `allowed_signer`.
            from: "{{ fromJson(item.value).path }}"
            privateKey: "{{ secret('PGP_PRIVATE_KEY') }}"
            privateKeyPassphrase: "{{ secret('PGP_PRIVATE_KEY_PASSPHRASE') }}"
            signUsersKey:
              - "{{ secret('PARTNER_PGP_PUBLIC_KEY') }}"
            requiredSignerUsers:
              - "{{ vars.allowed_signer }}"

          - id: fingerprint
            type: io.kestra.plugin.core.output.OutputValues
            description: SHA-256 of the decrypted content read as UTF-8 text, and the replay
              key, computed with CRLF and CR line endings turned into LF because
              a signature over text can verify with any line-ending variant of
              it.
            values:
              sha256: "{{ read(outputs.decrypt_and_verify.uri) | sha256 }}"
              content_key: "{{ read(outputs.decrypt_and_verify.uri) | replace({'\r

                ': '

                '}) | replace({'\r': '

                '}) | sha256 }}"

          - id: previous
            type: io.kestra.plugin.core.output.OutputValues
            description: Whether the same content was already accepted within the replay
              window.
            values:
              seen: "{{ kv(flow.id ~ '-' ~ outputs.fingerprint.values.content_key,
                errorOnMissing=false) ?? '' }}"

          - id: reject_replay
            type: io.kestra.plugin.core.execution.Fail
            description: A validly signed file whose content was already accepted is a
              replay, even under another file name.
            condition: "{{ outputs.previous.values.seen != '' }}"
            errorMessage: "{{ fromJson(item.value).name }} has the same content as {{
              outputs.previous.values.seen }}"

          - id: archive_accepted
            type: io.kestra.plugin.fs.sftp.Move
            description: Moves the encrypted original to the accepted directory, prefixed
              with the execution ID, so a later file with the same name cannot
              replace it.
            host: "{{ secret('SFTP_HOST') }}"
            username: "{{ secret('SFTP_USERNAME') }}"
            password: "{{ secret('SFTP_PASSWORD') }}"
            from: "{{ vars.received_dir }}{{ fromJson(item.value).name }}"
            to: "{{ vars.accepted_dir }}{{ execution.id }}_{{ fromJson(item.value).name }}"
            overwrite: false

          - id: log_verified
            type: io.kestra.plugin.core.log.Log
            message: "Verified {{ fromJson(item.value).name }} (sha256 {{
              outputs.fingerprint.values.sha256 }}): signed by {{
              vars.allowed_signer }}, decrypted to {{
              outputs.decrypt_and_verify.uri }}"

          - id: remember_content
            type: io.kestra.plugin.core.kv.Set
            description: Records the content so the same signed file is refused if it is
              sent again within `replay_window`. It runs after the archive move,
              so a file that fails any earlier step is not remembered and can be
              re-sent.
            key: "{{ flow.id }}-{{ outputs.fingerprint.values.content_key }}"
            value: "{{ fromJson(item.value).name }}, accepted {{ now() }}"
            ttl: "{{ vars.replay_window }}"

          - id: mark_accepted
            type: io.kestra.plugin.core.output.OutputValues
            values:
              status: accepted
        errors:
          - id: quarantine
            type: io.kestra.plugin.fs.sftp.Move
            description: Moves the encrypted original to quarantine, prefixed with the
              execution ID so earlier evidence is never overwritten.
            host: "{{ secret('SFTP_HOST') }}"
            username: "{{ secret('SFTP_USERNAME') }}"
            password: "{{ secret('SFTP_PASSWORD') }}"
            from: "{{ (outputs.archive_accepted.to ?? '') != '' ? vars.accepted_dir ~
              execution.id ~ '_' : vars.received_dir }}{{
              fromJson(item.value).name }}"
            to: "{{ vars.quarantine_dir }}{{ execution.id }}_{{ fromJson(item.value).name
              }}"
            overwrite: false

          - id: log_quarantined
            type: io.kestra.plugin.core.log.Log
            level: WARN
            message: "{{ (outputs.quarantine.to ?? '') != '' ? 'Quarantined ' ~
              fromJson(item.value).name ~ ' as ' ~ vars.quarantine_dir ~
              execution.id ~ '_' ~ fromJson(item.value).name ~ '.' : 'Could not
              quarantine ' ~ fromJson(item.value).name ~ ': the original was not
              moved, see the quarantine task log.' }} The log of the task that
              failed for this file gives the exact reason."

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: "{{ (outputs.each_file.outputs ?? []) | jq('[.[] | {file:
      .outputs.file, iteration: .item.iteration, status: .outputs.status,
      reason: .outputs.reason}]') | first }}"

  - id: flag_not_accepted
    type: io.kestra.plugin.core.execution.Fail
    description: Ends the execution in WARNING when at least one file was not
      accepted, so your existing alerts pick it up. `allowFailure` keeps the
      flow outputs, which `Exit` would skip.
    allowFailure: true
    condition: "{{ ((outputs.each_file.outputs ?? []) | jq('[.[] |
      select(.outputs.status != \"accepted\")] | length') | first) > 0 }}"
    errorMessage: "Not accepted: {{ (outputs.each_file.outputs ?? []) | jq('[.[] |
      select(.outputs.status != \"accepted\") | {file: .outputs.file, iteration:
      .item.iteration, status: .outputs.status, reason: .outputs.reason}]') |
      first }}"

outputs:
  - id: accepted
    type: JSON
    description: Accepted files with the SHA-256 (of the content read as UTF-8 text)
      and internal storage URI of the decrypted content, the URI of the
      encrypted file as downloaded, and where the original was archived.
    value: "{{ (outputs.each_file.outputs ?? []) | jq('[.[] | select(.outputs.status
      == \"accepted\") | .outputs | {file, sha256, uri, encrypted_uri,
      archived_as}]') | first }}"
  - id: not_accepted
    type: JSON
    description: Rejected, replayed or failed files, with the loop iteration, the
      reason, the URI of the encrypted file as downloaded and where the original
      was quarantined.
    value: "{{ (outputs.each_file.outputs ?? []) | jq('[.[] | select(.outputs.status
      != \"accepted\") | {file: .outputs.file, iteration: .item.iteration,
      status: .outputs.status, reason: .outputs.reason, encrypted_uri:
      .outputs.encrypted_uri, quarantined_as: .outputs.quarantined_as}]') |
      first }}"

triggers:
  - id: watch_inbox
    type: io.kestra.plugin.fs.sftp.Trigger
    description: Polls the inbox every minute, oldest files first. Matching files,
      with ASCII names only, are downloaded and moved to the received directory
      before the execution starts.
    host: "{{ secret('SFTP_HOST') }}"
    username: "{{ secret('SFTP_USERNAME') }}"
    password: "{{ secret('SFTP_PASSWORD') }}"
    from: "{{ vars.inbox_dir }}"
    regExp: "(?i).*/[\\x20-\\x2E\\x30-\\x7E]+\\.(pgp|gpg)$"
    action: MOVE
    moveDirectory: "{{ vars.received_dir }}"
    sort: LAST_MODIFIED_ASC
    interval: PT1M

Banks, payment processors and payroll providers often deliver PGP-encrypted, signed files on an SFTP server. This blueprint accepts a file only if it decrypts with your key, passes its integrity check, carries a valid document signature from the partner key you configured, and its content was not accepted within replay_window. Every other file it picks up is moved to quarantine and the execution ends in WARNING.

How it works

watch_inbox (io.kestra.plugin.fs.sftp.Trigger) polls the inbox every minute for .pgp and .gpg files (any case, ASCII names), oldest first, downloads them and moves the originals to the received directory.
each_file (io.kestra.plugin.core.flow.Loop) handles one file at a time inside an AllowFailure block:
- decrypt_and_verify (io.kestra.plugin.crypto.openpgp.Decrypt) checks decryption, integrity and a document signature by PARTNER_PGP_PUBLIC_KEY with the user ID allowed_signer.
- reject_replay fails if the same content was accepted within replay_window (KV store), whatever the file is called.
- archive_accepted moves the original to the accepted directory and remember_content stores the content hash. If any step fails, the errors branch moves it to quarantine instead.
flag_not_accepted ends the execution in WARNING when at least one file was not accepted.

concurrency.limit: 1 keeps two executions from checking the same content at once. Decrypted content never goes back to the SFTP server.

Outputs

accepted: file, sha256 (of the decrypted content read as UTF-8 text, so it equals the file's SHA-256 only for UTF-8 or ASCII content), uri of the decrypted file in internal storage, encrypted_uri (the encrypted file as downloaded) and archived_as.
not_accepted: file, iteration (whose logs give the exact message), status, reason, encrypted_uri and quarantined_as. The status is rejected (any failure of decrypt_and_verify, including a wrong private key or passphrase), duplicate (content already accepted) or error (verified, but a later step such as the SFTP move or the KV store failed; the content is not remembered, so the file can be re-sent).
On the server, originals end up in the accepted or quarantine directory, prefixed with the execution ID. If the quarantine move fails too, the original stays where it was (received, or accepted if it had already been archived).

Prerequisites

Kestra 2.0 or later and plugin-fs 2.11 or later (tested with 2.12.0).
plugin-crypto 1.3.9 or later: earlier versions do not check the MDC or the final AEAD tag (before 1.3.7), decrypt only for the first recipient key (before 1.3.8) or accept signatures that are not document signatures (before 1.3.9).
An SFTP account for Kestra that reads the inbox, moves files out of it and writes the received, accepted and quarantine directories (create them beforehand), and a separate partner account that can only write to the inbox.
Your OpenPGP key pair, and the partner's public key with its fingerprint confirmed by another channel.
Secrets: SFTP_HOST (port 22), SFTP_USERNAME, SFTP_PASSWORD, PGP_PRIVATE_KEY and PARTNER_PGP_PUBLIC_KEY (both ASCII-armored) and PGP_PRIVATE_KEY_PASSPHRASE.
Variables: the four directories (absolute, with the trailing slash), allowed_signer (the full user ID or its email, matched exactly; the example value rejects every real file) and replay_window (a duration in days or hours, default P90D; P3M, P1Y or P2W are rejected, and every valid file would end as error).

Quick start

Create the secrets and directories, adjust variables and save the flow.
Have the partner run gpg --encrypt --sign -r data-intake@example.com -u partner@example.com orders.csv, upload orders.csv.gpg under a temporary name that does not end in .pgp or .gpg (for example .orders.csv.gpg.part) and rename it.
Within a minute or two outputs.accepted lists the file; checksum mismatch in the decrypt_and_verify log means a wrong passphrase.

The Execute button processes no files: there is no trigger context.

Limits

A wrong private key or passphrase looks like a bad file: every file is quarantined as rejected. Test with a known-good file after any key change.
The signature only proves that the partner key signed the content as a document, up to line endings: Decrypt takes text mode from the unsigned one-pass header, so text whose line breaks are all CRLF also verifies with LF or CR (the replay key treats them as the same content). Decrypt does not check expiry, revocation, key flags, the hash algorithm, the signature time or the user ID certification, so trust rests on PARTNER_PGP_PUBLIC_KEY: replace it when the partner rotates a key. Only the outermost one-pass signature is verified.
No recipient binding: anyone holding a file the partner signed, for you or for someone else, can re-encrypt it to your key and it is accepted unless the same content was accepted within replay_window.
Replay detection hashes the decrypted content in memory as UTF-8 text, with line endings normalized. It suits UTF-8 or ASCII text files of moderate size: binary files can collide or end as error, text in other encodings (such as Windows-1252) can collide, and identical legitimate files are marked duplicate, so ask for a date or sequence number in the content.
SFTP: plugin-fs does not verify the host key, so a man in the middle could capture SFTP_PASSWORD or withhold files: use a trusted network, and prefer key-based authentication (keyfile) over a password. The trigger does not wait for uploads to finish (hence the rename) and moves files before the execution starts, replacing a same-name file left in the received directory: check that directory after an outage or while executions are queued, and ask for unique file names. Non-ASCII file names are skipped and stay in the inbox (tested with letters, digits, ., _ and -), because plugin-fs 2.12.0 garbles them when it moves them. Server copies can be changed by anyone with write access; encrypted_uri is the copy Kestra verified.
Internal storage keeps the decrypted files until purged (for example with a scheduled io.kestra.plugin.core.execution.PurgeExecutions), and anyone who can read this flow's executions can download them. The replay lookup also reads KV entries of parent namespaces, so any flow that can write KV here or in a parent namespace can create or change the entries this flow relies on.
Formats: detached signatures, .asc names, hidden recipients (gpg --throw-keyids) and messages without a one-pass signature are not handled; up to 25 files per poll. Tested on Kestra 2.0.5 with GnuPG 2.4 messages (AEAD and v1 with MDC, compressed or not, binary and text-mode signatures) and Ed25519/Cv25519 and RSA 3072 keys.

How to extend

Load accepted files from a second flow with an io.kestra.plugin.core.trigger.Flow trigger on this flow (states SUCCESS and WARNING) that reads each uri in trigger.outputs.accepted.
Add a notification to the errors branch, or copy the flow per partner with its own key, directories, allowed_signer and flow ID.

Decrypt and Verify PGP Files from a Partner's SFTP

How it works

Outputs

Prerequisites

Quick start

Limits

How to extend

Links

Tasks

Share this Blueprint

Create a Blueprint

New to Kestra?

Decrypt and Verify PGP Files from a Partner's SFTP

How it works

Outputs

Prerequisites

Quick start

Limits

How to extend

Links

Tasks

Share this Blueprint

Create a Blueprint

More Data Blueprints

OpenSearch ISM Policy Compliance Gate

OpenSearch Indexing Backpressure and Threadpool Rejection Sentinel

Automated OpenLineage Metadata Extraction and Marquez Governance Pipeline

Hourly OpenSearch Error Spike Gate with Discord Alerting

Ingest an HTTP JSON Dataset into OpenSearch with Load

Full OpenSearch Index Export to Internal Storage with Scroll

More Infrastructure Blueprints

Onboard Users to VMs with Ansible and the Kestra Assets Catalog

Okta System Log Suspicious IP Session Hijack Sentinel

Ingest PLC Telemetry Only When Every Reading Is Trustworthy

1Password SCIM Bridge Sync Status Sentinel

Run a task on an on-demand Databricks cluster

Promote OPA Policies from Git with a Test and Approval Gate

New to Kestra?