Find and revoke over-broad Cassandra permission grants

id: cassandra-overly-broad-permission-grant-guard
namespace: company.team

description: |
  Read Cassandra's own grant table, system_auth.role_permissions. Flag every
  non-superuser role that holds a permission far wider than a service needs:
  anything on ALL KEYSPACES, a write or structural permission on a whole keyspace,
  or anything on ALL ROLES. Superusers are skipped, because their rights come from
  the superuser bit and not from a grant. An allowlist of role plus resource pairs
  is skipped too, so a grant that is deliberate stays untouched. Reporting is the
  default. REVOKE runs only when auto_remediate is true, dry_run is false and a
  person approves the exact statement list, one statement at a time. The table is
  then read again and the run fails if an over-broad grant survived.

inputs:
  - id: cassandra_host
    type: STRING
    defaults: "cassandra"
    displayName: Cassandra host
    description: Hostname or IP of a Cassandra contact point reachable from the
      Kestra Worker.

  - id: cassandra_port
    type: INT
    defaults: 9042
    displayName: Cassandra CQL port
    description: Native CQL port the contact point listens on. 9042 by default.

  - id: cassandra_datacenter
    type: STRING
    defaults: "datacenter1"
    displayName: Local datacenter
    description: Datacenter the driver's load balancing policy treats as local. The
      official Cassandra image uses datacenter1.

  - id: cassandra_username
    type: STRING
    defaults: "cassandra"
    displayName: Cassandra role used by Kestra
    description: >-
      Role the flow logs in as. Reading needs SELECT on system_auth. Revoking
      also needs AUTHORIZE on the resource being revoked, which only a superuser
      or a role holding AUTHORIZE WITH GRANT OPTION has. A read-only role still
      produces a correct report.

  - id: cassandra_password
    type: STRING
    defaults: "{{ secret('CASSANDRA_PASSWORD') }}"
    displayName: Cassandra password
    description: Password for that role. Comes from a Kestra secret by default,
      never from the flow body.

  - id: flagged_keyspace_permissions
    type: MULTISELECT
    displayName: Keyspace permissions that count as over-broad
    description: >-
      Applies to a grant on a whole keyspace. A keyspace-wide SELECT is normal
      for a reporting role, so it is not flagged unless you add it here. A grant
      on ALL KEYSPACES or ALL ROLES is flagged whatever its permissions, because
      the scope is the problem.
    values:
      - AUTHORIZE
      - CREATE
      - ALTER
      - DROP
      - MODIFY
      - SELECT
      - DESCRIBE
    defaults:
      - AUTHORIZE
      - CREATE
      - ALTER
      - DROP
      - MODIFY

  - id: grant_allowlist
    type: STRING
    defaults: ""
    displayName: Allowlisted grants
    description: >-
      Comma separated role=resource pairs that are deliberate and must never be
      revoked, written with Cassandra's own resource names, for example
      backup_svc=data,reporting=data/analytics. Spaces are ignored. An
      allowlisted pair is dropped before remediation is planned, so it cannot
      reach a REVOKE.

  - id: auto_remediate
    type: BOOL
    defaults: false
    displayName: Auto remediate
    description: Leave false to report only. True still needs dry_run false and a
      human approval before any REVOKE runs.

  - id: dry_run
    type: BOOL
    defaults: true
    displayName: Dry run
    description: True names every statement that would run and changes nothing. Must
      be false, with auto_remediate true, before the approval step is reached.

variables:
  audit_jq: >-
    [ .roles[] | select(.is_superuser == true) | .role ] as $su | (.allowlist |
    gsub(" ";"") | split(",") | map(select(length > 0))) as $allow | [
    .flagged[] | ascii_upcase ] as $flag | [ .grants[] | . as $row | ($row.role
    // "") as $role | ($row.resource // "") as $res | [ ($row.permissions //
    [])[] | ascii_upcase ] as $perms | ($res | split("/")) as $parts | (if $res
    == "data" then "ALL_KEYSPACES" elif ($parts | length) == 2 and $parts[0] ==
    "data" then "KEYSPACE" elif $res == "roles" then "ALL_ROLES" else "NARROW"
    end) as $breadth | (if $breadth == "ALL_KEYSPACES" then "ALL KEYSPACES" elif
    $breadth == "KEYSPACE" then "KEYSPACE " + $parts[1] elif $breadth ==
    "ALL_ROLES" then "ALL ROLES" else "" end) as $scope | (if $breadth ==
    "ALL_KEYSPACES" then "ALL KEYSPACES" elif $breadth == "KEYSPACE" then
    "KEYSPACE \"" + ($parts[1] | gsub("\"";"\"\"")) + "\"" elif $breadth ==
    "ALL_ROLES" then "ALL ROLES" else "" end) as $scope_cql | (if $breadth ==
    "KEYSPACE" then [ $perms[] | . as $p | select(($flag | index($p)) != null) ]
    else $perms end) as $offending | select($breadth != "NARROW") |
    select(($offending | length) > 0) | select(($su | index($role)) == null) |
    select(($allow | index($role + "=" + $res)) == null) | { role: $role,
    resource: $res, breadth: $breadth, scope: $scope, scope_cql: $scope_cql,
    permissions: ($offending | sort) } ] | sort_by(.role + "|" + .resource)
  plan_cql_jq: >-
    [ .[] | . as $f | $f.permissions[] | "REVOKE " + . + " ON " + $f.scope_cql +
    " FROM \"" + ($f.role | gsub("\"";"\"\"")) + "\"" ]
  plan_keys_jq: >-
    [ .[] | . as $f | $f.permissions[] | $f.role + "|" + $f.resource + "|" + . ]
  names_jq: >-
    [ .[] | .role + " holds " + (.permissions | join(",")) + " on " + .scope ] |
    join("; ")
  survivor_jq: >-
    [ .grants[] | . as $row | (($row.permissions // [])[] | (($row.role // "") +
    "|" + ($row.resource // "") + "|" + ascii_upcase)) ] as $have | [ .planned[]
    | . as $p | select(($have | index($p)) != null) ]

tasks:
  - id: read_roles
    type: io.kestra.plugin.cassandra.standard.Query
    description: >-
      Read only. Lists every role with its superuser bit from system_auth.roles.
      This is the exclusion list: a superuser already has every right, so a
      grant row against a superuser is noise and revoking it changes nothing a
      caller can do.
    session:
      endpoints:
        - hostname: "{{ inputs.cassandra_host }}"
          port: "{{ inputs.cassandra_port }}"
      localDatacenter: "{{ inputs.cassandra_datacenter }}"
      username: "{{ inputs.cassandra_username }}"
      password: "{{ inputs.cassandra_password }}"
    cql: "SELECT role, is_superuser FROM system_auth.roles"
    fetchType: FETCH
    allowFailure: true

  - id: read_grants
    type: io.kestra.plugin.cassandra.standard.Query
    description: >-
      Read only. Lists every grant Cassandra actually enforces from
      system_auth.role_permissions, the table behind LIST ALL PERMISSIONS. The
      resource column carries the scope: data means every keyspace, data/ks
      means one whole keyspace, data/ks/table means one table, roles means every
      role.
    session:
      endpoints:
        - hostname: "{{ inputs.cassandra_host }}"
          port: "{{ inputs.cassandra_port }}"
      localDatacenter: "{{ inputs.cassandra_datacenter }}"
      username: "{{ inputs.cassandra_username }}"
      password: "{{ inputs.cassandra_password }}"
    cql: "SELECT role, resource, permissions FROM system_auth.role_permissions"
    fetchType: FETCH
    allowFailure: true

  - id: audit_input
    type: io.kestra.plugin.core.output.OutputValues
    description: Put the grant rows, the superuser bits, the allowlist and the
      flagged permission set into one JSON document so a single jq program
      decides what counts as over-broad.
    values:
      payload: "{{ {'grants': (outputs.read_grants.rows is defined ?
        outputs.read_grants.rows : []), 'roles': (outputs.read_roles.rows is
        defined ? outputs.read_roles.rows : []), 'allowlist':
        inputs.grant_allowlist, 'flagged': inputs.flagged_keyspace_permissions}
        | toJson }}"

  - id: findings
    type: io.kestra.plugin.core.output.OutputValues
    description: Keep the over-broad grants, count them and render a readable line.
      Superuser roles and allowlisted role=resource pairs are already gone at
      this point.
    values:
      json: "{{ outputs.audit_input.values.payload | jq(render(vars.audit_jq)) | first
        | toJson }}"
      count: "{{ outputs.audit_input.values.payload | jq(render(vars.audit_jq) ~ ' |
        length') | first }}"
      names: "{{ outputs.audit_input.values.payload | jq(render(vars.audit_jq) ~ ' | '
        ~ render(vars.names_jq)) | first }}"

  - id: summary
    type: io.kestra.plugin.core.log.Log
    description: One line of audit history per run, so a clean cluster is recorded
      rather than silent.
    message: >-
      Grant guard on {{ inputs.cassandra_host }}: {{ outputs.read_grants.size ??
      0 }} grant rows read, {{ outputs.read_roles.size ?? 0 }} roles known, {{
      outputs.findings.values.count }} over-broad for a non-superuser.
      Allowlist: {{ inputs.grant_allowlist == '' ? 'empty' :
      inputs.grant_allowlist }}. Flagged keyspace permissions: {{
      inputs.flagged_keyspace_permissions | join(',') }}.

  - id: classify
    type: io.kestra.plugin.core.debug.Return
    description: UNREACHABLE when system_auth.roles produced nothing, OVER_BROAD
      when at least one grant survived every exclusion, otherwise CLEAN.
    format: "{{ (outputs.read_roles.rows is defined and (outputs.read_roles.rows |
      length) > 0) ? (((outputs.findings.values.count | number) > 0) ?
      'OVER_BROAD' : 'CLEAN') : 'UNREACHABLE' }}"

  - id: route
    type: io.kestra.plugin.core.flow.Switch
    description: Log and stop when clean. Fail loudly when the cluster could not be
      read. Go to the remediation gate when a grant is too wide.
    value: "{{ outputs.classify.value }}"
    defaults:
      - id: unexpected_state
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: "Grant guard reached an unexpected state {{ outputs.classify.value
          }} on {{ inputs.cassandra_host }}."
    cases:
      CLEAN:
        - id: clean
          type: io.kestra.plugin.core.log.Log
          message: >-
            No over-broad grant on {{ inputs.cassandra_host }}. Every
            non-superuser grant sits on a single table or is a keyspace read or
            is allowlisted. {{ outputs.read_grants.size ?? 0 }} rows checked.

      UNREACHABLE:
        - id: fail_unreachable
          type: io.kestra.plugin.core.execution.Fail
          errorMessage: >-
            Could not read system_auth.roles on {{ inputs.cassandra_host }}:{{
            inputs.cassandra_port }}. A guard that cannot read the grant table
            is not a clean result. Check the contact point, the local datacenter
            name plus whether {{ inputs.cassandra_username }} has SELECT on
            system_auth.

      OVER_BROAD:
        - id: alert
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {{ {'text': 'Cassandra ' ~ inputs.cassandra_host ~ ': ' ~ outputs.findings.values.count ~ ' over-broad grant(s) on non-superuser roles. ' ~ outputs.findings.values.names ~ '. ' ~ ((inputs.auto_remediate and not inputs.dry_run) ? 'Waiting for approval to revoke.' : 'Nothing revoked (auto_remediate=' ~ inputs.auto_remediate ~ ' dry_run=' ~ inputs.dry_run ~ ').')} | toJson }}

        - id: plan
          type: io.kestra.plugin.core.output.OutputValues
          description: >-
            Build one REVOKE per role, resource and permission, because CQL
            takes a single permission per REVOKE. The keys list is the same set
            as role|resource|permission. That list is what the re-read is
            checked against.
          values:
            cql: "{{ outputs.findings.values.json | jq(render(vars.plan_cql_jq)) | first |
              toJson }}"
            keys: "{{ outputs.findings.values.json | jq(render(vars.plan_keys_jq)) | first |
              toJson }}"
            statements: "{{ outputs.findings.values.json | jq(render(vars.plan_cql_jq) ~ ' |
              join(\"; \")') | first }}"

        - id: show_plan
          type: io.kestra.plugin.core.log.Log
          description: Print the exact statement list. On a dry run this is the whole
            output, so the plan can be read before anyone turns the gates off.
          message: >-
            Planned REVOKE statements ({{ outputs.plan.values.cql | jq('length')
            | first }}): {{ outputs.plan.values.statements }}. {{
            (inputs.auto_remediate and not inputs.dry_run) ? 'Gates open, asking
            for approval.' : 'Gates closed, nothing will run.' }}

        - id: remediate_gate
          type: io.kestra.plugin.core.flow.If
          description: Both gates must be set before a human is even asked. Report only is
            the default.
          condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
          then:
            - id: hold
              type: io.kestra.plugin.core.flow.Pause
              description: >-
                Human approval before any REVOKE. A wide grant is sometimes the
                only thing keeping a service running, so a person reads the
                statement list first. An unanswered pause leaves approved null,
                which the gate below reads as not approved.
              onResume:
                - id: approved
                  type: BOOL
                  required: false
                  description: Set true to run the statement list above. Anything else leaves
                    every grant in place.

            - id: approval_gate
              type: io.kestra.plugin.core.flow.If
              condition: "{{ outputs.hold.onResume.approved ?? false }}"
              then:
                - id: remediate
                  type: io.kestra.plugin.core.flow.Loop
                  description: >-
                    Sequential on purpose. Cassandra writes auth changes to
                    system_auth and keeps a permissions cache per node, so
                    concurrent REVOKE on the same role races and the losing
                    statement is lost quietly.
                  concurrencyLimit: 1
                  values: "{{ outputs.plan.values.cql }}"
                  tasks:
                    - id: revoke
                      type: io.kestra.plugin.cassandra.standard.Query
                      description: >-
                        One REVOKE per iteration. allowFailure is set because a
                        role without AUTHORIZE is refused here. The re-read
                        below is what decides whether anything landed.
                      allowFailure: true
                      session:
                        endpoints:
                          - hostname: "{{ inputs.cassandra_host }}"
                            port: "{{ inputs.cassandra_port }}"
                        localDatacenter: "{{ inputs.cassandra_datacenter }}"
                        username: "{{ inputs.cassandra_username }}"
                        password: "{{ inputs.cassandra_password }}"
                      cql: "{{ item.value }}"

                - id: settle
                  type: io.kestra.plugin.core.flow.Sleep
                  description: Short wait so every node has applied the auth mutation before the
                    table is read again.
                  duration: PT3S

                - id: reverify_read
                  type: io.kestra.plugin.cassandra.standard.Query
                  description: Read system_auth.role_permissions again. A REVOKE refused by
                    Cassandra still leaves the task green, so this re-read is
                    the only thing that proves the change landed.
                  session:
                    endpoints:
                      - hostname: "{{ inputs.cassandra_host }}"
                        port: "{{ inputs.cassandra_port }}"
                    localDatacenter: "{{ inputs.cassandra_datacenter }}"
                    username: "{{ inputs.cassandra_username }}"
                    password: "{{ inputs.cassandra_password }}"
                  cql: "SELECT role, resource, permissions FROM system_auth.role_permissions"
                  fetchType: FETCH
                  allowFailure: true

                - id: reverify_input
                  type: io.kestra.plugin.core.output.OutputValues
                  description: Same audit document shape as before, plus the planned keys, so the
                    second pass runs the same rules as the first.
                  values:
                    audit: "{{ {'grants': (outputs.reverify_read.rows is defined ?
                      outputs.reverify_read.rows : []), 'roles':
                      (outputs.read_roles.rows is defined ?
                      outputs.read_roles.rows : []), 'allowlist':
                      inputs.grant_allowlist, 'flagged':
                      inputs.flagged_keyspace_permissions} | toJson }}"
                    survivors: "{{ {'grants': (outputs.reverify_read.rows is defined ?
                      outputs.reverify_read.rows : []), 'planned':
                      fromJson(outputs.plan.values.keys)} | toJson }}"

                - id: reverify
                  type: io.kestra.plugin.core.output.OutputValues
                  description: Count what is still over-broad, then separately name the planned
                    statements whose grant is still present.
                  values:
                    remaining: "{{ outputs.reverify_input.values.audit | jq(render(vars.audit_jq) ~
                      ' | length') | first }}"
                    remaining_names: "{{ outputs.reverify_input.values.audit |
                      jq(render(vars.audit_jq) ~ ' | ' ~ render(vars.names_jq))
                      | first }}"
                    survived: "{{ outputs.reverify_input.values.survivors |
                      jq(render(vars.survivor_jq) ~ ' | join(\"; \")') | first
                      }}"

                - id: verify_result
                  type: io.kestra.plugin.core.flow.If
                  description: The run fails when the grant table still holds an over-broad grant,
                    whatever the REVOKE tasks reported.
                  condition: "{{ (outputs.reverify.values.remaining | number) > 0 }}"
                  then:
                    - id: fail_not_remediated
                      type: io.kestra.plugin.core.execution.Fail
                      errorMessage: >-
                        REVOKE did not land on {{ inputs.cassandra_host }}.
                        system_auth.role_permissions still holds {{
                        outputs.reverify.values.remaining }} over-broad
                        grant(s): {{ outputs.reverify.values.remaining_names }}.
                        Still present from the planned list: {{
                        outputs.reverify.values.survived }}. Likely causes: {{
                        inputs.cassandra_username }} does not hold AUTHORIZE on
                        the resource, so Cassandra refused the statement while
                        the task stayed green; or a new grant was made while the
                        run waited for approval.
                  else:
                    - id: remediated
                      type: io.kestra.plugin.core.log.Log
                      message: >-
                        Revoked {{ outputs.plan.values.cql | jq('length') |
                        first }} grant(s) on {{ inputs.cassandra_host }}.
                        system_auth.role_permissions now reports 0 over-broad
                        grants for a non-superuser.

              else:
                - id: declined
                  type: io.kestra.plugin.core.log.Log
                  message: >-
                    Approval not given, so nothing was revoked on {{
                    inputs.cassandra_host }}. {{ outputs.findings.values.count
                    }} over-broad grant(s) are still in place: {{
                    outputs.findings.values.names }}.

          else:
            - id: report_only
              type: io.kestra.plugin.core.log.Log
              message: >-
                Report only on {{ inputs.cassandra_host }} (auto_remediate={{
                inputs.auto_remediate }}, dry_run={{ inputs.dry_run }}). {{
                outputs.findings.values.count }} over-broad grant(s) found and
                nothing was changed. Statements that would run: {{
                outputs.plan.values.statements }}.

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Separate channel for the guard itself breaking, which is not the
      same event as finding a wide grant.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Cassandra grant guard failed on ' ~ inputs.cassandra_host ~ ' (execution ' ~ execution.id ~ '). Either the grant table could not be read or a REVOKE was planned and the re-read showed it did not land. Grants are unchanged unless the log says otherwise.'} | toJson }}

triggers:
  - id: every_six_hours
    type: io.kestra.plugin.core.trigger.Schedule
    description: >-
      Audit the grant table every six hours. Shipped disabled. Enable it once
      cassandra_host and cassandra_datacenter point at a real cluster and a
      manual run has been read. The trigger always forces report only, whatever
      this flow's own defaults are.
    cron: "0 */6 * * *"
    disabled: true
    inputs:
      cassandra_host: "cassandra"
      cassandra_port: 9042
      cassandra_datacenter: "datacenter1"
      cassandra_username: "cassandra"
      grant_allowlist: ""
      auto_remediate: false
      dry_run: true