mgmt/scripts/audit-kc-groups.py

#!/usr/bin/env python3
"""Audit Keycloak project groups against the expected naming and membership.

Read-only. For every project in the local database this script checks:

  1. The top-level Keycloak group exists.
  2. All standard child groups exist ({KEY}-{app}-{read|write|admin} for
     mgmtsuite).
  3. No child groups use legacy/unexpected naming — in particular the old
     `{KEY}-mgmt-*` convention, which the permission model does not read
     (users in those groups silently have no app permissions).
  4. No users are members of the top-level group without also being in at
     least one `{KEY}-mgmtsuite-*` child group (i.e. project membership
     that grants no permission in this app).

It also reports top-level Keycloak groups that have no matching project in
the database.

Usage:
    python scripts/audit-kc-groups.py             # audit all projects
    python scripts/audit-kc-groups.py KEY1 KEY2   # audit specific projects

Exits 0 when clean, 1 when findings were reported (cron-friendly).
"""

import argparse
import json
import re
import subprocess
import sys
import urllib.parse
import urllib.request
from urllib.error import HTTPError

# --- CONFIG ---
# These should match scripts/delete-projects.py
DB_URL = "postgresql://user:pass@host/db"
KC_URL = "https://keycloak.osa.af.mil"
KC_REALM = ""
KC_CLIENT_ID = ""
KC_CLIENT_SECRET = ""
# --------------

# Must match backend/app/constants.py STANDARD_APPS
STANDARD_APPS = {
    "mgmtsuite": ["read", "write", "admin"],
}

# Old child-group convention replaced by `mgmtsuite`; flagged loudly because
# memberships in these groups grant nothing.
LEGACY_APP_NAMES = ("mgmt",)

# Top-level KC groups that intentionally have no DB project. They are still
# audited for legacy child naming (VELA-mgmtsuite-admin gates global admin),
# but not for missing standard children or orphan status.
KC_ONLY_GROUPS = ("VELA",)

PAGE_SIZE = 100


def run_query(query, single_val=False):
    """Executes a PostgreSQL query via psql and returns results cleanly."""
    cmd = ["psql", DB_URL, "-t", "-A", "-c", query]
    try:
        res = subprocess.check_output(cmd, stderr=subprocess.DEVNULL).decode().strip()
        if single_val:
            return res
        return [line for line in res.splitlines() if line]
    except subprocess.CalledProcessError:
        return "" if single_val else []


def req(method, url, data=None, headers=None):
    """Standard HTTP request handling."""
    req_obj = urllib.request.Request(
        url, data=data, method=method, headers=headers or {}
    )
    try:
        with urllib.request.urlopen(req_obj) as r:
            body = r.read().decode("utf-8")
            ct = r.getheader("Content-Type") or ""
            return json.loads(body) if "application/json" in ct else body
    except HTTPError as e:
        return e


def get_token():
    url = f"{KC_URL}/realms/{KC_REALM}/protocol/openid-connect/token"
    data = f"grant_type=client_credentials&client_id={KC_CLIENT_ID}&client_secret={KC_CLIENT_SECRET}".encode()
    with urllib.request.urlopen(urllib.request.Request(url, data=data)) as r:
        return json.loads(r.read().decode())["access_token"]


def paginate(url_base, headers, params=None):
    """Yield items from a paginated KC admin endpoint."""
    first = 0
    while True:
        query = dict(params or {})
        query.update({"first": first, "max": PAGE_SIZE})
        url = f"{url_base}?{urllib.parse.urlencode(query)}"
        page = req("GET", url, headers=headers)
        if not isinstance(page, list):
            print(f"  WARN: failed to fetch {url_base}: {page}")
            return
        yield from page
        if len(page) < PAGE_SIZE:
            return
        first += PAGE_SIZE


def get_top_level_groups(headers):
    """Return {name: id} for all top-level groups in the realm."""
    url = f"{KC_URL}/admin/realms/{KC_REALM}/groups"
    return {g["name"]: g["id"] for g in paginate(url, headers)}


def get_children(group_id, headers):
    """Return [{name, id}, ...] child groups of a group."""
    url = f"{KC_URL}/admin/realms/{KC_REALM}/groups/{group_id}/children"
    return [{"name": g["name"], "id": g["id"]} for g in paginate(url, headers)]


def get_members(group_id, headers):
    """Return {username: user_id} members of a group."""
    url = f"{KC_URL}/admin/realms/{KC_REALM}/groups/{group_id}/members"
    return {
        u.get("username", u["id"]): u["id"]
        for u in paginate(url, headers, params={"briefRepresentation": "true"})
    }


def expected_children(project_key):
    return {
        f"{project_key}-{app}-{perm}"
        for app, perms in STANDARD_APPS.items()
        for perm in perms
    }


def classify_child(project_key, name):
    """Return 'standard', 'legacy', or 'custom' for a child group name."""
    if name in expected_children(project_key):
        return "standard"
    for legacy in LEGACY_APP_NAMES:
        if re.fullmatch(rf"{re.escape(project_key)}-{legacy}-(read|write|admin)", name):
            return "legacy"
    return "custom"


def audit_project(project_key, group_map, headers, kc_only=False):
    """Audit one project. Returns a list of finding strings."""
    findings = []

    group_id = group_map.get(project_key)
    if not group_id:
        return [f"top-level group '{project_key}' is MISSING in Keycloak"]

    children = get_children(group_id, headers)
    child_names = {c["name"] for c in children}

    # Missing standard children (KC-only groups like VELA aren't created by
    # the app, so their expected children may legitimately differ)
    if not kc_only:
        for missing in sorted(expected_children(project_key) - child_names):
            findings.append(f"missing standard child group: {missing}")

    # Legacy / unexpected child names
    legacy_groups = []
    for child in sorted(children, key=lambda c: c["name"]):
        kind = classify_child(project_key, child["name"])
        if kind == "legacy":
            members = get_members(child["id"], headers)
            legacy_groups.append((child["name"], members))

    for name, members in legacy_groups:
        detail = f" (members: {', '.join(sorted(members))})" if members else " (empty)"
        findings.append(
            f"LEGACY child group '{name}' — memberships here grant no permissions{detail}"
        )

    # Top-level members with no mgmtsuite child membership
    top_members = get_members(group_id, headers)
    if top_members:
        mgmtsuite_members = set()
        for child in children:
            if child["name"].startswith(f"{project_key}-mgmtsuite-"):
                mgmtsuite_members.update(get_members(child["id"], headers))
        for username in sorted(set(top_members) - mgmtsuite_members):
            findings.append(
                f"user '{username}' is in '{project_key}' but in no "
                f"{project_key}-mgmtsuite-* child group (no app permissions)"
            )

    return findings


def main():
    parser = argparse.ArgumentParser(
        description="Audit Keycloak project group naming and membership."
    )
    parser.add_argument(
        "keys", nargs="*",
        help="Project keys to audit (default: all projects in the database)",
    )
    args = parser.parse_args()

    print("Fetching projects from database...")
    db_projects = run_query("SELECT key FROM projects ORDER BY key;")
    if not db_projects:
        print("No projects found in database. Check DB_URL.")
        sys.exit(1)
    print(f"Found {len(db_projects)} projects in DB.")

    if args.keys:
        unknown = [k for k in args.keys if k not in db_projects]
        if unknown:
            print(f"Unknown project keys (not in DB): {', '.join(unknown)}")
            sys.exit(1)
        targets = args.keys
    else:
        targets = db_projects

    try:
        token = get_token()
        headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
        }
    except Exception as e:
        print(f"Authentication Failed with Keycloak Server: {e}")
        sys.exit(1)

    print("Fetching top-level groups from Keycloak...")
    group_map = get_top_level_groups(headers)

    total_findings = 0
    print("\n--- Audit Results ---")
    for key in targets:
        findings = audit_project(key, group_map, headers)
        if findings:
            total_findings += len(findings)
            print(f"\n{key}:")
            for f in findings:
                print(f"  - {f}")

    # KC-only groups (e.g. VELA) and orphan detection — only meaningful when
    # auditing everything.
    if not args.keys:
        for name in KC_ONLY_GROUPS:
            if name in group_map:
                findings = audit_project(name, group_map, headers, kc_only=True)
                if findings:
                    total_findings += len(findings)
                    print(f"\n{name} (KC-only group):")
                    for f in findings:
                        print(f"  - {f}")

        orphans = sorted(set(group_map) - set(db_projects) - set(KC_ONLY_GROUPS))
        if orphans:
            total_findings += len(orphans)
            print("\nTop-level Keycloak groups with no matching DB project:")
            for name in orphans:
                print(f"  - {name}")

    print()
    if total_findings:
        print(f"Found {total_findings} finding(s).")
        sys.exit(1)
    print("All audited projects look clean.")
    sys.exit(0)


if __name__ == "__main__":
    main()