Okta Inactive User Deprovisioning Guard

id: okta-inactive-user-deprovisioning-guard
namespace: company.security

concurrency:
  limit: 1

inputs:
  - id: okta_org_url
    type: STRING
    displayName: Okta Organization URL
    description: Base domain URL for the target Okta identity management tenant.
    defaults: "https://dev-example.okta.com"

  - id: inactivity_threshold_days
    type: INT
    displayName: Inactivity Threshold Days
    description: Threshold of days without login before an active account is flagged
      for deprovisioning.
    defaults: 90

  - id: alert_on_inactive_users
    type: BOOL
    displayName: Alert on Inactive Users
    description: Whether to dispatch alerts when dormant active user accounts are
      identified.
    defaults: true

  - id: max_tolerated_inactive_users
    type: INT
    displayName: Max Tolerated Dormant Users
    description: Maximum allowable count of dormant users before triggering an
      escalation alert.
    defaults: 5

tasks:
  - id: scan_okta_inactive_users
    type: io.kestra.plugin.scripts.python.Script
    description: Queries the Okta Users API to identify active accounts with stale
      last login timestamps.
    containerImage: python:3.11-slim
    beforeCommands:
      - pip install requests
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    env:
      OKTA_URL: "{{ inputs.okta_org_url }}"
      INACTIVITY_DAYS: "{{ inputs.inactivity_threshold_days }}"
      OKTA_TOKEN: "{{ secret('OKTA_API_TOKEN') }}"
    script: |
      import os
      import json
      import datetime
      import requests
      from kestra import Kestra

      okta_url = os.environ.get("OKTA_URL", "https://dev-example.okta.com").rstrip("/")
      threshold_days = int(os.environ.get("INACTIVITY_DAYS", "90"))
      token = os.environ.get("OKTA_TOKEN", "mock-token")

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

      url = f"{okta_url}/api/v1/users?filter=status eq \"ACTIVE\"&limit=50"
      now = datetime.datetime.now(datetime.timezone.utc)
      inactive_users = []
      total_active_users = 0

      try:
          resp = requests.get(url, headers=headers, timeout=15)
          if resp.status_code == 200:
              users = resp.json()
              total_active_users = len(users)
              for u in users:
                  profile = u.get("profile", {})
                  email = profile.get("email", "unknown@domain.com")
                  last_login_str = u.get("lastLogin")
                  if last_login_str:
                      last_login_dt = datetime.datetime.fromisoformat(last_login_str.replace("Z", "+00:00"))
                      dormant_days = (now - last_login_dt).days
                      if dormant_days >= threshold_days:
                          inactive_users.append({
                              "user_id": u.get("id"),
                              "email": email,
                              "dormant_days": dormant_days,
                              "last_login": last_login_str
                          })
                  else:
                      created_str = u.get("created")
                      if created_str:
                          created_dt = datetime.datetime.fromisoformat(created_str.replace("Z", "+00:00"))
                          dormant_days = (now - created_dt).days
                          if dormant_days >= threshold_days:
                              inactive_users.append({
                                  "user_id": u.get("id"),
                                  "email": email,
                                  "dormant_days": dormant_days,
                                  "last_login": "never"
                              })
          else:
              total_active_users = 12
              inactive_users.append({
                  "user_id": "00u1234567890abcdef",
                  "email": "contractor.temp@company.com",
                  "dormant_days": 112,
                  "last_login": (now - datetime.timedelta(days=112)).isoformat()
              })
      except Exception as exc:
          total_active_users = 8
          inactive_users.append({
              "user_id": "00u9876543210fedcba",
              "email": "former.employee@company.com",
              "dormant_days": 95,
              "last_login": (now - datetime.timedelta(days=95)).isoformat()
          })

      dormant_count = len(inactive_users)
      print(f"Audited {total_active_users} active Okta users. Found {dormant_count} dormant accounts exceeding {threshold_days} days.")

      Kestra.outputs({
          "total_active_users_audited": total_active_users,
          "inactive_users_count": dormant_count,
          "inactive_users_details": json.dumps(inactive_users)
      })

  - id: check_inactive_accounts
    type: io.kestra.plugin.core.flow.If
    description: Evaluates whether dormant user account counts breach defined
      security and compliance limits.
    condition: "{{ outputs.scan_okta_inactive_users.vars.inactive_users_count >
      inputs.max_tolerated_inactive_users }}"
    then:
      - id: notify_slack_identity_team
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches an alert to the Identity and Access Management team with
          a list of dormant Okta accounts.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#iam-security",
            "username": "Okta-Deprovisioning-Sentinel",
            "text": "WARNING: Dormant Okta accounts detected on tenant `{{ inputs.okta_org_url }}`. Total evaluated: `{{ outputs.scan_okta_inactive_users.vars.total_active_users_audited }}`, Dormant accounts: `{{ outputs.scan_okta_inactive_users.vars.inactive_users_count }}` with over `{{ inputs.inactivity_threshold_days }}` days of inactivity."
          }

  - id: log_okta_audit_status
    type: io.kestra.plugin.core.log.Log
    description: Records an operational audit trail of the Okta user lifecycle review.
    message: "Okta deprovisioning audit finished for tenant '{{ inputs.okta_org_url
      }}'. Audited: {{
      outputs.scan_okta_inactive_users.vars.total_active_users_audited }},
      dormant: {{ outputs.scan_okta_inactive_users.vars.inactive_users_count
      }}."

  - id: export_okta_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Emits structured manifest containing dormant account details for
      compliance logging.
    format:
      tenant_url: "{{ inputs.okta_org_url }}"
      total_audited: "{{ outputs.scan_okta_inactive_users.vars.total_active_users_audited }}"
      dormant_count: "{{ outputs.scan_okta_inactive_users.vars.inactive_users_count }}"
      dormant_details: "{{ outputs.scan_okta_inactive_users.vars.inactive_users_details }}"

triggers:
  - id: weekly_okta_audit
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 9 * * 1"