Just-in-Time PostgreSQL Access with Approval, Automatic Revocation and a Drift Sweep

id: just-in-time-database-access
namespace: company.team
description: |
  Replace standing database privileges with approved, time-boxed access: an
  engineer requests a PostgreSQL role with a reason, a lead approves, the role is
  granted, and Kestra revokes it when the time is up. A sweep revokes anything
  that outlived its window and reports grants made outside the process.

labels:
  team: security
  use-case: just-in-time-access

variables:
  # Managed roles and the longest access each may be granted for, in minutes.
  max_minutes:
    app_readonly: 480
    app_readwrite: 240
    app_admin: 60

triggers:
  - id: access_request
    type: io.kestra.plugin.core.trigger.Webhook
    description: Call from a Slack command, an internal portal or a ticketing tool
      with requester, role, minutes, reason and ticket in a JSON body.
    key: "{{ secret('JIT_WEBHOOK_KEY') }}"
    inputs:
      mode: request
      requester: "{{ trigger.body.requester ?? '' }}"
      role: "{{ trigger.body.role ?? 'app_readonly' }}"
      minutes: "{{ trigger.body.minutes ?? 60 }}"
      reason: "{{ trigger.body.reason ?? '' }}"
      ticket: "{{ trigger.body.ticket ?? '' }}"

  - id: sweep
    type: io.kestra.plugin.core.trigger.Schedule
    description: Every 15 minutes, revoke grants whose window has passed and report
      members of the managed roles that were not granted through this flow.
      Shipped disabled until the database secrets are set.
    cron: "*/15 * * * *"
    disabled: true
    inputs:
      mode: sweep

inputs:
  - id: mode
    type: SELECT
    displayName: Mode
    values:
      - request
      - sweep
    defaults: request
  - id: requester
    type: STRING
    displayName: Your database login (required for requests)
    description: Your existing personal PostgreSQL login. The role is granted to it,
      so no password is created or sent.
    required: false
  - id: role
    type: SELECT
    displayName: Role
    values:
      - app_readonly
      - app_readwrite
      - app_admin
    defaults: app_readonly
  - id: minutes
    type: INT
    displayName: Minutes of access
    description: Capped per role (readonly 480, readwrite 240, admin 60).
    defaults: 60
  - id: reason
    type: STRING
    displayName: Reason (required for requests)
    description: Why you need the access, at least 10 characters. It is shown to the
      approver and kept in the audit trail.
    required: false
  - id: ticket
    type: STRING
    displayName: Ticket or incident
    required: false

tasks:
  - id: by_mode
    type: io.kestra.plugin.core.flow.Switch
    value: "{{ inputs.mode }}"
    cases:
      request:
        - id: check_login
          type: io.kestra.plugin.core.execution.Fail
          description: The login is used as an identifier in GRANT, so it must be a plain
            lowercase PostgreSQL name.
          condition: "{{ not ((inputs.requester ?? '') |
            regexMatch(regex='^[a-z_][a-z0-9_]{0,62}$')) }}"
          errorMessage: "'{{ inputs.requester ?? '' }}' is not a valid login. Enter your
            personal PostgreSQL login: lowercase letters, digits and
            underscores."

        - id: check_reason
          type: io.kestra.plugin.core.execution.Fail
          condition: "{{ (inputs.reason ?? '') | trim | length < 10 }}"
          errorMessage: "A reason of at least 10 characters is required, so the approver
            and the audit trail know why the access was needed."

        - id: check_minutes
          type: io.kestra.plugin.core.execution.Fail
          condition: "{{ inputs.minutes < 1 or inputs.minutes >
            vars.max_minutes[inputs.role] }}"
          errorMessage: "{{ inputs.role }} can be granted for 1 to {{
            vars.max_minutes[inputs.role] }} minutes; {{ inputs.minutes }} was
            requested."

        - id: user_exists
          type: io.kestra.plugin.jdbc.postgresql.Query
          description: Confirm the requester's login exists, so access is only ever
            granted to a real personal login.
          url: "{{ secret('JIT_DB_URL') }}"
          username: "{{ secret('JIT_DB_USERNAME') }}"
          password: "{{ secret('JIT_DB_PASSWORD') }}"
          sql: SELECT rolname FROM pg_roles WHERE rolname = :login AND rolcanlogin
          parameters:
            login: "{{ inputs.requester }}"
          fetchType: FETCH_ONE
          retry:
            type: exponential
            interval: PT10S
            maxInterval: PT1M
            maxAttempts: 3

        - id: check_user
          type: io.kestra.plugin.core.execution.Fail
          condition: "{{ outputs.user_exists.row is null }}"
          errorMessage: "No PostgreSQL login named {{ inputs.requester }} exists."

        - id: already_active
          type: io.kestra.plugin.core.flow.If
          description: One active grant per login and role. A second request while one is
            active ends quietly.
          condition: "{{ (kv('jit-grant-' ~ inputs.requester ~ '-' ~ inputs.role,
            errorOnMissing=false).expires_epoch ?? 0) > (now() | timestamp) }}"
          then:
            - id: tell_already_active
              type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
              url: "{{ secret('SLACK_WEBHOOK_URL') }}"
              messageText: "ℹ️ {{ inputs.requester }} already has {{ inputs.role }} until {{
                kv('jit-grant-' ~ inputs.requester ~ '-' ~
                inputs.role).expires_at }}. No new request was opened."
              retry:
                type: exponential
                interval: PT10S
                maxInterval: PT1M
                maxAttempts: 3
            - id: stop_duplicate
              type: io.kestra.plugin.core.execution.Exit
              state: SUCCESS

        - id: ask_approver
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          messageText: |
            🔐 *Access request*: {{ inputs.requester }} asks for *{{ inputs.role }}* for {{ inputs.minutes }} minutes{{ (inputs.ticket ?? '') == '' ? '' : ' (' ~ inputs.ticket ~ ')' }}.
            Reason: {{ inputs.reason }}
            Approve or deny within 2 hours: {{ kestra.url ?? '' }}/ui/main/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}
          retry:
            type: exponential
            interval: PT10S
            maxInterval: PT1M
            maxAttempts: 3

        - id: approval
          type: io.kestra.plugin.core.flow.Pause
          description: The approver decides and may shorten the window. An unanswered
            request is cancelled after two hours.
          pauseDuration: PT2H
          behavior: CANCEL
          onResume:
            - id: decision
              type: SELECT
              displayName: Decision
              values:
                - Approve
                - Deny
              defaults: Approve
            - id: approved_minutes
              type: INT
              displayName: Minutes to grant (empty keeps the requested time)
              required: false
            - id: note
              type: STRING
              displayName: Note
              required: false

        - id: on_decision
          type: io.kestra.plugin.core.flow.Switch
          value: "{{ outputs.approval.onResume.decision }}"
          cases:
            Deny:
              - id: tell_denied
                type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
                url: "{{ secret('SLACK_WEBHOOK_URL') }}"
                messageText: "⛔ {{ inputs.role }} for {{ inputs.requester }} was denied by {{
                  outputs.approval.resumed.by ?? 'the approver' }}. {{
                  outputs.approval.onResume.note ?? '' }}"
                retry:
                  type: exponential
                  interval: PT10S
                  maxInterval: PT1M
                  maxAttempts: 3
            Approve:
              - id: window
                type: io.kestra.plugin.core.output.OutputValues
                description: The granted window, never longer than requested or than the role
                  allows.
                values:
                  minutes: "{{ [outputs.approval.onResume.approved_minutes ?? inputs.minutes,
                    inputs.minutes, vars.max_minutes[inputs.role]] | sort |
                    first }}"
                  expires_at: "{{ now() | dateAdd([outputs.approval.onResume.approved_minutes ??
                    inputs.minutes, inputs.minutes,
                    vars.max_minutes[inputs.role]] | sort | first, 'MINUTES') |
                    date(\"yyyy-MM-dd HH:mm 'UTC'\", timeZone='UTC') }}"
                  expires_epoch: "{{ now() | dateAdd([outputs.approval.onResume.approved_minutes
                    ?? inputs.minutes, inputs.minutes,
                    vars.max_minutes[inputs.role]] | sort | first, 'MINUTES') |
                    timestamp }}"

              - id: grant
                type: io.kestra.plugin.jdbc.postgresql.Queries
                description: Grant the managed role to the requester's own login. Identifiers
                  cannot be bound as parameters; the login was validated against
                  PostgreSQL's naming rules and pg_roles, and the role comes
                  from a fixed list.
                url: "{{ secret('JIT_DB_URL') }}"
                username: "{{ secret('JIT_DB_USERNAME') }}"
                password: "{{ secret('JIT_DB_PASSWORD') }}"
                transaction: true
                sql: GRANT "{{ inputs.role }}" TO "{{ inputs.requester }}";
                retry:
                  type: exponential
                  interval: PT10S
                  maxInterval: PT1M
                  maxAttempts: 3

              - id: verify_grant
                type: io.kestra.plugin.jdbc.postgresql.Query
                description: Read the membership back from the catalog instead of trusting the
                  GRANT.
                url: "{{ secret('JIT_DB_URL') }}"
                username: "{{ secret('JIT_DB_USERNAME') }}"
                password: "{{ secret('JIT_DB_PASSWORD') }}"
                sql: |
                  SELECT count(*) AS memberships FROM pg_auth_members m
                  JOIN pg_roles r ON r.oid = m.roleid JOIN pg_roles u ON u.oid = m.member
                  WHERE r.rolname = :role AND u.rolname = :login AND m.grantor = (SELECT oid FROM pg_roles WHERE rolname = current_user)
                parameters:
                  role: "{{ inputs.role }}"
                  login: "{{ inputs.requester }}"
                fetchType: FETCH_ONE
                retry:
                  type: exponential
                  interval: PT10S
                  maxInterval: PT1M
                  maxAttempts: 3

              - id: check_granted
                type: io.kestra.plugin.core.execution.Fail
                condition: "{{ outputs.verify_grant.row.memberships < 1 }}"
                errorMessage: "The GRANT ran but {{ inputs.requester }} is not a member of {{
                  inputs.role }}."

              - id: register
                type: io.kestra.plugin.core.kv.Set
                description: Record the active grant. The sweep uses it to revoke grants that
                  outlive their window.
                key: "jit-grant-{{ inputs.requester }}-{{ inputs.role }}"
                kvType: JSON
                value: |
                  {
                    "login": {{ inputs.requester | toJson }},
                    "role": {{ inputs.role | toJson }},
                    "reason": {{ inputs.reason | toJson }},
                    "ticket": {{ (inputs.ticket ?? '') | toJson }},
                    "approved_by": {{ (outputs.approval.resumed.by ?? 'not recorded') | toJson }},
                    "approved_at": {{ (outputs.approval.resumed.on ?? '') | toJson }},
                    "approver_note": {{ (outputs.approval.onResume.note ?? '') | toJson }},
                    "granted_at": "{{ now() | date("yyyy-MM-dd HH:mm 'UTC'", timeZone='UTC') }}",
                    "expires_at": {{ outputs.window.values.expires_at | toJson }},
                    "expires_epoch": {{ outputs.window.values.expires_epoch }},
                    "execution": "{{ execution.id }}"
                  }

              - id: tell_granted
                type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
                url: "{{ secret('SLACK_WEBHOOK_URL') }}"
                messageText: "✅ {{ inputs.requester }} has *{{ inputs.role }}* until {{
                  outputs.window.values.expires_at }} ({{
                  outputs.window.values.minutes }} minutes, approved by {{
                  outputs.approval.resumed.by ?? 'the approver' }}). To end it
                  early, resume: {{ kestra.url ?? '' }}/ui/main/executions/{{
                  flow.namespace }}/{{ flow.id }}/{{ execution.id }}"
                retry:
                  type: exponential
                  interval: PT10S
                  maxInterval: PT1M
                  maxAttempts: 3

              - id: access_window
                type: io.kestra.plugin.core.flow.Pause
                description: Wait for the granted window. Resume this task to end the access
                  early.
                pauseDuration: "PT{{ outputs.window.values.minutes }}M"
                behavior: RESUME

              - id: revoke
                type: io.kestra.plugin.jdbc.postgresql.Queries
                url: "{{ secret('JIT_DB_URL') }}"
                username: "{{ secret('JIT_DB_USERNAME') }}"
                password: "{{ secret('JIT_DB_PASSWORD') }}"
                transaction: true
                sql: REVOKE "{{ inputs.role }}" FROM "{{ inputs.requester }}";
                retry:
                  type: exponential
                  interval: PT10S
                  maxInterval: PT1M
                  maxAttempts: 5

              - id: verify_revoke
                type: io.kestra.plugin.jdbc.postgresql.Query
                url: "{{ secret('JIT_DB_URL') }}"
                username: "{{ secret('JIT_DB_USERNAME') }}"
                password: "{{ secret('JIT_DB_PASSWORD') }}"
                sql: |
                  SELECT count(*) AS memberships FROM pg_auth_members m
                  JOIN pg_roles r ON r.oid = m.roleid JOIN pg_roles u ON u.oid = m.member
                  WHERE r.rolname = :role AND u.rolname = :login AND m.grantor = (SELECT oid FROM pg_roles WHERE rolname = current_user)
                parameters:
                  role: "{{ inputs.role }}"
                  login: "{{ inputs.requester }}"
                fetchType: FETCH_ONE
                retry:
                  type: exponential
                  interval: PT10S
                  maxInterval: PT1M
                  maxAttempts: 3

              - id: check_revoked
                type: io.kestra.plugin.core.execution.Fail
                condition: "{{ outputs.verify_revoke.row.memberships > 0 }}"
                errorMessage: "REVOKE ran but {{ inputs.requester }} is still a member of {{
                  inputs.role }}."

              - id: unregister
                type: io.kestra.plugin.core.kv.Delete
                key: "jit-grant-{{ inputs.requester }}-{{ inputs.role }}"

              - id: tell_revoked
                type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
                url: "{{ secret('SLACK_WEBHOOK_URL') }}"
                messageText: "🔒 {{ inputs.role }} was revoked from {{ inputs.requester }}."
                retry:
                  type: exponential
                  interval: PT10S
                  maxInterval: PT1M
                  maxAttempts: 3

      sweep:
        - id: active_grants
          type: io.kestra.plugin.core.kv.GetKeys
          description: Every grant this flow still considers active.
          prefix: jit-grant-

        - id: members
          type: io.kestra.plugin.jdbc.postgresql.Query
          description: Everyone who is currently a member of a managed role, and who
            granted it.
          url: "{{ secret('JIT_DB_URL') }}"
          username: "{{ secret('JIT_DB_USERNAME') }}"
          password: "{{ secret('JIT_DB_PASSWORD') }}"
          sql: |
            SELECT r.rolname AS role, u.rolname AS login, g.rolname AS grantor
            FROM pg_auth_members m
            JOIN pg_roles r ON r.oid = m.roleid
            JOIN pg_roles u ON u.oid = m.member
            LEFT JOIN pg_roles g ON g.oid = m.grantor
            WHERE r.rolname IN ('app_readonly', 'app_readwrite', 'app_admin') AND u.rolname <> current_user
          fetchType: FETCH
          retry:
            type: exponential
            interval: PT10S
            maxInterval: PT1M
            maxAttempts: 3

        - id: reconcile
          type: io.kestra.plugin.scripts.python.Script
          description: Find registered grants whose window has passed, and members of the
            managed roles that have no registered grant.
          dependencies:
            - kestra
          inputFiles:
            members.json: "{{ outputs.members.rows | toJson }}"
            grants.json: "{ {% for key in outputs.active_grants.keys %}{{ key | toJson }}:
              {{ kv(key, errorOnMissing=false) | toJson | base64encode | toJson
              }}{% if not loop.last %}, {% endif %}{% endfor %} }"
          script: |
            import base64, json, time
            from kestra import Kestra

            grants = {k: json.loads(base64.b64decode(v)) for k, v in json.load(open("grants.json")).items()}
            members = json.load(open("members.json"))
            now = time.time()
            expired = [{"key": k, "login": g["login"], "role": g["role"], "expires_at": g["expires_at"]}
                       for k, g in grants.items() if g and g.get("expires_epoch", 0) <= now]
            registered = {(g["login"], g["role"]) for g in grants.values() if g}
            drift = [f"{m['login']} in {m['role']} (granted by {m['grantor'] or 'unknown'})"
                     for m in members if (m["login"], m["role"]) not in registered]
            Kestra.outputs({"expired": expired, "drift": drift, "active": len(grants) - len(expired)})

        - id: cleanup
          type: io.kestra.plugin.core.flow.Loop
          description: Revoke every grant that outlived its window, for example because
            the request execution was killed or Kestra was down when the timer
            ended.
          values: "{{ outputs.reconcile.vars.expired }}"
          concurrencyLimit: 2
          tasks:
            - id: sweep_revoke
              type: io.kestra.plugin.jdbc.postgresql.Queries
              url: "{{ secret('JIT_DB_URL') }}"
              username: "{{ secret('JIT_DB_USERNAME') }}"
              password: "{{ secret('JIT_DB_PASSWORD') }}"
              transaction: true
              sql: REVOKE "{{ fromJson(item.value).role }}" FROM "{{
                fromJson(item.value).login }}";
              retry:
                type: exponential
                interval: PT10S
                maxInterval: PT1M
                maxAttempts: 5
            - id: sweep_unregister
              type: io.kestra.plugin.core.kv.Delete
              key: "{{ fromJson(item.value).key }}"

        - id: sweep_report
          type: io.kestra.plugin.core.flow.If
          condition: "{{ outputs.reconcile.vars.expired | length > 0 or
            outputs.reconcile.vars.drift | length > 0 }}"
          then:
            - id: tell_sweep
              type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
              url: "{{ secret('SLACK_WEBHOOK_URL') }}"
              messageText: |
                🧹 *Just-in-time access sweep*: {{ outputs.reconcile.vars.expired | length }} expired grant(s) revoked{{ outputs.reconcile.vars.expired | length > 0 ? ': ' ~ (outputs.reconcile.vars.expired | jq('[.[] | .login + " (" + .role + ")"] | join(", ")') | first) : '' }}.
                {{ outputs.reconcile.vars.drift | length > 0 ? '⚠️ Members of managed roles not granted through this flow, review them: ' ~ (outputs.reconcile.vars.drift | join('; ')) : 'No access outside this flow.' }}
              retry:
                type: exponential
                interval: PT10S
                maxInterval: PT1M
                maxAttempts: 3

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert with the failed task and a link. If the grant already
      happened, the access stays active until the sweep revokes it at expiry.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    retry:
      type: constant
      interval: PT10S
      maxAttempts: 3
    messageText: |
      ❌ *Just-in-time access FAILED* ({{ inputs.mode }}{{ inputs.mode == 'request' ? ': ' ~ (inputs.requester ?? '?') ~ ' / ' ~ inputs.role : '' }}), execution {{ execution.id }}.
      Failed task: {{ (errorLogs() | length) > 0 ? errorLogs()[0].taskId : 'unknown' }}. Error: {{ (errorLogs() | length) > 0 ? errorLogs()[0].message : 'see the execution logs' }}
      {{ kestra.url ?? '' }}/ui/main/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}
      If access was already granted, it stays registered and the sweep revokes it when its window ends.

outputs:
  - id: decision
    type: STRING
    displayName: Decision
    value: "{{ outputs.approval.onResume.decision ?? 'none' }}"
  - id: expires_at
    type: STRING
    displayName: Access expires at
    value: "{{ outputs.window.values.expires_at ?? 'none' }}"