Remove LDAP Group Members Whose Entries No Longer Exist

id: ldap-dangling-group-member-cleanup-guard
namespace: company.team

description: |
  Find LDAP group entries whose member DNs point at user entries that no longer exist, then remove those dangling values with an LDIF delete and re-search to prove they are gone.

inputs:
  - id: dry_run
    type: BOOL
    defaults: true
    description: When true the flow only reports dangling member values and never
      writes to the directory.
  - id: auto_remediate
    type: BOOL
    defaults: false
    description: When true (and dry_run is false) the flow removes the dangling
      member values after approval.
  - id: ldap_host
    type: STRING
    defaults: ldap.example.com
    description: Hostname of the LDAP server.
  - id: ldap_port
    type: INT
    defaults: 389
    description: LDAP port.
  - id: bind_dn
    type: STRING
    defaults: cn=admin,dc=example,dc=org
    description: Bind DN with permission to search the tree and modify group entries.
  - id: ldap_password
    type: STRING
    defaults: "{{ secret('LDAP_PASSWORD') }}"
    description: Bind password. Defaults to the LDAP_PASSWORD secret.
  - id: base_dn
    type: STRING
    defaults: dc=example,dc=org
    description: Base DN the group search runs under.
  - id: group_filter
    type: STRING
    defaults: "(|(objectClass=groupOfNames)(objectClass=groupOfUniqueNames))"
    description: LDAP filter that selects the group entries to audit.
  - id: member_attribute
    type: SELECT
    values:
      - member
      - uniqueMember
    defaults: member
    description: Attribute holding the member DNs. member for groupOfNames,
      uniqueMember for groupOfUniqueNames.
  - id: resolve_base_dn
    type: STRING
    defaults: dc=example,dc=org
    description: Base DN of the existence search. A member DN outside this subtree
      is reported as out of scope and never removed, because this search could
      not have seen it.
  - id: resolve_filter
    type: STRING
    defaults: "(objectClass=*)"
    description: Filter of the existence search. Keep it wide enough to return every
      entry a group may legitimately reference.
  - id: protected_object_classes
    type: STRING
    defaults: groupOfNames,groupOfUniqueNames
    description: Comma separated object classes whose schema requires at least one
      member. A group carrying one of these is refused rather than emptied.
  - id: min_members_required
    type: INT
    defaults: 1
    description: Lowest member count a protected group may be left with. Removing
      values that would cross this floor is refused.
  - id: max_removals
    type: INT
    defaults: 200
    description: Ceiling on how many member values one run may remove. A larger
      count means the existence search is probably mis-scoped, so the run
      reports and refuses to write.

tasks:
  - id: detect_groups
    type: io.kestra.plugin.ldap.Search
    description: Read every group entry under the base DN with its object classes
      and its member values. The search writes the entries to an LDIF file.
    hostname: "{{ inputs.ldap_host }}"
    port: "{{ inputs.ldap_port }}"
    userDn: "{{ inputs.bind_dn }}"
    password: "{{ inputs.ldap_password }}"
    baseDn: "{{ inputs.base_dn }}"
    filter: "{{ inputs.group_filter }}"
    attributes:
      - cn
      - objectClass
      - member
      - uniqueMember

  - id: resolve_entries
    type: io.kestra.plugin.ldap.Search
    description: Read the DN of every entry under the resolve base. This is the
      existence universe the member DNs are checked against, so a member that
      resolves here is left alone.
    hostname: "{{ inputs.ldap_host }}"
    port: "{{ inputs.ldap_port }}"
    userDn: "{{ inputs.bind_dn }}"
    password: "{{ inputs.ldap_password }}"
    baseDn: "{{ inputs.resolve_base_dn }}"
    filter: "{{ inputs.resolve_filter }}"
    attributes:
      - objectClass

  - id: analyze
    type: io.kestra.plugin.scripts.python.Script
    description: >-
      Diff the member values of every group against the set of DNs that exist. A
      value that is in scope and does not resolve is dangling. A group whose
      schema needs a member is refused rather than emptied. The target list is
      emptied altogether when the existence search looks mis-scoped.
    containerImage: python:3.12-slim
    script: |
      import base64, json, re

      MEMBER_ATTR = "{{ inputs.member_attribute }}".strip().lower()
      RESOLVE_BASE = "{{ inputs.resolve_base_dn }}"
      PROTECTED = set(
          c.strip().lower()
          for c in "{{ inputs.protected_object_classes }}".split(",")
          if c.strip()
      )
      MIN_MEMBERS = {{ inputs.min_members_required }}
      MAX_REMOVALS = {{ inputs.max_removals }}


      def split_rdns(dn):
          out = []
          buf = ""
          esc = False
          for ch in dn:
              if esc:
                  buf += ch
                  esc = False
              elif ch == "\\":
                  buf += ch
                  esc = True
              elif ch == ",":
                  out.append(buf)
                  buf = ""
              else:
                  buf += ch
          out.append(buf)
          return out


      def norm(dn):
          return ",".join(p.strip() for p in split_rdns(dn)).lower()


      def parse_ldif(path):
          try:
              raw = open(path, "r", encoding="utf-8", errors="replace").read()
          except OSError:
              return []
          unfolded = re.sub(r"\n[ ]", "", raw)
          entries = []
          cur = None
          for line in unfolded.splitlines():
              if not line.strip():
                  if cur is not None:
                      entries.append(cur)
                  cur = None
                  continue
              if line.startswith("#") or ":" not in line:
                  continue
              key, _, val = line.partition(":")
              key = key.strip().lower()
              val = val.lstrip()
              if val.startswith(":"):
                  try:
                      val = base64.b64decode(val[1:].strip()).decode("utf-8", "replace")
                  except Exception:
                      continue
              elif val.startswith("<"):
                  continue
              if key == "version":
                  continue
              if key == "dn":
                  cur = {"dn": val, "attrs": {}}
                  continue
              if cur is None:
                  continue
              cur["attrs"].setdefault(key, []).append(val)
          if cur is not None:
              entries.append(cur)
          return entries


      groups = parse_ldif("groups.ldif")
      present = parse_ldif("entries.ldif")
      existing = set(norm(e["dn"]) for e in present)
      resolve_suffix = norm(RESOLVE_BASE)

      targets = []
      refused = []
      dangling_lines = []
      out_of_scope_lines = []
      dangling_total = 0
      out_of_scope_total = 0
      removable_total = 0
      affected = 0

      for g in groups:
          members = g["attrs"].get(MEMBER_ATTR, [])
          classes = set(c.strip().lower() for c in g["attrs"].get("objectclass", []))
          dangling = []
          out_of_scope = []
          for m in members:
              nm = norm(m)
              if not nm:
                  continue
              in_scope = nm == resolve_suffix or nm.endswith("," + resolve_suffix)
              if not in_scope:
                  out_of_scope.append(m)
              elif nm not in existing:
                  dangling.append(m)
          out_of_scope_total += len(out_of_scope)
          for m in out_of_scope:
              out_of_scope_lines.append(g["dn"] + " -> " + m)
          if not dangling:
              continue
          affected += 1
          dangling_total += len(dangling)
          for m in dangling:
              dangling_lines.append(g["dn"] + " -> " + m)
          remaining = len(members) - len(dangling)
          if classes & PROTECTED and remaining < MIN_MEMBERS:
              refused.append(
                  g["dn"]
                  + " (removing "
                  + str(len(dangling))
                  + " of "
                  + str(len(members))
                  + " would leave "
                  + str(remaining)
                  + ", schema floor is "
                  + str(MIN_MEMBERS)
                  + ")"
              )
              continue
          removable_total += len(dangling)
          targets.append(
              {"group_dn": g["dn"], "attr": MEMBER_ATTR, "dns": dangling}
          )

      abort = ""
      if len(existing) == 0:
          abort = "the existence search under " + RESOLVE_BASE + " returned no entries, so every member DN would look dangling"
      elif removable_total > MAX_REMOVALS:
          abort = (
              "removable values "
              + str(removable_total)
              + " exceed the max_removals ceiling "
              + str(MAX_REMOVALS)
          )
      if abort:
          targets = []
          removable_total = 0

      print(
          "SCAN groups={} resolved_entries={} affected_groups={} dangling={} removable={} refused={} out_of_scope={}".format(
              len(groups),
              len(existing),
              affected,
              dangling_total,
              removable_total,
              len(refused),
              out_of_scope_total,
          )
      )
      for line in dangling_lines:
          print("DANGLING " + line)
      for line in refused:
          print("REFUSED " + line)
      for line in out_of_scope_lines:
          print("OUT_OF_SCOPE " + line)
      if abort:
          print("ABORT " + abort)

      out = {
          "group_count": str(len(groups)),
          "resolved_count": str(len(existing)),
          "affected_groups": str(affected),
          "dangling_total": str(dangling_total),
          "removable_total": str(removable_total),
          "refused_count": str(len(refused)),
          "out_of_scope_total": str(out_of_scope_total),
          "has_targets": "yes" if targets else "no",
          "has_findings": "yes" if (dangling_total or refused or abort) else "no",
          "safe_to_write": "no" if abort else "yes",
          "abort_reason": abort,
          "dangling_dns": "; ".join(dangling_lines) if dangling_lines else "none",
          "refused_groups": "; ".join(refused) if refused else "none",
          "out_of_scope_dns": "; ".join(out_of_scope_lines) if out_of_scope_lines else "none",
          "targets_json": json.dumps(targets),
      }
      print("::" + json.dumps({"outputs": out}) + "::")
    inputFiles:
      groups.ldif: "{{ outputs.detect_groups.uri }}"
      entries.ldif: "{{ outputs.resolve_entries.uri }}"

  - id: findings_alert
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.analyze.vars.has_findings == 'yes' }}"
    then:
      - id: log_findings
        type: io.kestra.plugin.core.log.Log
        message: "Dangling member values: {{ outputs.analyze.vars.dangling_total }}
          across {{ outputs.analyze.vars.affected_groups }} group(s). Removable:
          {{ outputs.analyze.vars.removable_total }}. Refused: {{
          outputs.analyze.vars.refused_groups }}. Out of scope: {{
          outputs.analyze.vars.out_of_scope_dns }}. Abort reason: {{
          outputs.analyze.vars.abort_reason }}."
      - id: notify_findings
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Post the detection result to Slack so an operator sees the drift
          even on a read-only run.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "LDAP dangling group member guard: {{ outputs.analyze.vars.dangling_total }} member value(s) in {{ outputs.analyze.vars.affected_groups }} group(s) under {{ inputs.base_dn }} point at entries that do not exist. Removable now: {{ outputs.analyze.vars.removable_total }}. Refused on the schema floor: {{ outputs.analyze.vars.refused_groups }}. Dangling: {{ outputs.analyze.vars.dangling_dns }}. dry_run={{ inputs.dry_run }} auto_remediate={{ inputs.auto_remediate }}."
          }

  - id: remediate_gate
    type: io.kestra.plugin.core.flow.If
    condition: "{{ inputs.auto_remediate and not inputs.dry_run and
      outputs.analyze.vars.has_targets == 'yes' and
      outputs.analyze.vars.safe_to_write == 'yes' }}"
    then:
      - id: approval_hold
        type: io.kestra.plugin.core.flow.Pause
        description: >-
          Removing a member value changes who is in a group, so the write waits
          for a human. Resume with approved set to true to apply the delete. Set
          it false to cancel.
        onResume:
          - id: approved
            type: BOOL
            required: false
            description: Set true to remove the dangling member values listed by the
              detection step.
      - id: apply_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ outputs.approval_hold.onResume.approved ?? false }}"
        then:
          - id: remediate_members
            type: io.kestra.plugin.core.flow.Loop
            description: Remediation loop. One LDIF delete per group, concurrencyLimit 1 so
              the modifies run in turn and never race on the same group entry.
            concurrencyLimit: 1
            values: "{{ outputs.analyze.vars.targets_json }}"
            tasks:
              - id: build_record
                type: io.kestra.plugin.scripts.python.Script
                description: Write the LDIF delete record for this group. Each dangling DN is
                  listed under one delete of the member attribute, so resolving
                  members are never named and never touched.
                containerImage: python:3.12-slim
                outputFiles:
                  - one.ldif
                script: |
                  import json

                  item = json.loads(r"""{{ item.value }}""")
                  lines = [
                      "dn: " + item["group_dn"],
                      "changetype: modify",
                      "delete: " + item["attr"],
                  ]
                  for dn in item["dns"]:
                      lines.append(item["attr"] + ": " + dn)
                  lines.append("-")
                  lines.append("")
                  open("one.ldif", "w", encoding="utf-8").write("\n".join(lines) + "\n")
                  print(
                      "LDIF delete {} value(s) from {}".format(
                          len(item["dns"]), item["group_dn"]
                      )
                  )
              - id: apply_record
                type: io.kestra.plugin.ldap.Modify
                description: Apply the LDIF delete. The plugin logs a rejected record and keeps
                  going without failing, so allowFailure is set and the
                  re-search below is the gate that decides the run.
                allowFailure: true
                hostname: "{{ inputs.ldap_host }}"
                port: "{{ inputs.ldap_port }}"
                userDn: "{{ inputs.bind_dn }}"
                password: "{{ inputs.ldap_password }}"
                inputs:
                  - "{{ outputs.build_record.outputFiles['one.ldif'] }}"
          - id: verify_groups
            type: io.kestra.plugin.ldap.Search
            description: Read the group entries again after the modifies.
            hostname: "{{ inputs.ldap_host }}"
            port: "{{ inputs.ldap_port }}"
            userDn: "{{ inputs.bind_dn }}"
            password: "{{ inputs.ldap_password }}"
            baseDn: "{{ inputs.base_dn }}"
            filter: "{{ inputs.group_filter }}"
            attributes:
              - cn
              - objectClass
              - member
              - uniqueMember
          - id: verify_entries
            type: io.kestra.plugin.ldap.Search
            description: Read the existence universe again, so a member created during the
              pause counts as resolving rather than as a failure.
            hostname: "{{ inputs.ldap_host }}"
            port: "{{ inputs.ldap_port }}"
            userDn: "{{ inputs.bind_dn }}"
            password: "{{ inputs.ldap_password }}"
            baseDn: "{{ inputs.resolve_base_dn }}"
            filter: "{{ inputs.resolve_filter }}"
            attributes:
              - objectClass
          - id: analyze_verify
            type: io.kestra.plugin.scripts.python.Script
            description: >-
              Re-run the same diff against the targeted groups only. Any
              targeted value still present means a modify was rejected, which
              the Modify task never reports, so this count is what the run is
              judged on.
            containerImage: python:3.12-slim
            script: |
              import base64, json, re

              MEMBER_ATTR = "{{ inputs.member_attribute }}".strip().lower()
              RESOLVE_BASE = "{{ inputs.resolve_base_dn }}"
              TARGETS = json.loads(r"""{{ outputs.analyze.vars.targets_json }}""")


              def split_rdns(dn):
                  out = []
                  buf = ""
                  esc = False
                  for ch in dn:
                      if esc:
                          buf += ch
                          esc = False
                      elif ch == "\\":
                          buf += ch
                          esc = True
                      elif ch == ",":
                          out.append(buf)
                          buf = ""
                      else:
                          buf += ch
                  out.append(buf)
                  return out


              def norm(dn):
                  return ",".join(p.strip() for p in split_rdns(dn)).lower()


              def parse_ldif(path):
                  try:
                      raw = open(path, "r", encoding="utf-8", errors="replace").read()
                  except OSError:
                      return []
                  unfolded = re.sub(r"\n[ ]", "", raw)
                  entries = []
                  cur = None
                  for line in unfolded.splitlines():
                      if not line.strip():
                          if cur is not None:
                              entries.append(cur)
                          cur = None
                          continue
                      if line.startswith("#") or ":" not in line:
                          continue
                      key, _, val = line.partition(":")
                      key = key.strip().lower()
                      val = val.lstrip()
                      if val.startswith(":"):
                          try:
                              val = base64.b64decode(val[1:].strip()).decode("utf-8", "replace")
                          except Exception:
                              continue
                      elif val.startswith("<"):
                          continue
                      if key == "version":
                          continue
                      if key == "dn":
                          cur = {"dn": val, "attrs": {}}
                          continue
                      if cur is None:
                          continue
                      cur["attrs"].setdefault(key, []).append(val)
                  if cur is not None:
                      entries.append(cur)
                  return entries


              groups = parse_ldif("groups.ldif")
              present = parse_ldif("entries.ldif")
              existing = set(norm(e["dn"]) for e in present)
              by_dn = {}
              for g in groups:
                  by_dn[norm(g["dn"])] = g

              wanted = {}
              for t in TARGETS:
                  wanted[norm(t["group_dn"])] = set(norm(d) for d in t["dns"])

              survivors = []
              removed = 0
              collateral = []
              for key, want in wanted.items():
                  g = by_dn.get(key)
                  if g is None:
                      survivors.append(key + " -> group entry missing from the re-search")
                      continue
                  live = [norm(m) for m in g["attrs"].get(MEMBER_ATTR, [])]
                  for dn in sorted(want):
                      if dn in live:
                          survivors.append(g["dn"] + " -> " + dn)
                      else:
                          removed += 1
                  for m in live:
                      if m in want:
                          continue
                      if m and m not in existing:
                          collateral.append(g["dn"] + " -> " + m)

              print(
                  "VERIFY targeted_groups={} removed_values={} still_present={} unresolved_left={}".format(
                      len(wanted), removed, len(survivors), len(collateral)
                  )
              )
              for line in survivors:
                  print("STILL_PRESENT " + line)
              for line in collateral:
                  print("STILL_UNRESOLVED " + line)
              out = {
                  "removed_values": str(removed),
                  "still_present": str(len(survivors)),
                  "has_survivors": "yes" if survivors else "no",
                  "survivor_dns": "; ".join(survivors) if survivors else "none",
                  "unresolved_left": "; ".join(collateral) if collateral else "none",
              }
              print("::" + json.dumps({"outputs": out}) + "::")
            inputFiles:
              groups.ldif: "{{ outputs.verify_groups.uri }}"
              entries.ldif: "{{ outputs.verify_entries.uri }}"
          - id: log_verify
            type: io.kestra.plugin.core.log.Log
            message: "Removed {{ outputs.analyze_verify.vars.removed_values }} dangling
              member value(s). Still present: {{
              outputs.analyze_verify.vars.survivor_dns }}."
          - id: assert_clean
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.analyze_verify.vars.has_survivors == 'yes' }}"
            then:
              - id: still_dangling
                type: io.kestra.plugin.core.execution.Fail
                errorMessage: "The re-search still finds targeted member values that do not
                  resolve: {{ outputs.analyze_verify.vars.survivor_dns }}. The
                  Modify task does not fail on a rejected record, so this read
                  is the only proof and it says the cleanup did not land."

triggers:
  - id: daily_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Daily audit. Disabled by default and pinned to safe mode so
      enabling it never writes without a deliberate change.
    cron: "0 5 * * *"
    disabled: true
    inputs:
      dry_run: true
      auto_remediate: false

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert on any failure, including a run that failed because a
      targeted member value is still in the group after the modify.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "LDAP dangling group member cleanup guard FAILED (execution {{ execution.id }}). Group membership may still reference entries that do not exist. Check the directory and the run logs."
      }