AWS Certificate Manager SSL/TLS Certificate Impending Expiration Sentinel

id: aws-acm-certificate-expiration-sentinel
namespace: company.security

inputs:
  - id: aws_region
    type: STRING
    defaults: us-east-1
    description: AWS region where ACM certificates are issued.
  - id: warning_threshold_days
    type: INT
    defaults: 30
    description: Minimum remaining validity in days before triggering an expiration warning.

tasks:
  - id: audit_acm_certificates
    type: io.kestra.plugin.aws.cli.AwsCLI
    region: "{{ inputs.aws_region }}"
    accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
    secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
    commands:
      - |
        aws acm list-certificates --output json > acm_certs.json
        python3 -c "
        import json, subprocess, datetime
        with open('acm_certs.json') as f:
            certs = json.load(f).get('CertificateSummaryList', [])
        expiring_certs = []
        now = datetime.datetime.utcnow()
        for c in certs[:30]:
            arn = c.get('CertificateArn')
            desc_cmd = f'aws acm describe-certificate --certificate-arn {arn} --output json'
            res = subprocess.run(desc_cmd, shell=True, capture_output=True, text=True)
            if res.returncode == 0:
                cert_detail = json.loads(res.stdout).get('Certificate', {})
                not_after = cert_detail.get('NotAfter')
                if not_after:
                    exp_date = datetime.datetime.fromtimestamp(not_after)
                    days_left = (exp_date - now).days
                    if days_left <= {{ inputs.warning_threshold_days }}:
                        expiring_certs.append({
                            'domain': cert_detail.get('DomainName'),
                            'arn': arn,
                            'days_remaining': days_left,
                            'renewal_eligibility': cert_detail.get('RenewalEligibility', 'unknown')
                        })
        result = {
            'region': '{{ inputs.aws_region }}',
            'expiring_count': len(expiring_certs),
            'expiring_certificates': expiring_certs[:5],
            'alert_triggered': len(expiring_certs) > 0
        }
        with open('acm_summary.json', 'w') as out:
            json.dump(result, out)
        print('ACM audit:', result)
        "
    outputFiles:
      - acm_summary.json

  - id: evaluate_cert_expiration
    type: io.kestra.plugin.core.flow.If
    condition: "{{
      fromJson(read(outputs.audit_acm_certificates.outputFiles['acm_summary.jso\
      n'])).alert_triggered }}"
    then:
      - id: alert_security_team
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK') }}"
        payload: |
          {
            "text": ":warning: *AWS Certificate Expiration Alert*: Detected {{ fromJson(read(outputs.audit_acm_certificates.outputFiles['acm_summary.json'])).expiring_count }} SSL/TLS certificates expiring in less than {{ inputs.warning_threshold_days }} days in {{ inputs.aws_region }}. Check DNS validation status or import renewed certificates."
          }

errors:
  - id: notify_sentinel_error
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "text": ":warning: *ACM Sentinel Execution Failure*: Failed to audit certificates in {{ inputs.aws_region }}."
      }

triggers:
  - id: weekly_cert_audit
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 8 * * 1"
    disabled: true