Kestra Flow Security, Governance and Linting Gate

id: kestra-flow-security-and-governance-gate
namespace: company.governance

description: |
  Automated GitOps governance and security validation gate that audits Kestra flow definitions
  before production deployment. Scans YAML files for plaintext credentials, deprecated syntax,
  missing error handlers, unbounded retries, and schema violations, enforcing organizational policies
  and dispatching actionable remediation reports to Slack.

inputs:
  - id: seed_demo_data
    type: BOOL
    defaults: true
    description: When true, simulates candidate flows containing compliant and
      non-compliant samples for zero-setup local execution.

  - id: fail_on_policy_breach
    type: BOOL
    defaults: true
    description: When true, halts execution and fails the gate if high-severity
      violations are detected.

  - id: target_namespace
    type: STRING
    defaults: company.production
    description: Production namespace targeted for deployment verification.

tasks:
  - id: load_candidate_flows
    type: io.kestra.plugin.scripts.python.Script
    description: Prepares candidate flow definitions from repository or loads
      synthetic compliance test cases.
    containerImage: python:3.11-slim
    script: |
      import json
      from kestra import Kestra

      demo_mode = {{ inputs.seed_demo_data }}

      candidate_flows = []

      if demo_mode:
          compliant_flow = {
              "filename": "flows/analytics_orders_etl.yaml",
              "content": """id: analytics-orders-etl
      namespace: company.production
      description: Production compliant orders ETL pipeline.
      tasks:
        - id: extract_orders
          type: io.kestra.plugin.jdbc.postgresql.Query
          url: jdbc:postgresql://postgres:5432/orders
          username: db_user
          password: "{{ secret('POSTGRES_PASSWORD') }}"
          sql: SELECT * FROM orders;
        - id: transform_orders
          type: io.kestra.plugin.scripts.shell.Commands
          containerImage: python:3.11-slim
          commands:
            - python3 -c "print('Transforming records...')"
      errors:
        - id: notify_on_failure
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: '{"text": "ETL Failed."}'"""
          }

          vulnerable_flow = {
              "filename": "flows/unverified_sync.yaml",
              "content": """id: unverified-sync
      namespace: company.production
      description: Legacy sync flow with governance violations.
      tasks:
        - id: loop_items
          type: io.kestra.plugin.core.flow.Loop
          values: ["a", "b"]
          tasks:
            - id: run_script
              type: io.kestra.plugin.scripts.shell.Commands
              runner: Docker
              commands:
                - echo "export API_TOKEN='sk_live_12345abcdef67890' for {{ item.value }}"
      pluginDefaults:
        - type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          values:
            url: "{{ secret('SLACK_WEBHOOK_URL') }}" """
          }

          candidate_flows = [compliant_flow, vulnerable_flow]

      Kestra.outputs({
          "flows_count": len(candidate_flows),
          "flows_json": json.dumps(candidate_flows)
      })

  - id: scan_flow_security_and_syntax
    type: io.kestra.plugin.scripts.python.Script
    description: Executes policy rules across candidate flows checking for secret
      leaks, deprecated primitives, and missing error handlers.
    containerImage: python:3.11-slim
    beforeCommands:
      - pip install --quiet pyyaml
    script: |
      import json
      import re
      import yaml
      from kestra import Kestra

      raw_flows = json.loads("""{{ outputs.load_candidate_flows.vars.flows_json }}""")

      total_scanned = len(raw_flows)
      passed_count = 0
      violations = []

      secret_pattern = re.compile(r'(?:(?:api[_-]?key|password|token|secret)\s*[:=]\s*["\'](?!\{\{\s*secret\([^\)]+\)\s*\}\})(?!\{\{\s*inputs\.)[a-zA-Z0-9_\-\.]{8,}["\']|sk_live_[a-zA-Z0-9]{10,}|ghp_[a-zA-Z0-9]{10,})', re.IGNORECASE)

      for item in raw_flows:
          fname = item.get("filename", "unknown.yaml")
          content = item.get("content", "")
          file_violations = []

          # Rule 1: Plaintext Secret Detection
          if secret_pattern.search(content):
              file_violations.append({
                  "severity": "CRITICAL",
                  "rule": "PLAINTEXT_SECRET_DETECTED",
                  "message": "Hardcoded secret detected. Credentials must use {{ secret('NAME') }}."
              })

          # Rule 2: Top-level pluginDefaults Prohibition
          if "pluginDefaults:" in content:
              file_violations.append({
                  "severity": "HIGH",
                  "rule": "PLUGIN_DEFAULTS_DEPRECATED",
                  "message": "Found top-level 'pluginDefaults'. In Kestra 2.0, inline credentials directly on tasks."
              })

          # Rule 3: Deprecated 1.x Properties
          if re.search(r'\s+runner:\s+Docker', content):
              file_violations.append({
                  "severity": "HIGH",
                  "rule": "DEPRECATED_RUNNER_PROPERTY",
                  "message": "Deprecated 'runner: Docker' detected. Replace with 'taskRunner:'."
              })

          # Rule 4: Structural Schema and Error Handling
          try:
              data = yaml.safe_load(content)
              if not isinstance(data, dict):
                  file_violations.append({
                      "severity": "CRITICAL",
                      "rule": "INVALID_YAML_STRUCTURE",
                      "message": "Flow root is not a valid mapping."
                  })
              else:
                  if "errors" not in data:
                      file_violations.append({
                          "severity": "MEDIUM",
                          "rule": "MISSING_ERROR_HANDLER",
                          "message": "Flow lacks a top-level 'errors' block for production alert dispatch."
                      })
          except Exception as err:
              file_violations.append({
                  "severity": "CRITICAL",
                  "rule": "YAML_SYNTAX_ERROR",
                  "message": f"Syntax error: {str(err)}"
              })

          if file_violations:
              violations.append({
                  "filename": fname,
                  "findings": file_violations
              })
          else:
              passed_count += 1

      has_critical = any(
          any(f["severity"] in ("CRITICAL", "HIGH") for f in v["findings"])
          for v in violations
      )

      Kestra.outputs({
          "total_scanned": total_scanned,
          "passed_count": passed_count,
          "violations_count": len(violations),
          "has_critical_breach": has_critical,
          "violations_report": json.dumps(violations)
      })

  - id: evaluate_governance_gate
    type: io.kestra.plugin.core.flow.If
    description: Evaluates whether critical security or architectural policies were
      breached.
    condition: "{{ outputs.scan_flow_security_and_syntax.vars.has_critical_breach
      and inputs.fail_on_policy_breach }}"
    then:
      - id: alert_governance_blocker
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches high-priority governance violation card to Slack.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "🛡️ *Kestra Flow Governance Gate Blocked*\n• *Target Namespace:* `{{ inputs.target_namespace }}`\n• *Total Flows Audited:* {{ outputs.scan_flow_security_and_syntax.vars.total_scanned }}\n• *Compliant Flows:* {{ outputs.scan_flow_security_and_syntax.vars.passed_count }}\n• *Violations Found:* {{ outputs.scan_flow_security_and_syntax.vars.violations_count }}\n• *Status:* Deployment halted due to policy violations (plaintext secrets or deprecated syntax)."
          }

      - id: log_governance_remediation
        type: io.kestra.plugin.core.log.Log
        description: Emits detailed itemized violations list in workflow execution logs.
        message: "Policy violations detected during GitOps audit: {{
          outputs.scan_flow_security_and_syntax.vars.violations_report }}"
    else:
      - id: notify_governance_approval
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches green compliance verification notice to platform
          engineering.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "✅ *Kestra Flow Governance Gate Passed*\n• *Target Namespace:* `{{ inputs.target_namespace }}`\n• *All candidate flows meet security, error-handling, and Kestra 2.0 schema standards.*"
          }

triggers:
  - id: scheduled_security_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Scheduled weekly compliance sweep across all repository flows.
      Shipped disabled by default.
    cron: "0 4 * * 1"
    disabled: true

  - id: gitops_pr_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Authenticated webhook for event-driven PR validation in CI/CD pipelines.
    key: "{{ secret('GITOPS_WEBHOOK_KEY') }}"

errors:
  - id: alert_audit_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Catches execution errors and alerts platform on-call.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "⚠️ Governance Gate execution {{ execution.id }} failed unexpectedly on flow {{ flow.id }}."
      }