MySQL Account Hygiene Guard

id: mysql-account-hygiene-guard
namespace: company.team

description: |
  Audit MySQL accounts in mysql.user for the findings security reviews ask about: anonymous
  accounts, accounts with an empty password, root reachable from remote hosts, global admin
  privileges (SUPER, FILE, SHUTDOWN, CREATE USER, GRANT OPTION) outside an allowlist, and
  accounts open to every host. Alert Slack, and optionally drop, lock or revoke with generated
  statements, never touching the account the flow is connected as, then audit again to verify.

inputs:
  - id: jdbc_url
    type: STRING
    displayName: JDBC URL
    defaults: "jdbc:mysql://localhost:3306/mysql"

  - id: privileged_accounts
    type: ARRAY
    itemType: STRING
    displayName: Accounts allowed admin privileges
    description: As user@host.
    defaults:
      - "root@localhost"

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    description: Drop anonymous accounts, lock empty-password and remote root
      accounts, revoke global admin privileges. Wildcard hosts are only
      reported.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Must be false, together with auto_remediate, for any statement to run.
    defaults: true

variables:
  audit_sql: |
    WITH acc AS (
      SELECT CONVERT(u.User USING utf8mb4) AS user_name, CONVERT(u.Host USING utf8mb4) AS host,
             CONVERT(CONCAT(QUOTE(u.User), '@', QUOTE(u.Host)) USING utf8mb4) AS account,
             CONVERT(CONCAT(u.User, '@', u.Host) USING utf8mb4) AS id, u.authentication_string = '' AS empty_password, u.account_locked = 'Y' AS locked,
             CONVERT(CONCAT_WS(', ',
               IF(u.Super_priv = 'Y', 'SUPER', NULL), IF(u.File_priv = 'Y', 'FILE', NULL), IF(u.Shutdown_priv = 'Y', 'SHUTDOWN', NULL),
               IF(u.Create_user_priv = 'Y', 'CREATE USER', NULL), IF(u.Grant_priv = 'Y', 'GRANT OPTION', NULL)) USING utf8mb4) AS admin_privs,
             CONVERT(CONCAT(u.User, '@', u.Host) USING utf8mb4) = CONVERT(CURRENT_USER() USING utf8mb4) AS is_self
        FROM mysql.user u
       WHERE u.User NOT IN ('mysql.infoschema', 'mysql.session', 'mysql.sys')
    )
    SELECT * FROM (
      SELECT 'ANONYMOUS_ACCOUNT' AS finding, 'HIGH' AS severity, id, is_self, 'Any user name can log in from this host' AS detail,
             CONCAT('DROP USER ', account) AS fix_sql FROM acc WHERE user_name = ''
      UNION ALL
      SELECT 'EMPTY_PASSWORD', 'HIGH', id, is_self, 'Account with no password', CONCAT('ALTER USER ', account, ' ACCOUNT LOCK')
        FROM acc WHERE empty_password AND NOT locked AND user_name <> ''
      UNION ALL
      SELECT 'REMOTE_ROOT', 'HIGH', id, is_self, CONCAT('root can log in from ', host), CONCAT('ALTER USER ', account, ' ACCOUNT LOCK')
        FROM acc WHERE user_name = 'root' AND host NOT IN ('localhost', '127.0.0.1', '::1') AND NOT locked
      UNION ALL
      SELECT 'GLOBAL_ADMIN_PRIVILEGES', 'HIGH', id, is_self, CONCAT('Global ', admin_privs),
             CONCAT('REVOKE IF EXISTS ', admin_privs, ' ON *.* FROM ', account)
        FROM acc WHERE admin_privs <> '' AND user_name <> 'root' AND id NOT IN ({{ inputs.privileged_accounts | jq('map("\u0027" + gsub("\u0027"; "\u0027\u0027") + "\u0027") | join(",")') | first }})
      UNION ALL
      SELECT 'WILDCARD_HOST', 'MEDIUM', id, is_self, 'Can connect from any host (%)', NULL
        FROM acc WHERE host = '%' AND user_name NOT IN ('', 'root')
    ) f
    ORDER BY severity, finding, id

tasks:
  - id: audit
    type: io.kestra.plugin.jdbc.mysql.Query
    description: Read-only query on mysql.user. Statements quote names with QUOTE(),
      and the connecting account is flagged so it is never changed.
    url: "{{ inputs.jdbc_url }}"
    username: "{{ secret('MYSQL_USERNAME') }}"
    password: "{{ secret('MYSQL_PASSWORD') }}"
    fetchType: FETCH
    sql: "{{ render(vars.audit_sql) }}"

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: >-
      MySQL account audit on {{ inputs.jdbc_url }}: {{ outputs.audit.size }}
      findings. {% for r in outputs.audit.rows %}{{ r.severity }} {{ r.finding
      }} {{ r.id }}{{ r.is_self ? ' (connected account, never changed)' : '' }}:
      {{ r.detail }}. {% endfor %}

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.audit.size > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': 'MySQL account audit on ' ~ inputs.jdbc_url ~ ': ' ~ (outputs.audit.rows | jq('map("\(.severity) \(.finding) \(.id)") | join(", ")') | first) ~ '.' ~ (inputs.auto_remediate and not inputs.dry_run ? ' Applying the fixes now.' : ' No change applied (auto_remediate=' ~ inputs.auto_remediate ~ ', dry_run=' ~ inputs.dry_run ~ ').')} | toJson }}

      - id: remediate_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run and
          outputs.audit.rows | jq('map(select(.fix_sql != null and (.is_self |
          tostring) != \"1\" and (.is_self | tostring) != \"true\")) | length')
          | first > 0 }}"
        then:
          - id: remediate
            type: io.kestra.plugin.core.flow.Loop
            concurrencyLimit: 1
            description: One statement per finding, each isolated. The connecting account is
              skipped.
            values: "{{ outputs.audit.rows | jq('map(select(.fix_sql != null and (.is_self |
              tostring) != \"1\" and (.is_self | tostring) != \"true\") |
              .fix_sql)') | first | toJson }}"
            tasks:
              - id: fix
                type: io.kestra.plugin.jdbc.mysql.Query
                allowFailure: true
                url: "{{ inputs.jdbc_url }}"
                username: "{{ secret('MYSQL_USERNAME') }}"
                password: "{{ secret('MYSQL_PASSWORD') }}"
                fetchType: NONE
                sql: "{{ item.value }}"

          - id: reaudit
            type: io.kestra.plugin.jdbc.mysql.Query
            url: "{{ inputs.jdbc_url }}"
            username: "{{ secret('MYSQL_USERNAME') }}"
            password: "{{ secret('MYSQL_PASSWORD') }}"
            fetchType: FETCH
            sql: "{{ render(vars.audit_sql) }}"

          - id: verify_result
            type: io.kestra.plugin.core.flow.If
            description: Every finding that was fixed must be gone from the second audit.
            condition: >-
              {{ {'b': outputs.audit.rows, 'a': outputs.reaudit.rows ?? []} |
              jq('([.a[] | .finding + "|" + .id]) as $a
                 | [.b[] | select(.fix_sql != null and (.is_self | tostring) != "1" and (.is_self | tostring) != "true") | .finding + "|" + .id | select(. as $x | $a | index($x))] | length') | first > 0 }}
            then:
              - id: remediation_failed
                type: io.kestra.plugin.core.execution.Fail
                errorMessage: "Some fixes did not take: {{ outputs.reaudit.rows |
                  jq('map(\"\\(.finding) \\(.id)\") | join(\", \")') | first }}.
                  The connecting account may lack CREATE USER or GRANT OPTION."
            else:
              - id: remediated
                type: io.kestra.plugin.core.log.Log
                message: >-
                  Fixed and verified: {{ outputs.audit.rows |
                  jq('map(select(.fix_sql != null and (.is_self | tostring) !=
                  "1" and (.is_self | tostring) != "true") | .fix_sql) | join(";
                  ")') | first }}. Left for review: {{ outputs.reaudit.rows |
                  jq('if length == 0 then "none" else map("\(.finding) \(.id)")
                  | join(", ") end') | first }}.
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: "No account finding."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'MySQL account audit failed on ' ~ inputs.jdbc_url ~ ' (execution ' ~ execution.id ~ '). Treat the accounts as unaudited.'} | toJson }}

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled, and always in safe mode.
    cron: "0 6 * * *"
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true