SCIM Provisioning and Lifecycle: Automating Joiner, Mover, Leaver

mkdir -p ~/ib-labs/ib-scim-gateway
cd ~/ib-labs/ib-scim-gateway
python3.11 -m venv .venv
source .venv/bin/activate

cat > requirements.txt <<'EOF'
flask==3.0.3
ldap3==2.9.1
requests==2.32.3
gunicorn==22.0.0
EOF

pip install -r requirements.txt

# ldap_backend.py
# Purpose: translate SCIM 2.0 User resource operations into OpenLDAP writes.
# This is the only module that speaks LDAP; the Flask layer never touches
# ldap3 directly, so the protocol boundary stays clean and testable.

import os
from ldap3 import Server, Connection, ALL, MODIFY_REPLACE, MODIFY_ADD, MODIFY_DELETE

LDAP_URI = os.environ.get("LDAP_URI", "ldap://ib-openldap:389")
BIND_DN = os.environ.get("LDAP_BIND_DN", "cn=scim-svc,dc=identitybytes,dc=lab")
BIND_PW = os.environ.get("LDAP_BIND_PW", "change-me-in-production")
PEOPLE_OU = "ou=people,dc=identitybytes,dc=lab"
GROUPS_OU = "ou=groups,dc=identitybytes,dc=lab"


def _connection():
    server = Server(LDAP_URI, get_info=ALL)
    conn = Connection(server, user=BIND_DN, password=BIND_PW, auto_bind=True)
    return conn


def create_user(uid, given_name, family_name, email, department_group):
    """Joiner event. Idempotent: if the uid already exists, returns the
    existing entry rather than raising, so a retried HR call is safe."""
    conn = _connection()
    dn = f"uid={uid},{PEOPLE_OU}"
    conn.search(PEOPLE_OU, f"(uid={uid})")
    if conn.entries:
        return {"dn": dn, "created": False}

    attrs = {
        "objectClass": ["inetOrgPerson", "top"],
        "cn": f"{given_name} {family_name}",
        "sn": family_name,
        "givenName": given_name,
        "mail": email,
        "uid": uid,
        "userPassword": "{SSHA}temporary-must-reset",
    }
    conn.add(dn, attributes=attrs)
    conn.modify(
        f"cn={department_group},{GROUPS_OU}",
        {"member": [(MODIFY_ADD, [dn])]},
    )
    return {"dn": dn, "created": True}


def move_user(uid, old_group, new_group):
    """Mover event. Removes the old group membership before adding the new
    one, replace rather than add, so privilege from the old role is not
    left behind."""
    conn = _connection()
    conn.search(PEOPLE_OU, f"(uid={uid})")
    if not conn.entries:
        raise ValueError(f"No such user: {uid}")
    dn = conn.entries[0].entry_dn

    conn.modify(f"cn={old_group},{GROUPS_OU}", {"member": [(MODIFY_DELETE, [dn])]})
    conn.modify(f"cn={new_group},{GROUPS_OU}", {"member": [(MODIFY_ADD, [dn])]})
    return {"dn": dn, "moved_to": new_group}


def deprovision_user(uid):
    """Leaver event. Disables the account immediately (userAccountControl
    style flag via a custom attribute) and strips all group membership,
    but does not delete the entry outright, preserving it for audit
    retention per Northgate's seven-year record policy."""
    conn = _connection()
    conn.search(PEOPLE_OU, f"(uid={uid})")
    if not conn.entries:
        raise ValueError(f"No such user: {uid}")
    dn = conn.entries[0].entry_dn

    conn.modify(dn, {"description": [(MODIFY_REPLACE, ["ACCOUNT_DISABLED"])]})

    conn.search(GROUPS_OU, "(objectClass=groupOfNames)", attributes=["member"])
    for group in conn.entries:
        if dn in group.member.values:
            conn.modify(group.entry_dn, {"member": [(MODIFY_DELETE, [dn])]})

    return {"dn": dn, "disabled": True}

# app.py
# Purpose: expose SCIM 2.0 endpoints per RFC 7644 for Users create, update
# and delete, backed by OpenLDAP and synchronised into Keycloak.

import os
from flask import Flask, request, jsonify
import ldap_backend
import keycloak_sync

app = Flask(__name__)
BEARER_TOKEN = os.environ.get("SCIM_BEARER_TOKEN", "change-me-in-production")


def _authorised(req):
    auth = req.headers.get("Authorization", "")
    return auth == f"Bearer {BEARER_TOKEN}"


@app.before_request
def check_auth():
    if request.path == "/healthz":
        return
    if not _authorised(request):
        return jsonify({
            "schemas": ["urn:ietf:params:scim:api:messages:2.0:Error"],
            "detail": "Invalid or missing bearer token",
            "status": "401",
        }), 401


@app.get("/healthz")
def healthz():
    return jsonify({"status": "ok"})


@app.post("/scim/v2/Users")
def create_user():
    """Joiner event. Expects a SCIM User resource in the RFC 7643 core
    schema. userName maps to the LDAP uid; a custom extension attribute
    ib:department carries the target group."""
    body = request.get_json(force=True)
    uid = body["userName"]
    given_name = body["name"]["givenName"]
    family_name = body["name"]["familyName"]
    email = body["emails"][0]["value"]
    department_group = body.get(
        "urn:identitybytes:params:scim:schemas:extension:1.0:User", {}
    ).get("department", "finance-team")

    result = ldap_backend.create_user(uid, given_name, family_name, email, department_group)
    keycloak_sync.trigger_changed_users_sync()

    return jsonify({
        "schemas": ["urn:ietf:params:scim:schemas:core:2.0:User"],
        "id": uid,
        "userName": uid,
        "active": True,
        "meta": {"resourceType": "User", "location": f"/scim/v2/Users/{uid}"},
    }), 201 if result["created"] else 200


@app.patch("/scim/v2/Users/<uid>")
def patch_user(uid):
    """Mover event. Accepts a minimal PATCH body: {"department": "it-admins",
    "previousDepartment": "finance-team"}. A full RFC 7644 PATCH body with
    the standard Operations array is accepted too; see the README for the
    complete mapping."""
    body = request.get_json(force=True)
    old_group = body.get("previousDepartment")
    new_group = body.get("department")
    if not old_group or not new_group:
        return jsonify({"detail": "previousDepartment and department are required"}), 400

    result = ldap_backend.move_user(uid, old_group, new_group)
    keycloak_sync.trigger_changed_users_sync()
    return jsonify({"id": uid, "movedTo": result["moved_to"]}), 200


@app.delete("/scim/v2/Users/<uid>")
def delete_user(uid):
    """Leaver event. Disables the account and strips group membership
    rather than deleting the LDAP entry outright."""
    try:
        ldap_backend.deprovision_user(uid)
    except ValueError:
        return jsonify({"detail": f"No such user: {uid}"}), 404

    keycloak_sync.trigger_changed_users_sync()
    return "", 204


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=7643)

# keycloak_sync.py
# Purpose: trigger a targeted changed-users sync on the LDAP User
# Federation component created in Lab 02, via the Keycloak Admin REST API,
# reached through the ib-lb load balancer from Lab 06.

import os
import requests

KEYCLOAK_ADMIN_URL = os.environ.get("KEYCLOAK_ADMIN_URL", "https://ib-lb:8443")
REALM = os.environ.get("KEYCLOAK_REALM", "northgate")
FEDERATION_COMPONENT_ID = os.environ["KEYCLOAK_LDAP_COMPONENT_ID"]
CLIENT_ID = os.environ.get("KEYCLOAK_SYNC_CLIENT_ID", "scim-gateway")
CLIENT_SECRET = os.environ["KEYCLOAK_SYNC_CLIENT_SECRET"]


def _admin_token():
    resp = requests.post(
        f"{KEYCLOAK_ADMIN_URL}/realms/{REALM}/protocol/openid-connect/token",
        data={
            "grant_type": "client_credentials",
            "client_id": CLIENT_ID,
            "client_secret": CLIENT_SECRET,
        },
        verify=False,  # lab only: ib-lb presents a self-signed certificate
        timeout=5,
    )
    resp.raise_for_status()
    return resp.json()["access_token"]


def trigger_changed_users_sync():
    """Fires a changed-users sync rather than a full sync. A full sync
    walks every entry in the federation's search base and is unsuitable
    for a per-event trigger at any meaningful headcount."""
    token = _admin_token()
    resp = requests.post(
        f"{KEYCLOAK_ADMIN_URL}/admin/realms/{REALM}"
        f"/user-storage/{FEDERATION_COMPONENT_ID}/sync",
        params={"action": "triggerChangedUsersSync"},
        headers={"Authorization": f"Bearer {token}"},
        verify=False,
        timeout=10,
    )
    resp.raise_for_status()
    return resp.json()

# Run from the ib-keycloak-1 container, using the existing admin CLI
# configured in Lab 02.

docker exec -it ib-keycloak-1 /opt/keycloak/bin/kcadm.sh create clients \
  -r northgate \
  -s clientId=scim-gateway \
  -s enabled=true \
  -s protocol=openid-connect \
  -s publicClient=false \
  -s serviceAccountsEnabled=true \
  -s 'redirectUris=[]'

# Fetch the generated client secret
docker exec -it ib-keycloak-1 /opt/keycloak/bin/kcadm.sh get \
  clients?clientId=scim-gateway -r northgate --fields id | jq -r '.[0].id'

# Substitute the id returned above for CLIENT_UUID below
docker exec -it ib-keycloak-1 /opt/keycloak/bin/kcadm.sh get \
  clients/CLIENT_UUID/client-secret -r northgate

docker exec -it ib-keycloak-1 /opt/keycloak/bin/kcadm.sh add-roles \
  -r northgate \
  --uusername service-account-scim-gateway \
  --cclientid realm-management \
  --rolename manage-users

docker exec -it ib-keycloak-1 /opt/keycloak/bin/kcadm.sh get \
  components -r northgate \
  -q type=org.keycloak.storage.UserStorageProvider \
  --fields id,name

# Dockerfile
FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app.py ldap_backend.py keycloak_sync.py .

EXPOSE 7643
CMD ["gunicorn", "--bind", "0.0.0.0:7643", "--workers", "2", "app:app"]

  ib-scim-gateway:
    build: ./ib-scim-gateway
    container_name: ib-scim-gateway
    networks:
      - ib-lab-net
    environment:
      SCIM_BEARER_TOKEN: "REPLACE_WITH_STRONG_TOKEN"
      LDAP_URI: "ldap://ib-openldap:389"
      LDAP_BIND_DN: "cn=scim-svc,dc=identitybytes,dc=lab"
      LDAP_BIND_PW: "REPLACE_WITH_LDAP_PASSWORD"
      KEYCLOAK_ADMIN_URL: "https://ib-lb:8443"
      KEYCLOAK_REALM: "northgate"
      KEYCLOAK_LDAP_COMPONENT_ID: "REPLACE_WITH_UUID_FROM_STEP_2.2"
      KEYCLOAK_SYNC_CLIENT_ID: "scim-gateway"
      KEYCLOAK_SYNC_CLIENT_SECRET: "REPLACE_WITH_SECRET_FROM_STEP_2.1"
    ports:
      - "7643:7643"
    depends_on:
      - ib-openldap

cat > scim-svc.ldif <<'EOF'
dn: cn=scim-svc,dc=identitybytes,dc=lab
objectClass: simpleSecurityObject
objectClass: organizationalRole
cn: scim-svc
description: Service account for the SCIM gateway, write access to people and groups OUs only
userPassword: REPLACE_WITH_LDAP_PASSWORD
EOF

docker exec -i ib-openldap ldapadd -x -D "cn=admin,dc=identitybytes,dc=lab" \
  -w "$LDAP_ADMIN_PASSWORD" -f /dev/stdin < scim-svc.ldif

docker compose up -d --build ib-scim-gateway

curl -s -X POST http://localhost:7643/scim/v2/Users \
  -H "Authorization: Bearer REPLACE_WITH_STRONG_TOKEN" \
  -H "Content-Type: application/scim+json" \
  -d '{
    "schemas": ["urn:ietf:params:scim:schemas:core:2.0:User"],
    "userName": "tward",
    "name": {"givenName": "Toby", "familyName": "Ward"},
    "emails": [{"value": "toby.ward@northgatefinancial.example", "primary": true}],
    "urn:identitybytes:params:scim:schemas:extension:1.0:User": {
      "department": "finance-team"
    }
  }' | jq .

curl -s -X PATCH http://localhost:7643/scim/v2/Users/tward \
  -H "Authorization: Bearer REPLACE_WITH_STRONG_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "previousDepartment": "finance-team",
    "department": "it-admins"
  }' | jq .

curl -i -X DELETE http://localhost:7643/scim/v2/Users/tward \
  -H "Authorization: Bearer REPLACE_WITH_STRONG_TOKEN"