sicabe-core/deploy/lib/authz.py

#!/usr/bin/env python3
# =============================================================================
# SiCABE — lib/authz.py
# -----------------------------------------------------------------------------
# Model authorization (config/opa/authz.yml) <-> data ACL OPA (config/opa/data.json).
#
# SUMBER KEBENARAN = authz.yml. data.json ditandai `_generated` dan diperiksa
# lewat `render --check`. Fungsi model murni (pure) agar mudah diuji; hanya
# pembacaan/penulisan berkas yang menyentuh sisi luar.
#
# Pemakaian (biasanya via pembungkus scripts/sicabe-authz):
#   python3 lib/authz.py fmt|render|check|list|matrix|show|grant|role|user|catalog
# Env: SICABE_AUTHZ_DRY_RUN=1  -> jangan tulis apa pun, cetak rencana.
# =============================================================================
from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import sys
import tempfile
from pathlib import Path

import yaml

import authz_ldap as L

CAPS = ("select", "write", "ddl")
DEPLOY_DIR = Path(__file__).resolve().parent.parent
DEFAULT_MODEL = DEPLOY_DIR / "config" / "opa" / "authz.yml"
DEFAULT_OUT = DEPLOY_DIR / "config" / "opa" / "data.json"
TRINO_CATALOG_DIR = DEPLOY_DIR / "config" / "trino" / "catalog"
DRY = os.environ.get("SICABE_AUTHZ_DRY_RUN") == "1"

HEADER = """\
# =============================================================================
# SiCABE — Model Authorization (SUMBER KEBENARAN) — OPA / Trino
# -----------------------------------------------------------------------------
# Berkas ini SATU-SATUNYA sumber kebenaran ACL. JANGAN mengedit
# config/opa/data.json manual — itu HASIL RENDER:
#     ./scripts/sicabe-authz render        # authz.yml -> data.json
#     ./scripts/sicabe-authz apply         # render + reload OPA
#     ./scripts/sicabe-authz fmt           # rapikan format berkas ini
# Policy Rego membaca data.authz.catalogs / data.authz.write_schemas — nama
# catalog/schema TIDAK di-hardcode di Rego. Ganti nama catalog = ubah `catalogs:`
# di sini -> render -> reload. Dokumentasi: docs/authorization-cli.md
# =============================================================================
"""


class ModelError(Exception):
    """Model authz.yml tidak valid / operasi ditolak."""


# ── Muat & validasi ─────────────────────────────────────────────────────────

def load_model(path: Path) -> dict:
    try:
        with open(path, encoding="utf-8") as fh:
            return yaml.safe_load(fh) or {}
    except FileNotFoundError:
        raise ModelError(f"berkas model tidak ditemukan: {path}")
    except yaml.YAMLError as exc:
        raise ModelError(f"YAML tidak valid di {path}: {exc}")


def _as_list(value, what: str) -> list:
    if not isinstance(value, list):
        raise ModelError(f"{what} harus berupa daftar")
    return value


def _table_key(table: str) -> str:
    parts = str(table).split(".", 1)
    if len(parts) != 2 or not parts[0] or not parts[1]:
        raise ModelError(f"table harus 'schema.nama', dapat: {table!r}")
    return f"{parts[0]}_{parts[1]}"


def _grant(grant: dict) -> dict:
    if not isinstance(grant, dict) or "table" not in grant:
        raise ModelError(f"grant tidak valid: {grant!r}")
    caps = _as_list(grant.get("caps", []), f"caps untuk {grant['table']}")
    unknown = [c for c in caps if c not in CAPS]
    if unknown:
        raise ModelError(f"caps tak dikenal {unknown} pada {grant['table']}")
    columns = grant.get("columns", "*")
    if columns != "*" and not isinstance(columns, list):
        raise ModelError(f"columns harus '*' atau daftar, pada {grant['table']}")
    return {"table": grant["table"], "caps": list(caps), "columns": columns}


def validate(model: dict) -> None:
    if not _as_list(model.get("catalogs", []), "catalogs"):
        raise ModelError("catalogs tidak boleh kosong")
    _as_list(model.get("schemas", []), "schemas")
    roles = model.get("roles")
    if not isinstance(roles, dict) or not roles:
        raise ModelError("roles harus berupa objek tak kosong")
    for name, spec in roles.items():
        if spec is not None and not isinstance(spec, dict):
            raise ModelError(f"role {name!r} harus objek")
        for grant in (spec or {}).get("grants", []):
            _grant(grant)
    users = model.get("users", {})
    if not isinstance(users, dict):
        raise ModelError("users harus berupa objek")
    for user, role in users.items():
        if role not in roles:
            raise ModelError(f"user {user!r} menunjuk role tak ada: {role!r}")


# ── Render (untuk OPA) ──────────────────────────────────────────────────────

def _render_role(spec: dict) -> dict:
    spec = spec or {}
    if spec.get("superuser"):
        return {"superuser": True}
    tables: dict = {}
    for grant in spec.get("grants", []):
        key = _table_key(grant["table"])
        out: dict = {}
        caps = grant.get("caps", [])
        for cap in CAPS:
            if cap in caps:
                out[cap] = True
        out["columns"] = grant.get("columns", "*")
        tables[key] = out
    return {"tables": tables} if tables else {}


def render(model: dict, digest: str) -> dict:
    return {
        "authz": {"catalogs": list(model.get("catalogs", [])),
                  "write_schemas": list(model.get("schemas", []))},
        "roles": {n: _render_role(s) for n, s in model.get("roles", {}).items()},
        "users": dict(model.get("users", {})),
        "_generated": {"from": DEFAULT_MODEL.name, "sha256": digest},
    }


def digest_of(path: Path) -> str:
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()


def build(model_path: Path) -> dict:
    model = load_model(model_path)
    validate(model)
    return render(model, digest_of(model_path))


# ── Serialisasi berkas ──────────────────────────────────────────────────────

def dump_json(obj: dict) -> str:
    return json.dumps(obj, indent=2, ensure_ascii=False) + "\n"


def dump_model(model: dict) -> str:
    ordered = {k: model[k] for k in ("version", "catalogs", "schemas", "roles", "users") if k in model}
    body = yaml.safe_dump(ordered, default_flow_style=None, sort_keys=False,
                          allow_unicode=True, width=100)
    return HEADER + body


def write_atomic(path: Path, text: str) -> None:
    path = Path(path)
    fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=".authz-", suffix=".tmp")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as fh:
            fh.write(text)
        os.replace(tmp, path)
    except BaseException:
        if os.path.exists(tmp):
            os.unlink(tmp)
        raise


def read_json(path: Path) -> dict:
    try:
        with open(path, encoding="utf-8") as fh:
            return json.load(fh)
    except FileNotFoundError:
        return {}
    except json.JSONDecodeError as exc:
        raise ModelError(f"data.json tidak valid: {exc}")


# ── Mutator model (murni: kembalikan model baru) ────────────────────────────

def get_grant(spec: dict, table: str):
    for grant in (spec or {}).get("grants", []):
        if grant["table"] == table:
            return grant
    return None


def set_grant(model: dict, role: str, table: str, caps: list, columns, replace: bool) -> str:
    roles = model.setdefault("roles", {})
    if role not in roles:
        raise ModelError(f"role {role!r} belum ada — buat dulu: role add {role}")
    _table_key(table)
    for c in caps:
        if c not in CAPS:
            raise ModelError(f"cap tak dikenal: {c} (pilihan: {', '.join(CAPS)})")
    spec = roles[role]
    if spec is None:                       # role kosong -> jadikan objek di tempat
        spec = {}
        roles[role] = spec
    if spec.get("superuser"):
        raise ModelError(f"role {role!r} superuser — tak perlu grant per-tabel")
    grants = spec.setdefault("grants", [])
    existing = get_grant(spec, table)
    if existing is None:
        grants.append({"table": table, "caps": list(dict.fromkeys(caps)),
                       "columns": columns if columns is not None else "*"})
    elif replace:
        existing["caps"] = list(dict.fromkeys(caps))
        existing["columns"] = columns if columns is not None else existing.get("columns", "*")
    else:
        existing["caps"] = list(dict.fromkeys(list(existing.get("caps", [])) + list(caps)))
        if columns is not None:
            existing["columns"] = columns
    return ("set" if replace else "set/merge") + f" grant {role} -> {table}"


def remove_grant(model: dict, role: str, table: str, caps: list | None) -> str:
    spec = (model.get("roles", {}) or {}).get(role)
    if not spec or not spec.get("grants"):
        raise ModelError(f"role {role!r} tak punya grant")
    existing = get_grant(spec, table)
    if existing is None:
        raise ModelError(f"grant {role} -> {table} tidak ada")
    if not caps:
        spec["grants"] = [g for g in spec["grants"] if g["table"] != table]
        return f"hapus grant {role} -> {table}"
    left = [c for c in existing.get("caps", []) if c not in caps]
    if left:
        existing["caps"] = left
    else:
        spec["grants"] = [g for g in spec["grants"] if g["table"] != table]
    return f"hapus cap {','.join(caps)} pada {role} -> {table}"


def roles_referencing(model: dict, role: str) -> list:
    return [u for u, r in model.get("users", {}).items() if r == role]


# ── Query ───────────────────────────────────────────────────────────────────

def all_tables(model: dict) -> list:
    keys = set()
    for spec in (model.get("roles", {}) or {}).values():
        for grant in (spec or {}).get("grants", []) or []:
            keys.add(_table_key(grant["table"]))
    return sorted(keys)


def caps_letters(caps: list) -> str:
    return "".join(c[0] for c in CAPS if c in caps) or "-"


# ── Perintah ────────────────────────────────────────────────────────────────

def _say(msg: str) -> None:
    print(f"[DRY-RUN] {msg}" if DRY else f"{msg}")


def cmd_fmt(model_path: Path) -> int:
    model = load_model(model_path)
    validate(model)
    if DRY:
        _say(f"rapikan {model_path}")
        return 0
    write_atomic(model_path, dump_model(model))
    print(f"fmt: {model_path}")
    return 0


def cmd_render(model_path: Path, out_path: Path, check: bool) -> int:
    expected = build(model_path)
    on_disk = read_json(out_path)
    if check:
        if on_disk == expected:
            print(f"SINKRON: {out_path} sesuai {model_path}")
            return 0
        print(f"TIDAK SINKRON: {out_path} berbeda dari {model_path}", file=sys.stderr)
        print("  jalankan: ./scripts/sicabe-authz render", file=sys.stderr)
        return 1
    if DRY:
        _say(f"render {model_path} -> {out_path}")
        return 0
    write_atomic(out_path, dump_json(expected))
    print(f"render: {model_path} -> {out_path}")
    return 0


def cmd_list(model_path: Path, what: str) -> int:
    model = load_model(model_path)
    if what == "catalogs":
        for c in model.get("catalogs", []):
            print(c)
    elif what == "roles":
        for name, spec in (model.get("roles", {}) or {}).items():
            tag = " (superuser)" if (spec or {}).get("superuser") else ""
            print(f"{name}{tag}")
    elif what == "users":
        for user, role in (model.get("users", {}) or {}).items():
            print(f"{user} -> {role}")
    elif what == "tables":
        for t in all_tables(model):
            print(t)
    return 0


def cmd_matrix(model_path: Path, schema: str | None) -> int:
    model = load_model(model_path)
    tables = all_tables(model)
    if schema:
        tables = [t for t in tables if t.startswith(f"{schema}_")]
    roles = list(model.get("roles", {}) or {})
    w = max([len(r) for r in roles] + [5])
    print("  Tabel:")
    for i, t in enumerate(tables):
        print(f"    [{i}] {t}")
    print()
    print("  " + " " * (w + 2) + " ".join(f"{i:>3}" for i in range(len(tables))))
    for r in roles:
        spec = (model.get("roles", {}) or {}).get(r) or {}
        if spec.get("superuser"):
            print(f"  {r:<{w}}  " + " ".join("ALL" for _ in tables))
            continue
        cells = []
        for t in tables:
            g = get_grant(spec, t.replace("_", ".", 1))
            cells.append(f"{caps_letters(g.get('caps', [])):>3}" if g else "  .")
        print(f"  {r:<{w}}  " + " ".join(cells))
    print("\n  Legenda caps: s=select w=write d=ddl | ALL=superuser | .=tak ada grant")
    return 0


def cmd_show(model_path: Path, what: str, name: str) -> int:
    model = load_model(model_path)
    if what == "role":
        spec = (model.get("roles", {}) or {}).get(name)
        if spec is None:
            raise ModelError(f"role {name!r} tidak ada")
        if spec.get("superuser"):
            print(f"{name}: SUPERUSER (semua operasi)")
        for g in spec.get("grants", []) or []:
            print(f"{name}: {g['table']}  caps=[{','.join(g.get('caps', []))}]  columns={g.get('columns')}")
        if not spec.get("superuser") and not spec.get("grants"):
            print(f"{name}: (tanpa grant)")
    elif what == "user":
        role = (model.get("users", {}) or {}).get(name)
        print(f"{name}: {role}" if role else f"{name}: (bukan service account; role dari grup LDAP)")
    elif what == "table":
        key = _table_key(name)
        found = False
        for r, spec in (model.get("roles", {}) or {}).items():
            g = get_grant(spec or {}, name)
            if g:
                found = True
                print(f"{name}: {r}  caps=[{','.join(g.get('caps', []))}]  columns={g.get('columns')}")
            elif (spec or {}).get("superuser"):
                print(f"{name}: {r}  (superuser)")
        if not found:
            print(f"{name}: (tak ada grant eksplisit; cek superuser/role lain)")
    return 0


def _split_csv(value: str | None) -> list:
    if not value:
        return []
    return [v.strip() for v in value.split(",") if v.strip()]


def _columns_arg(value: str | None):
    if value is None or value == "*":
        return "*"
    return _split_csv(value)


def save_model(model_path: Path, model: dict) -> None:
    validate(model)
    if DRY:
        _say(f"tulis {model_path}")
        return
    write_atomic(model_path, dump_model(model))


def cmd_grant(args, model_path: Path) -> int:
    model = load_model(model_path)
    if args.action == "add":
        msg = set_grant(model, args.role, args.table, _split_csv(args.caps), _columns_arg(args.columns), False)
    elif args.action == "set":
        msg = set_grant(model, args.role, args.table, _split_csv(args.caps), _columns_arg(args.columns), True)
    else:
        msg = remove_grant(model, args.role, args.table, _split_csv(getattr(args, "caps", None)) or None)
    _say(msg)
    save_model(model_path, model)
    return 0


def cmd_role(args, model_path: Path) -> int:
    model = load_model(model_path)
    roles = model.setdefault("roles", {})
    if args.action == "add":
        if args.role in roles:
            raise ModelError(f"role {args.role!r} sudah ada")
        roles[args.role] = {"superuser": True} if args.superuser else {}
        _say(f"tambah role {args.role}" + (" (superuser)" if args.superuser else ""))
    else:
        if args.role not in roles:
            raise ModelError(f"role {args.role!r} tidak ada")
        refs = roles_referencing(model, args.role)
        if refs and not args.force:
            raise ModelError(f"role {args.role!r} masih dipakai user: {', '.join(refs)} (pakai --force)")
        for u in refs:
            del model["users"][u]
        del roles[args.role]
        _say(f"hapus role {args.role}" + (f" (+ unassign {', '.join(refs)})" if refs else ""))
    save_model(model_path, model)
    return 0


def cmd_user(args, model_path: Path) -> int:
    model = load_model(model_path)
    users_ = model.setdefault("users", {})
    if args.action == "assign":
        if args.role not in model.get("roles", {}):
            raise ModelError(f"role {args.role!r} tidak ada")
        users_[args.user] = args.role
        _say(f"assign {args.user} -> {args.role}")
    else:
        if args.user not in users_:
            raise ModelError(f"service account {args.user!r} tidak terdaftar")
        del users_[args.user]
        _say(f"unassign {args.user}")
    save_model(model_path, model)
    return 0


def cmd_user_ldap(args) -> int:
    conf = L.config()
    if args.action == "list":
        return L.user_list(conf, args.service)
    if not args.user:
        raise ModelError(f"user {args.action}: butuh <uid>")
    if args.action == "add":
        return L.user_add(conf, args.user, args.cn, args.mail, args.password, args.group)
    return L.user_remove(conf, args.user)


def cmd_group(args, model_path: Path) -> int:
    conf = L.config()
    a = args.action
    if a == "list":
        return L.group_list(conf, args.members)
    if a == "check":
        return L.group_check(conf, model_path)
    if a == "sync":
        return L.group_sync(conf, model_path, args.apply)
    if not args.a:
        raise ModelError(f"group {a}: butuh <cn>")
    if a == "add":
        return L.group_add(conf, args.a, args.member)
    if a == "remove":
        return L.group_remove(conf, args.a)
    if not args.b:
        raise ModelError(f"group {a}: butuh dua argumen")
    if a == "rename":
        return L.group_rename(conf, model_path, args.a, args.b)
    return L.group_member(conf, args.a, args.b, a == "add-member")


CATALOG_SCAN_PATTERNS = ("deploy/scripts", "deploy/lib", "deploy/config", "docs")


def _catalog_regexes(old: str, schemas: list) -> list:
    """Pola BERTARGET: hanya rujukan yang benar-benar berarti nama catalog —
    bukan `/iceberg/` (prefix REST), `iceberg.catalog.*` (kunci konektor),
    `io.trino.plugin.iceberg`, `iceberg-data`, atau prosa."""
    o = re.escape(old)
    sch = "|".join(re.escape(s) for s in schemas) or "bronze|silver|gold"
    return [
        re.compile(rf"(?<![\w-]){o}\.(?:{sch})\b"),        # iceberg.bronze / iceberg.gold
        re.compile(rf"(?<![\w-]){o}\.properties\b"),         # iceberg.properties
        re.compile(rf"SEED_CATALOG\s*[=(:]\s*{o}\b"),        # SEED_CATALOG=iceberg
        re.compile(rf"catalog(?:_name)?\s*[:=]\s*{o}\b"),    # catalog: iceberg
        re.compile(rf"database\s*[:=]\s*{o}\b"),             # dbt database: iceberg
        re.compile(rf"""["']{o}["']\s*\)"""),                # getenv("X", "iceberg")
    ]


def _scan_catalog_refs(old: str, schemas: list) -> dict:
    regexes = _catalog_regexes(old, schemas)
    root = DEPLOY_DIR.parent
    skip = {DEFAULT_MODEL.resolve(), DEFAULT_OUT.resolve(),
            (DEPLOY_DIR / "lib" / "authz.py").resolve(),
            (DEPLOY_DIR / "scripts" / "sicabe-authz").resolve()}
    exts = {".sh", ".py", ".yml", ".yaml", ".json", ".md", ".properties", ".cfg", ".ini"}
    code, docs = [], []
    for base in CATALOG_SCAN_PATTERNS:
        top = root / base
        if not top.exists():
            continue
        for path in sorted(top.rglob("*")):
            if not path.is_file() or path.resolve() in skip or path.suffix not in exts:
                continue
            if "trino/catalog" in path.as_posix():
                continue
            bucket = docs if base == "docs" else code
            for n, line in enumerate(path.read_text(errors="replace").splitlines(), 1):
                if any(rx.search(line) for rx in regexes):
                    bucket.append(f"{path.relative_to(root)}:{n}: {line.strip()}")
    return {"code": code, "docs": docs}


def cmd_catalog(args, model_path: Path) -> int:
    model = load_model(model_path)
    if args.action == "list":
        for c in model.get("catalogs", []):
            print(c)
        return 0
    if args.action == "add":
        if args.new in model.get("catalogs", []):
            raise ModelError(f"catalog {args.new!r} sudah terdaftar")
        model.setdefault("catalogs", []).append(args.new)
        _say(f"tambah catalog {args.new}")
        save_model(model_path, model)
        return 0
    if args.action != "rename":
        raise ModelError("catalog: aksi harus list|add|rename")
    old, new = args.old, args.new
    if old not in model.get("catalogs", []):
        raise ModelError(f"catalog {old!r} tidak terdaftar di model")
    model["catalogs"] = [new if c == old else c for c in model["catalogs"]]
    src = TRINO_CATALOG_DIR / f"{old}.properties"
    dst = TRINO_CATALOG_DIR / f"{new}.properties"
    _say(f"catalog rename {old} -> {new}")
    print(f"  - model: catalogs: {old} -> {new}")
    if src.exists():
        print(f"  - berkas Trino: {src.relative_to(DEPLOY_DIR.parent)} -> {dst.name}")
        if not DRY:
            dst.write_text(src.read_text())
            src.unlink()
    else:
        print(f"  - berkas Trino {src.name} tidak ditemukan (dilewati)")
    save_model(model_path, model)
    if not DRY:
        cmd_render(model_path, DEFAULT_OUT, False)
    print("  - referensi KODE/KONFIGURASI (perlu ditinjau, TIDAK diubah otomatis):")
    hits = _scan_catalog_refs(old, model.get("schemas", []))
    for h in hits["code"] or ["(tidak ada)"]:
        print(f"      {h}")
    if hits["docs"]:
        print(f"  - referensi DOKUMENTASI ({len(hits['docs'])} baris, perlu diperbarui):")
        print("      (jalankan: grep -rn '" + old + "\\.' docs/ )")
    return 0


# ── Argparse ────────────────────────────────────────────────────────────────

def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(prog="sicabe-authz", description="Kelola authorization OPA (authz.yml -> data.json)")
    p.add_argument("--model", type=Path, default=DEFAULT_MODEL)
    p.add_argument("--out", type=Path, default=DEFAULT_OUT)
    sub = p.add_subparsers(dest="cmd", required=True)

    sub.add_parser("fmt")
    sub.add_parser("render")
    sub.add_parser("check")

    pl = sub.add_parser("list")
    pl.add_argument("what", choices=["catalogs", "roles", "users", "tables"])

    pm = sub.add_parser("matrix")
    pm.add_argument("--schema")

    ps = sub.add_parser("show")
    ps.add_argument("what", choices=["role", "user", "table"])
    ps.add_argument("name")

    pg = sub.add_parser("grant")
    pg.add_argument("action", choices=["add", "set", "remove"])
    pg.add_argument("role")
    pg.add_argument("table")
    pg.add_argument("--caps")
    pg.add_argument("--columns")

    pr = sub.add_parser("role")
    pr.add_argument("action", choices=["add", "remove"])
    pr.add_argument("role")
    pr.add_argument("--superuser", action="store_true")
    pr.add_argument("--force", action="store_true")

    pu = sub.add_parser("user")
    pu.add_argument("action", choices=["assign", "unassign", "list", "add", "remove"])
    pu.add_argument("user", nargs="?")
    pu.add_argument("role", nargs="?")
    pu.add_argument("--service", action="store_true", help="user list: tampilkan service account (model)")
    pu.add_argument("--cn")
    pu.add_argument("--mail")
    pu.add_argument("--password")
    pu.add_argument("--in", dest="group")

    pgr = sub.add_parser("group")
    pgr.add_argument("action", choices=["list", "add", "remove", "rename",
                                        "add-member", "remove-member", "check", "sync"])
    pgr.add_argument("a", nargs="?", help="cn (atau cn lama untuk rename)")
    pgr.add_argument("b", nargs="?", help="uid (atau cn baru)")
    pgr.add_argument("--members", action="store_true", help="list: tampilkan anggota")
    pgr.add_argument("--member", action="append", default=[], help="add: uid anggota (boleh berulang)")
    pgr.add_argument("--apply", action="store_true", help="sync: tulis perubahan")

    pc = sub.add_parser("catalog")
    pc.add_argument("action", choices=["list", "add", "rename"])
    pc.add_argument("old", nargs="?")
    pc.add_argument("new", nargs="?")
    return p


def dispatch(args) -> int:
    if args.cmd == "fmt":
        return cmd_fmt(args.model)
    if args.cmd == "render":
        return cmd_render(args.model, args.out, False)
    if args.cmd == "check":
        return cmd_render(args.model, args.out, True)
    if args.cmd == "list":
        return cmd_list(args.model, args.what)
    if args.cmd == "matrix":
        return cmd_matrix(args.model, args.schema)
    if args.cmd == "show":
        return cmd_show(args.model, args.what, args.name)
    if args.cmd == "grant":
        return cmd_grant(args, args.model)
    if args.cmd == "role":
        return cmd_role(args, args.model)
    if args.cmd == "user":
        if args.action in ("assign", "unassign"):
            return cmd_user(args, args.model)
        return cmd_user_ldap(args)
    if args.cmd == "group":
        return cmd_group(args, args.model)
    if args.cmd == "catalog":
        return cmd_catalog(args, args.model)
    raise ModelError(f"perintah tak dikenal: {args.cmd}")


def main(argv: list | None = None) -> int:
    args = build_parser().parse_args(argv)
    try:
        return dispatch(args)
    except (ModelError, L.LdapError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())