Auth0 Tenant Failed Login Anomaly Sentinel

id: auth0-tenant-failed-login-anomaly-sentinel
namespace: company.sre

description: |
  Audits Auth0 tenant authentication event streams to detect anomalous failed login velocity,
  isolating credential stuffing and brute-force attacks to protect identity infrastructure.

concurrency:
  limit: 1

inputs:
  - id: auth0_domain
    type: STRING
    displayName: Auth0 Domain
    defaults: company.us.auth0.com
    description: Auth0 tenant domain identifier hosting customer or workforce
      authentication.

  - id: failed_logins_threshold
    type: INT
    displayName: Failed Logins Threshold
    defaults: 25
    description: Total failed login events within the query lookback window
      triggering security alerts.

  - id: lookback_minutes
    type: INT
    displayName: Lookback Window (Minutes)
    defaults: 15
    description: Time range in minutes across which Auth0 authentication logs are audited.

  - id: alert_on_breach
    type: BOOL
    displayName: Alert on Breach
    defaults: true
    description: Whether to dispatch automated Slack incident notifications when
      threshold breaches are detected.

tasks:
  - id: scan_auth0_failed_logins
    type: io.kestra.plugin.scripts.python.Script
    description: Queries Auth0 Management API v2 for failed authentication and
      credential stuffing anomalies.
    containerImage: python:3.11-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    beforeCommands:
      - pip install requests
    env:
      AUTH0_DOMAIN: "{{ inputs.auth0_domain }}"
      AUTH0_TOKEN: "{{ secret('AUTH0_MANAGEMENT_TOKEN') }}"
      LOOKBACK_MINS: "{{ inputs.lookback_minutes }}"
      THRESHOLD: "{{ inputs.failed_logins_threshold }}"
    script: |
      import os
      import json
      from datetime import datetime, timedelta, timezone
      import requests
      from kestra import Kestra

      domain = os.environ.get("AUTH0_DOMAIN", "").strip()
      token = os.environ.get("AUTH0_TOKEN", "").strip()
      lookback_mins = int(os.environ.get("LOOKBACK_MINS", 15))
      threshold = int(os.environ.get("THRESHOLD", 25))

      headers = {
          "Authorization": f"Bearer {token}",
          "Accept": "application/json"
      }

      url = f"https://{domain}/api/v2/logs"
      params = {
          "q": 'type:"f" OR type:"fu"',
          "per_page": 100,
          "sort": "date:-1"
      }

      failed_count = 0
      ip_counts = {}
      targeted_usernames = []
      now = datetime.now(timezone.utc)
      cutoff = now - timedelta(minutes=lookback_mins)

      try:
          resp = requests.get(url, headers=headers, params=params, timeout=20)
          if resp.status_code == 200:
              logs = resp.json()
              for log in logs:
                  log_date_str = log.get("date")
                  if log_date_str:
                      try:
                          log_dt = datetime.fromisoformat(log_date_str.replace("Z", "+00:00"))
                          if log_dt < cutoff:
                              continue
                      except Exception:
                          pass
                  
                  failed_count += 1
                  ip = log.get("ip", "unknown")
                  ip_counts[ip] = ip_counts.get(ip, 0) + 1
                  user = log.get("user_name") or log.get("user_id")
                  if user and user not in targeted_usernames:
                      targeted_usernames.append(user)
          else:
              print(f"Auth0 API returned status code {resp.status_code}: {resp.text}")
      except Exception as e:
          print(f"Error querying Auth0 logs: {e}")

      is_anomaly = failed_count >= threshold
      top_suspicious_ips = sorted(ip_counts.items(), key=lambda x: x[1], reverse=True)[:5]

      report = {
          "domain": domain,
          "failed_count": failed_count,
          "threshold": threshold,
          "is_anomaly": is_anomaly,
          "top_ips": [ip[0] for ip in top_suspicious_ips],
          "unique_targeted_users": len(targeted_usernames),
          "recommended_action": f"Inspect Auth0 Attack Protection dashboard at https://manage.auth0.com/#/security and configure IP rate limits." if is_anomaly else "Authentication traffic nominal."
      }

      with open("auth0_report.json", "w") as f:
          json.dump(report, f)

      Kestra.outputs({
          "domain": domain,
          "failed_login_count": failed_count,
          "is_anomaly": is_anomaly,
          "top_suspicious_ips": json.dumps([ip[0] for ip in top_suspicious_ips]),
          "unique_targeted_users": len(targeted_usernames)
      })

  - id: check_failed_login_anomaly
    type: io.kestra.plugin.core.flow.If
    description: Branch execution based on whether failed login events exceeded
      operational security thresholds.
    condition: "{{ outputs.scan_auth0_failed_logins.vars.is_anomaly and
      inputs.alert_on_breach }}"
    then:
      - id: notify_slack_security_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches high-priority credential stuffing and brute-force
          security alert to SecOps on Slack.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#secops-alerts",
            "username": "Auth0-Security-Sentinel",
            "text": "CRITICAL: Auth0 tenant `{{ inputs.auth0_domain }}` detected anomalous failed login surge! Failed Logins: `{{ outputs.scan_auth0_failed_logins.vars.failed_login_count }}` (Threshold: `{{ inputs.failed_logins_threshold }}`). Suspicious Source IPs: `{{ outputs.scan_auth0_failed_logins.vars.top_suspicious_ips }}`. Unique Targeted Users: `{{ outputs.scan_auth0_failed_logins.vars.unique_targeted_users }}`."
          }
    else:
      - id: log_nominal_auth_state
        type: io.kestra.plugin.core.log.Log
        description: Records operational telemetry confirming authentication activity
          complies with security baselines.
        message: "Auth0 tenant {{ inputs.auth0_domain }} authentication audit nominal:
          {{ outputs.scan_auth0_failed_logins.vars.failed_login_count }} failed
          attempts within {{ inputs.lookback_minutes }} minutes."

  - id: export_auth0_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Exports structured compliance manifest detailing authentication
      anomalies for SIEM ingestion.
    format:
      domain: "{{ inputs.auth0_domain }}"
      failed_logins: "{{ outputs.scan_auth0_failed_logins.vars.failed_login_count }}"
      is_anomaly: "{{ outputs.scan_auth0_failed_logins.vars.is_anomaly }}"
      suspicious_ips: "{{ outputs.scan_auth0_failed_logins.vars.top_suspicious_ips }}"
      audited_at: "{{ execution.startDate }}"

triggers:
  - id: scheduled_auth0_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Automated scheduled trigger inspecting Auth0 tenant logs every
      fifteen minutes.
    cron: "*/15 * * * *"