platform/sicabe-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat sicabe-authz

Browse Source
This commit is contained in:
subroto
2026-10-13 14:44:55 +07:00
parent 082de8e3b2
commit f534e8da54
6 changed files with 1030 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/lib/authz.py
+610
View File
@@ -0,0 +1,610 @@
#!/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
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
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"])
pu.add_argument("user")
pu.add_argument("role", nargs="?")
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":
return cmd_user(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 as exc:
print(f"ERROR: {exc}", file=sys.stderr)
return 2
if __name__ == "__main__":
raise SystemExit(main())