How to Sync Employee Data from an HR System to OpenLDAP

#!/bin/bash

DB_NAME="hr"
DB_USER="hr_admin"

echo "Creating PostgreSQL role $DB_USER..."
sudo -u postgres createuser --no-password --login "$DB_USER"
echo "Creating database $DB_NAME with owner $DB_USER..."
sudo -u postgres createdb --owner="$DB_USER" "$DB_NAME"

echo "Creating table and trigger..."
sudo -u postgres psql -d "$DB_NAME" <<'EOF'
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    username TEXT UNIQUE NOT NULL,
    first_name TEXT NOT NULL,
    last_name TEXT NOT NULL,
    display_name TEXT,
    email TEXT UNIQUE NOT NULL,
    internal_email TEXT UNIQUE,
    department TEXT,
    position TEXT,
    status TEXT NOT NULL DEFAULT 'active',
    hire_date DATE NOT NULL,
    terminated_at DATE,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE OR REPLACE FUNCTION update_updated_at_column()
RETURNS TRIGGER AS $$
BEGIN
   NEW.updated_at = CURRENT_TIMESTAMP;
   RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER set_updated_at
BEFORE UPDATE ON users
FOR EACH ROW
EXECUTE PROCEDURE update_updated_at_column();
EOF

echo "Inserting sample users..."
sudo -u postgres psql -d "$DB_NAME" <<'EOF'
INSERT INTO users (
  username, first_name, last_name, display_name,
  email, internal_email, department, position,
  status, hire_date
) VALUES
  ('jdoe', 'John', 'Doe', 'John Doe',
   'john.doe@gmail.com', 'john.doe@company.com', 'IT', 'Developer',
   'active', '2023-05-01'),

  ('asmith', 'Alice', 'Smith', 'Alice Smith',
   'alice.smith@yahoo.com', 'alice.smith@company.com', 'HR', 'HR Manager',
   'active', '2022-11-15'),

  ('bchan', 'Bob', 'Chan', 'Bob Chan',
   'bob.chan@outlook.com', 'bob.chan@company.com', 'Finance', 'Accountant',
   'active', '2024-02-01')
ON CONFLICT (username) DO NOTHING;
EOF

echo "HR DB ready with sample data."

chmod +x create_hr_db.sh
./create_hr_db.sh

pip install psycopg[binary] ldap3

from ldap3 import Server, Connection, MODIFY_REPLACE
import psycopg


def fetch_all_users_from_pg(conninfo: str) -> list[tuple]:
    with psycopg.connect(conninfo) as conn:
        with conn.cursor() as cur:
            cur.execute("""
                SELECT id, username, first_name, last_name, display_name,
                       email, internal_email, status
                FROM users
            """)
            return cur.fetchall()


def sync_user_to_ldap(conn: Connection, base_dn: str, user: tuple):
    (user_id, username, first_name, last_name, display_name,
     email, internal_email, status) = user

    full_name = display_name or f"{first_name} {last_name}"
    primary_email = internal_email or email
    dn = f"uid={username},{base_dn}"

    conn.search(base_dn, f"(uid={username})", attributes=["employeeType"])
    attributes = {
        "uid": username,
        "cn": full_name,
        "displayName": full_name,
        "givenName": first_name,
        "sn": last_name,
        "mail": primary_email,
        "employeeNumber": str(user_id),
        "employeeType": status
    }

    if conn.entries:
        current_status = conn.entries[0].employeeType.value
        if status == "active" and current_status != "active":
            print(f"[REACTIVATE] {username}")
        else:
            print(f"[UPDATE] {username}")
        conn.modify(dn, {k: [(MODIFY_REPLACE, [v])] for k, v in attributes.items()})
    elif status == 'active':
        print(f"[ADD] {username}")
        conn.add(dn, ["inetOrgPerson"], attributes)


def disable_stale_ldap_users(conn: Connection, base_dn: str, active_uids: set[str]):
    conn.search(base_dn, "(objectClass=inetOrgPerson)", attributes=["uid"])
    for entry in conn.entries:
        uid = entry.uid.value
        if uid not in active_uids:
            dn = f"uid={uid},{base_dn}"
            print(f"[DISABLE] {uid}")
            conn.modify(
                dn,
                {
                    "employeeType": [(MODIFY_REPLACE, ["inactive"])]
                }
            )


def sync_all():
    conninfo = "host=hr.maksonlee.com dbname=hr user=hr_admin password=supersecure123"
    ldap_host = "ldaps://ldap.maksonlee.com"
    ldap_bind_dn = "cn=admin,dc=maksonlee,dc=com"
    ldap_password = "adminpassword"
    base_dn = "ou=people,dc=maksonlee,dc=com"

    all_users = fetch_all_users_from_pg(conninfo)
    active_users = [u for u in all_users if u[-1] == 'active']
    active_uids = {u[1] for u in active_users}

    server = Server(ldap_host)
    conn = Connection(server, user=ldap_bind_dn, password=ldap_password, auto_bind=True)

    for user in all_users:
        sync_user_to_ldap(conn, base_dn, user)

    disable_stale_ldap_users(conn, base_dn, active_uids)
    conn.unbind()


if __name__ == "__main__":
    sync_all()

dn: uid=jdoe,ou=people,dc=maksonlee,dc=com
objectClass: inetOrgPerson
uid: jdoe
cn: John Doe
sn: Doe
givenName: John
displayName: John Doe
mail: john.doe@company.com
employeeNumber: 1
employeeType: active