Turn Aikido Compliance Reports into a Dated Evidence Pack That Shows Control Drift, Not Just a Score

id: aikido-compliance-drift-evidence-pack
namespace: company.security

inputs:
  - id: reporting_period
    type: STRING
    defaults: 2026-Q3
    description: Label stored with the evidence pack so an auditor can line packs up
      with a control period.

  - id: drift_alert_threshold
    type: INT
    defaults: 1
    description: How many controls may regress before the flow raises an alert
      rather than filing quietly.

tasks:
  - id: collect_frameworks
    type: io.kestra.plugin.core.flow.Parallel
    description: Pull all three framework overviews at once. Each is a separate
      Aikido endpoint and they do not depend on each other.
    tasks:
      - id: soc2
        type: io.kestra.plugin.aikido.compliance.GetReport
        clientId: "{{ secret('AIKIDO_CLIENT_ID') }}"
        clientSecret: "{{ secret('AIKIDO_CLIENT_SECRET') }}"
        framework: SOC2

      - id: iso27001
        type: io.kestra.plugin.aikido.compliance.GetReport
        clientId: "{{ secret('AIKIDO_CLIENT_ID') }}"
        clientSecret: "{{ secret('AIKIDO_CLIENT_SECRET') }}"
        framework: ISO27001

      - id: nis2
        type: io.kestra.plugin.aikido.compliance.GetReport
        clientId: "{{ secret('AIKIDO_CLIENT_ID') }}"
        clientSecret: "{{ secret('AIKIDO_CLIENT_SECRET') }}"
        framework: NIS2

  - id: open_findings
    type: io.kestra.plugin.aikido.issues.ListOpen
    description: The open critical and high findings are the evidence behind a
      failing vulnerability management control.
    clientId: "{{ secret('AIKIDO_CLIENT_ID') }}"
    clientSecret: "{{ secret('AIKIDO_CLIENT_SECRET') }}"
    severities:
      - CRITICAL
      - HIGH
    fetchType: FETCH

  - id: read_previous
    type: io.kestra.plugin.core.kv.Get
    description: Load the previous posture snapshot so the flow reports movement
      rather than a static score.
    key: compliance_posture
    errorOnMissing: false

  - id: detect_drift
    type: io.kestra.plugin.scripts.python.Script
    description: Compare each control against the previous snapshot and separate
      regressions from improvements.
    containerImage: python:3.12-slim
    inputFiles:
      current.json: |
        {
          "soc2": {"overview": {{ outputs.soc2.overview | toJson }}, "complying": {{ outputs.soc2.totalComplyingRuleCount }}, "total": {{ outputs.soc2.totalRuleCount }}},
          "iso27001": {"overview": {{ outputs.iso27001.overview | toJson }}, "complying": {{ outputs.iso27001.totalComplyingRuleCount }}, "total": {{ outputs.iso27001.totalRuleCount }}},
          "nis2": {"overview": {{ outputs.nis2.overview | toJson }}, "complying": {{ outputs.nis2.totalComplyingRuleCount }}, "total": {{ outputs.nis2.totalRuleCount }}}
        }
      previous.json: "{{ outputs.read_previous.value ?? '{}' }}"
    script: |
      import json

      with open("current.json") as fh:
          current = json.load(fh)
      with open("previous.json") as fh:
          raw = fh.read().strip()
          previous = json.loads(raw) if raw else {}

      def controls(snapshot, framework):
          data = (snapshot.get(framework) or {}).get("overview") or {}
          flat = {}
          for key, value in data.items():
              if isinstance(value, dict):
                  flat[key] = str(value.get("status", "unknown"))
              else:
                  flat[key] = str(value)
          return flat

      regressions = []
      improvements = []
      scores = {}

      for framework in ("soc2", "iso27001", "nis2"):
          now_controls = controls(current, framework)
          was_controls = controls(previous, framework)
          for name, status in now_controls.items():
              before = was_controls.get(name)
              if before is None:
                  continue
              if before != status:
                  entry = {"framework": framework, "control": name, "from": before, "to": status}
                  if status.lower() in ("failing", "false", "non_complying"):
                      regressions.append(entry)
                  else:
                      improvements.append(entry)
          complying = current[framework]["complying"]
          total = current[framework]["total"]
          was_complying = (previous.get(framework) or {}).get("complying")
          scores[framework] = {
              "complying": complying,
              "total": total,
              "pct": round(100.0 * complying / total, 1) if total else 0.0,
              "delta": (complying - was_complying) if isinstance(was_complying, int) else None,
          }

      def describe(items):
          if not items:
              return "none"
          return "; ".join(f"{i['framework']} {i['control']} ({i['from']} to {i['to']})" for i in items).replace('"', "'")

      summary = {
          "first_run": not previous,
          "regressions": regressions,
          "improvements": improvements,
          "regression_count": len(regressions),
          "improvement_count": len(improvements),
          "regressions_text": describe(regressions),
          "improvements_text": describe(improvements),
          "scores_text": "; ".join(
              f"{k.upper()} {v['complying']}/{v['total']} ({v['pct']}%)" + (f" delta {v['delta']:+d}" if v["delta"] is not None else "")
              for k, v in scores.items()
          ),
          "snapshot": current,
      }

      print("::" + json.dumps({"outputs": summary}) + "::")
      print(f"regressions={len(regressions)} improvements={len(improvements)}")

  - id: write_evidence
    type: io.kestra.plugin.ai.agent.AIAgent
    description: Turn the posture delta into the narrative an auditor actually asks
      for, grounded only in the numbers above.
    provider:
      type: io.kestra.plugin.ai.provider.OpenAI
      modelName: gpt-4o-mini
      apiKey: "{{ secret('OPENAI_API_KEY') }}"
    configuration:
      temperature: 0.2
    systemMessage: |
      You write control evidence narratives for a security compliance auditor.
      Use only the figures supplied. Never invent a control, a date, or a remediation that was not given to you.
      Write three short paragraphs of plain prose: current posture, what changed since the last period and why, and what compensating controls or remediation work is in flight.
      Do not use bullet points and do not use headings.
    prompt: |
      Reporting period: {{ inputs.reporting_period }}
      Framework scores: {{ outputs.detect_drift.vars.scores_text }}
      Controls that regressed: {{ outputs.detect_drift.vars.regressions_text }}
      Controls that improved: {{ outputs.detect_drift.vars.improvements_text }}
      Open critical and high findings currently tracked in Aikido: {{ outputs.open_findings.size }}
      First run with no prior snapshot: {{ outputs.detect_drift.vars.first_run }}

  - id: archive_pack
    type: io.kestra.plugin.core.storage.Write
    description: Persist the evidence pack to internal storage so the URI can be
      attached to an audit request later.
    extension: .json
    content: |
      {
        "reporting_period": "{{ inputs.reporting_period }}",
        "generated_at": "{{ now() }}",
        "execution_id": "{{ execution.id }}",
        "scores": "{{ outputs.detect_drift.vars.scores_text }}",
        "regressions": {{ outputs.detect_drift.vars.regressions | toJson }},
        "improvements": {{ outputs.detect_drift.vars.improvements | toJson }},
        "open_critical_high": {{ outputs.open_findings.size }},
        "narrative": {{ outputs.write_evidence.textOutput | toJson }}
      }

  - id: record_snapshot
    type: io.kestra.plugin.core.kv.Set
    description: Store this period's posture as the comparison point for the next run.
    key: compliance_posture
    kvType: JSON
    overwrite: true
    value: "{{ outputs.detect_drift.vars.snapshot | toJson }}"

  - id: drift_alert
    type: io.kestra.plugin.core.flow.If
    description: A regression is a finding in its own right, so it is announced
      rather than left inside the archived pack.
    condition: "{{ outputs.detect_drift.vars.regression_count >=
      inputs.drift_alert_threshold }}"
    then:
      - id: announce_drift
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK') }}"
        payload: |
          {
            "text": "Compliance drift detected for {{ inputs.reporting_period }}",
            "blocks": [
              {"type": "header", "text": {"type": "plain_text", "text": "Compliance posture regressed"}},
              {"type": "section", "text": {"type": "mrkdwn", "text": "*Period* {{ inputs.reporting_period }}\n*Scores* {{ outputs.detect_drift.vars.scores_text }}"}},
              {"type": "section", "text": {"type": "mrkdwn", "text": "*Regressed controls*\n{{ outputs.detect_drift.vars.regressions_text }}"}},
              {"type": "section", "text": {"type": "mrkdwn", "text": "*Improved controls*\n{{ outputs.detect_drift.vars.improvements_text }}"}},
              {"type": "section", "text": {"type": "mrkdwn", "text": "Evidence pack archived at `{{ outputs.archive_pack.uri }}` from execution {{ execution.id }}."}}
            ]
          }
    else:
      - id: log_stable
        type: io.kestra.plugin.core.log.Log
        message: "EVIDENCE pack filed for {{ inputs.reporting_period }} with no
          regression. {{ outputs.detect_drift.vars.scores_text }}. Pack at {{
          outputs.archive_pack.uri }}."

errors:
  - id: evidence_failed
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "text": ":warning: Compliance evidence pack for {{ inputs.reporting_period }} failed to generate. No snapshot was recorded, so the next run still compares against the previous period. Execution {{ execution.id }}."
      }

outputs:
  - id: evidence_uri
    type: STRING
    value: "{{ outputs.archive_pack.uri }}"

  - id: regression_count
    type: INT
    value: "{{ outputs.detect_drift.vars.regression_count }}"