id: aws-cloudtrail-management-coverage-guard
namespace: company.team
description: Check a CloudTrail trail for disabled logging, incomplete
management-event coverage, and integrity configuration drift.
inputs:
- id: region
type: STRING
defaults: us-east-1
description: AWS Region where the trail was created. Run in the home account and Region.
- id: trail_name
type: STRING
defaults: production-audit
description: Exact CloudTrail trail name in its home Region.
- id: max_delivery_age_hours
type: INT
defaults: 0
description: Optional S3 delivery freshness check. 0 disables it; 1-168 enables
it for active accounts.
triggers:
- id: daily_audit
type: io.kestra.plugin.core.trigger.Schedule
cron: '0 7 * * *'
disabled: true
description: Enable after checking the actual trail and Slack delivery.
tasks:
- id: validate_inputs
type: io.kestra.plugin.scripts.node.Script
containerImage: node:22-alpine
description: Reject invalid trail and Region values before calling AWS.
env:
AWS_REGION: '{{ inputs.region }}'
TRAIL_NAME: '{{ inputs.trail_name }}'
MAX_DELIVERY_AGE_HOURS: '{{ inputs.max_delivery_age_hours }}'
script: |
const {AWS_REGION:region,TRAIL_NAME:name,MAX_DELIVERY_AGE_HOURS:limit}=process.env;
if(!/^[a-z]{2}(-gov)?-[a-z]+-\d$/.test(region||'')||!/^[A-Za-z0-9](?:[A-Za-z0-9_.-]{1,126})[A-Za-z0-9]$/.test(name||'')||!/^\d+$/.test(limit)||!Number.isSafeInteger(Number(limit))||Number(limit)>168)throw Error('Invalid CloudTrail Region, trail name, or delivery age');
console.log('CloudTrail input check passed');
- id: read_trail
type: io.kestra.plugin.scripts.python.Script
containerImage: python:3.12-slim
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
dependencies:
- boto3==1.40.58
env:
AWS_REGION: '{{ inputs.region }}'
TRAIL_NAME: '{{ inputs.trail_name }}'
AWS_ACCESS_KEY_ID: "{{ secret('AWS_ACCESS_KEY_ID') }}"
AWS_SECRET_ACCESS_KEY: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
outputFiles:
- trail-snapshot.json
description: Read one trail, its logging state, and its event selectors. No AWS
resources are changed.
script: |
import boto3
import json
import os
from datetime import datetime
region = os.environ['AWS_REGION']
name = os.environ['TRAIL_NAME']
client = boto3.client('cloudtrail', region_name=region)
trails = client.describe_trails(includeShadowTrails=False).get('trailList', [])
if not isinstance(trails, list) or len(trails) > 500:
raise ValueError('CloudTrail returned an invalid or oversized trail list')
found = [t for t in trails if t.get('Name') == name and t.get('HomeRegion') == region]
if len(found) > 1:
raise ValueError('CloudTrail returned duplicate trail identities')
snapshot = {'found': bool(found), 'region': region, 'name': name}
if found:
trail = found[0]
arn = trail.get('TrailARN')
if not isinstance(arn, str) or not arn:
raise ValueError('CloudTrail trail ARN is missing')
snapshot['trail'] = trail
snapshot['status'] = client.get_trail_status(Name=arn)
snapshot['selectors'] = client.get_event_selectors(TrailName=arn)
with open(os.path.join(os.environ['WORKING_DIR'], 'trail-snapshot.json'), 'w', encoding='utf-8') as output:
json.dump(snapshot, output, default=lambda v: v.isoformat() if isinstance(v, datetime) else str(v))
print('CloudTrail API snapshot saved')
- id: check_coverage
type: io.kestra.plugin.scripts.node.Script
containerImage: node:22-alpine
description: Evaluate configuration, management-event selectors, and optional
delivery freshness.
env:
AWS_REGION: '{{ inputs.region }}'
TRAIL_NAME: '{{ inputs.trail_name }}'
MAX_DELIVERY_AGE_HOURS: '{{ inputs.max_delivery_age_hours }}'
inputFiles:
trail-snapshot.json: '{{ outputs.read_trail.outputFiles["trail-snapshot.json"] }}'
outputFiles:
- trail-coverage-report.json
script: |
const fs=require('fs'),path=require('path');
function evaluate(s,now,limit,region,name){
const obj=x=>x!==null&&typeof x==='object'&&!Array.isArray(x);
const reasons=[],notes=[];
if(!obj(s)||s.found!==true){return {status:'BLOCK',trail:name,region,reasons:['Named trail not found in its home Region'],notes:[]};}
const t=s.trail,st=s.status,sel=s.selectors;
if(!obj(t)||!obj(st)||!obj(sel))return {status:'BLOCK',trail:name,region,reasons:['CloudTrail response is incomplete'],notes:[]};
if(t.Name!==name||t.HomeRegion!==region||typeof t.TrailARN!=='string'||!t.TrailARN)reasons.push('Trail identity or home Region does not match');
if(st.IsLogging!==true)reasons.push('CloudTrail logging is not enabled');
if(t.IsMultiRegionTrail!==true)reasons.push('Trail does not cover all Regions');
if(t.IncludeGlobalServiceEvents!==true)reasons.push('Global service events are not included');
if(t.LogFileValidationEnabled!==true)reasons.push('Log file validation is not enabled');
if(typeof t.S3BucketName!=='string'||!t.S3BucketName)reasons.push('Trail S3 destination is missing');
const standard=Array.isArray(sel.EventSelectors)?sel.EventSelectors:[];
const advanced=Array.isArray(sel.AdvancedEventSelectors)?sel.AdvancedEventSelectors:[];
const fullBasic=standard.filter(x=>obj(x)&&x.IncludeManagementEvents===true&&(x.ExcludeManagementEventSources===undefined||(Array.isArray(x.ExcludeManagementEventSources)&&x.ExcludeManagementEventSources.length===0)));
const allBasic=fullBasic.some(x=>x.ReadWriteType==='All')||(fullBasic.some(x=>x.ReadWriteType==='ReadOnly')&&fullBasic.some(x=>x.ReadWriteType==='WriteOnly'));
const allAdvanced=advanced.some(x=>obj(x)&&Array.isArray(x.FieldSelectors)&&x.FieldSelectors.length===1&&obj(x.FieldSelectors[0])&&x.FieldSelectors[0].Field==='eventCategory'&&Array.isArray(x.FieldSelectors[0].Equals)&&x.FieldSelectors[0].Equals.length===1&&x.FieldSelectors[0].Equals[0]==='Management'&&Object.keys(x.FieldSelectors[0]).every(k=>k==='Field'||k==='Equals'));
if(!allBasic&&!allAdvanced)reasons.push('Selectors do not show full read/write management-event coverage');
if(sel.TrailARN!==t.TrailARN)reasons.push('Event selectors belong to a different trail');
if(typeof st.LatestDeliveryError==='string'&&st.LatestDeliveryError)notes.push('CloudTrail reports an S3 delivery error; check its history');
if(typeof st.LatestDigestDeliveryError==='string'&&st.LatestDigestDeliveryError)notes.push('CloudTrail reports a digest delivery error; check its history');
let deliveryAge=null;
if(limit>0){
const delivered=Date.parse(st.LatestDeliveryTime);
if(!Number.isFinite(delivered)||delivered>now+120000)reasons.push('Latest S3 delivery time is missing or invalid');
else {deliveryAge=Math.max(0,Math.floor((now-delivered)/3600000));if(now-delivered>limit*3600000)reasons.push('Latest S3 delivery is older than the configured limit');}
}
return {status:reasons.length?'BLOCK':'PASS',trail:name,region,log_file_validation:t.LogFileValidationEnabled===true,multi_region:t.IsMultiRegionTrail===true,management_events_complete:allBasic||allAdvanced,logging:st.IsLogging===true,latest_delivery_age_hours:deliveryAge,reasons,notes};
}
const limit=Number(process.env.MAX_DELIVERY_AGE_HOURS);
if(!Number.isSafeInteger(limit)||limit<0||limit>168)throw Error('Invalid delivery age threshold');
const work=process.env.WORKING_DIR;
if(!work)throw Error('Kestra working directory is not set');
const raw=fs.readFileSync(path.join(work,'trail-snapshot.json'),'utf8');
if(Buffer.byteLength(raw)>1048576)throw Error('CloudTrail response exceeds 1 MiB');
const report=evaluate(JSON.parse(raw),Date.now(),limit,process.env.AWS_REGION,process.env.TRAIL_NAME);
fs.writeFileSync(path.join(work,'trail-coverage-report.json'),JSON.stringify(report,null,2));
console.log('::'+JSON.stringify({outputs:{status:report.status,reason_count:report.reasons.length,notes_count:report.notes.length}})+'::');
console.log('CloudTrail coverage: '+report.status);
- id: publish_report
type: io.kestra.plugin.core.output.OutputValues
description: Publish the decision and evidence URI before the BLOCK path.
values:
status: '{{ outputs.check_coverage.vars.status }}'
report_uri: "{{ outputs.check_coverage.outputFiles['trail-coverage-report.json'] }}"
reason_count: '{{ outputs.check_coverage.vars.reason_count }}'
- id: coverage_gate
type: io.kestra.plugin.core.flow.If
description: Block if the selected trail does not meet the management-event
logging policy.
condition: "{{ outputs.check_coverage.vars.status == 'BLOCK' }}"
then:
- id: notify_coverage_gap
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: '{"text":"CloudTrail BLOCK: {{ inputs.trail_name }} ({{ inputs.region
}}); findings={{ outputs.check_coverage.vars.reason_count }};
execution={{ execution.id }}. Read trail-coverage-report.json."}'
- id: stop_coverage_gate
type: io.kestra.plugin.core.execution.Fail
errorMessage: CloudTrail management-event coverage check blocked. Read
trail-coverage-report.json.
else:
- id: log_coverage_pass
type: io.kestra.plugin.core.log.Log
message: 'CloudTrail PASS: {{ inputs.trail_name }} in {{ inputs.region }}. This
is a configuration check.'
errors:
- id: log_cloudtrail_error
type: io.kestra.plugin.core.log.Log
message: 'CloudTrail API or coverage check failed. No PASS result. Execution {{
execution.id }}.'
