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

sync a58faaa refactor(authz)

Browse Source
This commit is contained in:
subroto
2026-10-13 18:28:09 +07:00
parent 5519ae2de7
commit 678885f681
6 changed files with 290 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/lib/authz.py
+121 -13
View File
@@ -43,9 +43,10 @@ HEADER = """\
# ./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
# Policy Rego membaca data.authz.catalogs + grant schema (data.roles.<r>.schema_grants)
# — nama catalog/schema & cakupan GRANT TIDAK di-hardcode di Rego. Ganti nama
# catalog = ubah `catalogs:` di sini -> render -> reload. Grant tingkat-schema:
# `roles.<r>.schemas: [{schema, caps, columns}]`. Dokumentasi: docs/authorization-cli.md
# =============================================================================
"""
@@ -92,6 +93,22 @@ def _grant(grant: dict) -> dict:
return {"table": grant["table"], "caps": list(caps), "columns": columns}
def _schema_grant(sg: dict) -> dict:
if not isinstance(sg, dict) or "schema" not in sg:
raise ModelError(f"schema grant tidak valid: {sg!r}")
schema = str(sg["schema"])
if not schema or "." in schema:
raise ModelError(f"schema grant harus nama schema (tanpa '.'): {sg['schema']!r}")
caps = _as_list(sg.get("caps", []), f"caps untuk schema {schema}")
unknown = [c for c in caps if c not in CAPS]
if unknown:
raise ModelError(f"caps tak dikenal {unknown} pada schema {schema}")
columns = sg.get("columns", "*")
if columns != "*" and not isinstance(columns, list):
raise ModelError(f"columns harus '*' atau daftar, pada schema {schema}")
return {"schema": schema, "caps": list(caps), "columns": columns}
def validate(model: dict) -> None:
if not _as_list(model.get("catalogs", []), "catalogs"):
raise ModelError("catalogs tidak boleh kosong")
@@ -104,6 +121,8 @@ def validate(model: dict) -> None:
raise ModelError(f"role {name!r} harus objek")
for grant in (spec or {}).get("grants", []):
_grant(grant)
for sg in (spec or {}).get("schemas", []) or []:
_schema_grant(sg)
users = model.get("users", {})
if not isinstance(users, dict):
raise ModelError("users harus berupa objek")
@@ -118,17 +137,29 @@ def _render_role(spec: dict) -> dict:
spec = spec or {}
if spec.get("superuser"):
return {"superuser": True}
out: dict = {}
tables: dict = {}
for grant in spec.get("grants", []):
key = _table_key(grant["table"])
out: dict = {}
g: 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 {}
g[cap] = True
g["columns"] = grant.get("columns", "*")
tables[_table_key(grant["table"])] = g
if tables:
out["tables"] = tables
schemas: dict = {}
for sg in spec.get("schemas", []) or []:
g = {}
for cap in CAPS:
if cap in sg.get("caps", []):
g[cap] = True
g["columns"] = sg.get("columns", "*")
schemas[sg["schema"]] = g
if schemas:
out["schema_grants"] = schemas
return out
def render(model: dict, digest: str) -> dict:
@@ -243,6 +274,66 @@ def remove_grant(model: dict, role: str, table: str, caps: list | None) -> str:
return f"hapus cap {','.join(caps)} pada {role} -> {table}"
def get_schema_grant(spec: dict, schema: str):
for sg in (spec or {}).get("schemas", []) or []:
if sg["schema"] == schema:
return sg
return None
def _spec_for(model: dict, role: str) -> dict:
roles = model.setdefault("roles", {})
if role not in roles:
raise ModelError(f"role {role!r} belum ada — buat dulu: role add {role}")
spec = roles[role]
if spec is None:
spec = {}
roles[role] = spec
if spec.get("superuser"):
raise ModelError(f"role {role!r} superuser — tak perlu grant")
return spec
def set_schema_grant(model: dict, role: str, schema: str, caps: list, columns, replace: bool) -> str:
if not schema or "." in schema:
raise ModelError(f"schema harus nama schema (tanpa '.'): {schema!r}")
for c in caps:
if c not in CAPS:
raise ModelError(f"cap tak dikenal: {c} (pilihan: {', '.join(CAPS)})")
spec = _spec_for(model, role)
sgs = spec.setdefault("schemas", [])
existing = get_schema_grant(spec, schema)
if existing is None:
sgs.append({"schema": schema, "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} -> {schema}.*"
def remove_schema_grant(model: dict, role: str, schema: str, caps: list | None) -> str:
spec = (model.get("roles", {}) or {}).get(role)
if not spec or not spec.get("schemas"):
raise ModelError(f"role {role!r} tak punya schema grant")
existing = get_schema_grant(spec, schema)
if existing is None:
raise ModelError(f"schema grant {role} -> {schema}.* tidak ada")
if not caps:
spec["schemas"] = [sg for sg in spec["schemas"] if sg["schema"] != schema]
return f"hapus schema grant {role} -> {schema}.*"
left = [c for c in existing.get("caps", []) if c not in caps]
if left:
existing["caps"] = left
else:
spec["schemas"] = [sg for sg in spec["schemas"] if sg["schema"] != schema]
return f"hapus cap {','.join(caps)} pada {role} -> {schema}.*"
def roles_referencing(model: dict, role: str) -> list:
return [u for u, r in model.get("users", {}).items() if r == role]
@@ -337,6 +428,14 @@ def cmd_matrix(model_path: Path, schema: str | None) -> int:
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")
schema_lines = []
for r in roles:
spec = (model.get("roles", {}) or {}).get(r) or {}
for sg in spec.get("schemas", []) or []:
schema_lines.append(f" {r:<{w}} {sg['schema']}.* [{caps_letters(sg.get('caps', []))}]")
if schema_lines:
print("\n Grant tingkat-SCHEMA (berlaku untuk SEMUA tabel di schema):")
print("\n".join(schema_lines))
return 0
@@ -350,7 +449,9 @@ def cmd_show(model_path: Path, what: str, name: str) -> int:
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"):
for sg in spec.get("schemas", []) or []:
print(f"{name}: {sg['schema']}.* caps=[{','.join(sg.get('caps', []))}] columns={sg.get('columns')} (seluruh schema)")
if not spec.get("superuser") and not spec.get("grants") and not spec.get("schemas"):
print(f"{name}: (tanpa grant)")
elif what == "user":
role = (model.get("users", {}) or {}).get(name)
@@ -392,12 +493,19 @@ def save_model(model_path: Path, model: dict) -> None:
def cmd_grant(args, model_path: Path) -> int:
model = load_model(model_path)
is_schema = args.table.endswith(".*")
target = args.table[:-2] if is_schema else args.table
caps = _split_csv(args.caps)
columns = _columns_arg(args.columns)
if args.action == "add":
msg = set_grant(model, args.role, args.table, _split_csv(args.caps), _columns_arg(args.columns), False)
fn = set_schema_grant if is_schema else set_grant
msg = fn(model, args.role, target, caps, columns, False)
elif args.action == "set":
msg = set_grant(model, args.role, args.table, _split_csv(args.caps), _columns_arg(args.columns), True)
fn = set_schema_grant if is_schema else set_grant
msg = fn(model, args.role, target, caps, columns, True)
else:
msg = remove_grant(model, args.role, args.table, _split_csv(getattr(args, "caps", None)) or None)
fn = remove_schema_grant if is_schema else remove_grant
msg = fn(model, args.role, target, caps or None)
_say(msg)
save_model(model_path, model)
return 0