sicabe-core/deploy/config/opa/policies/trino.rego

# =============================================================================
# OPA — policy Trino access-control (package trino) — OPA-ONLY
# -----------------------------------------------------------------------------
# Trino (access-control.name=opa) mengirim satu request per operasi:
#   POST http://opa:8181/v1/data/trino/allow
#   input.context.identity = { "user": "...", "groups": [...] }
#   input.action.operation = "SELECT" | "INSERT" | ...
#   input.action.resource  = { catalogName, schemaName, tableName, columns }
# OPA mengembalikan { "result": { "allow": true|false } }.
#
# Grup user berasal dari LDAP group provider Trino (OpenLDAP, cn = data-*).
# Izin didefinisikan sebagai data di data.json → data.roles.<grup>.tables.<schema_table>.
# Nama catalog/schema TIDAK di-hardcode di policy: dibaca dari data.authz.catalogs /
# data.authz.write_schemas (hasil render config/opa/authz.yml) → rename catalog/schema
# cukup mengubah data, bukan Rego. Lihat scripts/sicabe-authz.
#   grant = { select: bool, write: bool, ddl: bool, columns: "*" | [..] }
#   - data-admin : superuser (semua operasi)
#   - select      : SELECT/SHOW COLUMNS/DESCRIBE
#   - write       : INSERT/UPDATE/DELETE/MERGE
#   - ddl         : CREATE/DROP/ALTER/... TABLE
# Kolom: bila columns != "*", SELECT hanya boleh menyebut kolom tsb.
# deny-by-default.
# =============================================================================
package trino

import rego.v1

default allow := false

# Normalisasi operasi: uppercase, buang spasi & underscore.
op := upper(replace(replace(trim_space(input.action.operation), " ", ""), "_", ""))

identity_groups := input.context.identity.groups
identity_user := input.context.identity.user

# ── Nama catalog/schema dari DATA (authz.yml), JANGAN di-hardcode di sini ────
# `default` → aman (deny-by-default) bila data.authz belum ada. Rujuk path
# SPESIFIK (data.authz.*) — JANGAN meneruskan `data` utuh ke builtin, sebab itu
# membuat compiler menganggapnya bergantung pada `allow` (rego_recursion_error).
default data_catalogs := []
data_catalogs := data.authz.catalogs

# Catalog yang boleh di-listing = catalog terdaftar + 'system' (bawaan Trino).
allowed_catalogs contains c if { some c in data_catalogs }
allowed_catalogs contains "system"

# Role efektif = grup (dari LDAP) ∪ role user (data.users) — untuk service account
# yang tidak terdaftar di LDAP (mis. svc_dbt, svc_om dari password file Basic).
effective_roles contains r if {
    some g in identity_groups
    r := g
}
effective_roles contains r if {
    r := data.users[identity_user]
}

# Grup 'data-admin' = superuser (bypass semua pembatasan data/sistem).
is_super if {
    some r in effective_roles
    r == "data-admin"
}

# ── CREATE/DROP SCHEMA (DDL level-schema) ───────────────────────────────────
# Trino mengirim operasi ini tanpa resource tabel → tak bisa memakai grant
# per-tabel di op_caps. Izinkan untuk peran engineering (data-admin/engineer).
allow if {
    op in {"CREATESCHEMA", "DROPSCHEMA"}
    some r in effective_roles
    r in {"data-admin", "data-engineer"}
}

# ── Pemetaan operasi -> kemampuan yang dibutuhkan ───────────────────────────
# SELECTFROMCOLUMNS TIDAK ada di sini (ditangani khusus + cek kolom).
op_caps := {
    "SHOWCOLUMNS": "select",
    "DESCRIBE": "select",
    "INSERTINTOTABLE": "write",
    "UPDATE": "write",
    "DELETE": "write",
    "MERGE": "write",
    "CREATETABLE": "ddl",
    "DROPTABLE": "ddl",
    "TRUNCATETABLE": "ddl",
    "ALTERTABLE": "ddl",
    "RENAMETABLE": "ddl",
    "ADDCOLUMN": "ddl",
    "DROPCOLUMN": "ddl",
    "RENAMECOLUMN": "ddl",
    "COMMENTTABLE": "ddl",
}

# ── Operasi metadata yang diizinkan utk user terautentikasi ────────────────
meta_op(opx) if startswith(opx, "SHOW")
meta_op(opx) if opx in {"USE", "SETSESSION", "RESETSESSION", "SETROLE", "SHOWROLES", "SHOWCURRENTROLES"}

# ── Eksekusi query (level sistem) ───────────────────────────────────────────
allow if op == "EXECUTEQUERY"

# ── Operasi metadata SHOW/USE/... utk user terautentikasi ──────────────────
allow if meta_op(op)

# ── Superuser ───────────────────────────────────────────────────────────────
allow if is_super

# ── Schema sistem/metadata (listing/autocomplete; isi data tetap terbatas) ─
allow if {
    op == "SELECTFROMCOLUMNS"
    res := input.action.resource
    res.table.catalogName == "system"
    res.table.schemaName in {"jdbc", "metadata", "information_schema"}
}
allow if {
    op == "SELECTFROMCOLUMNS"
    res := input.action.resource
    res.table.schemaName == "information_schema"
}

# ── Akses & listing catalog / schema ────────────────────────────────────────
allow if {
    op in {"ACCESSCATALOG", "FILTERCATALOGS"}
    res := input.action.resource
    res.catalog.name in allowed_catalogs
}
allow if {
    op == "FILTERSCHEMAS"
    res := input.action.resource
    res.schema.catalogName in allowed_catalogs
}
allow if {
    op == "FILTERTABLES"
    res := input.action.resource
    res.table.catalogName == "system"
    res.table.schemaName in {"jdbc", "metadata", "information_schema"}
}

# ── Operasi data/metadata per tabel (non-SELECT) ────────────────────────────
# dbt membuat tabel sementara <name>__dbt_tmp (dan backup <name>__dbt_backup):
# cocokkan ke grant tabel dasarnya.
dbt_suffixes := {"__dbt_tmp", "__dbt_backup"}

has_dbt_suffix(name) if {
    some s in dbt_suffixes
    endswith(name, s)
}

base_table_name(name) := n if {
    some s in dbt_suffixes
    endswith(name, s)
    n := substring(name, 0, count(name) - count(s))
}

base_table_name(name) := name if {
    not has_dbt_suffix(name)
}

table_id(schema, table) := sprintf("%s_%s", [schema, base_table_name(table)])

# ── Grant tingkat-SCHEMA: berlaku untuk SEMUA tabel di schema tersebut ───────
# Bentuk data (dari authz.yml `roles.<role>.schemas`):
#   data.roles.<role>.schema_grants.<schema> = { select/write/ddl, columns }
# Digabung (union) dengan grant per-tabel. Tabel baru (termasuk sementara dbt/Sling
# milik engineer) otomatis ikut ter-cover bila grant-nya di level schema.
schema_grant(role, schema) := sg if {
    sg := data.roles[role].schema_grants[schema]
}

allow if {
    cap := op_caps[op]
    res := input.action.resource
    tid := table_id(res.table.schemaName, res.table.tableName)
    some r in effective_roles
    grant := data.roles[r].tables[tid]
    grant[cap] == true
}

# ── SELECT ... (per-kolom) ──────────────────────────────────────────────────
# Semua kolom yang diminta harus diizinkan oleh grant.columns.
cols_all_ok(columns, cols) if columns == "*"
cols_all_ok(columns, cols) if {
    columns != "*"
    every c in cols {
        c in columns
    }
}

allow if {
    op == "SELECTFROMCOLUMNS"
    res := input.action.resource
    res.table.catalogName in data_catalogs
    tid := table_id(res.table.schemaName, res.table.tableName)
    cols := res.table.columns
    count(cols) > 0
    some r in effective_roles
    grant := data.roles[r].tables[tid]
    grant.select == true
    cols_all_ok(grant.columns, cols)
}

# ── SELECT tanpa kolom eksplisit (mis. count(*)) ────────────────────────────
# Trino mengirim columns: [] untuk count(*)/agregasi tanpa kolom. Izinkan bila
# user punya 'select' pada tabel tsb (tidak membocorkan isi kolom). Tanpa rule
# ini, count(*) selalu ditolak karena rule SELECT di atas butuh count(cols) > 0.
allow if {
    op == "SELECTFROMCOLUMNS"
    res := input.action.resource
    res.table.catalogName in data_catalogs
    tid := table_id(res.table.schemaName, res.table.tableName)
    count(res.table.columns) == 0
    some r in effective_roles
    grant := data.roles[r].tables[tid]
    grant.select == true
}

# ── FilterColumns — daftar kolom utk metadata (information_schema/SHOW COLUMNS)
# Trino memakai operasi ini untuk memfilter kolom yang boleh DILIHAT. Tanpa rule
# ini (deny-by-default) SEMUA kolom tersembunyi dari information_schema,
# `SHOW COLUMNS`, dan `DESCRIBE` — padahal SELECT-nya sendiri diizinkan. Akibat
# nyata: katalog metadata (mis. ingestion OpenMetadata) melihat tabel TANPA kolom.
# Batas per-kolom tetap dihormati lewat cols_all_ok (least privilege).
allow if {
    op == "FILTERCOLUMNS"
    res := input.action.resource
    res.table.schemaName == "information_schema"
}
allow if {
    op == "FILTERCOLUMNS"
    res := input.action.resource
    res.table.catalogName == "system"
}
allow if {
    op == "FILTERCOLUMNS"
    res := input.action.resource
    tid := table_id(res.table.schemaName, res.table.tableName)
    some r in effective_roles
    grant := data.roles[r].tables[tid]
    grant.select == true
    grant.columns == "*"
}
allow if {
    op == "FILTERCOLUMNS"
    res := input.action.resource
    tid := table_id(res.table.schemaName, res.table.tableName)
    cols := res.table.columns
    count(cols) > 0
    some r in effective_roles
    grant := data.roles[r].tables[tid]
    grant.select == true
    cols_all_ok(grant.columns, cols)
}

# ── Listing tabel (FilterTables): tabel terlihat bila ada grant apapun ─────
any_cap(grant) if {
    some cap in ["select", "write", "ddl"]
    grant[cap] == true
}

allow if {
    op == "FILTERTABLES"
    res := input.action.resource
    res.table.catalogName in data_catalogs
    tid := table_id(res.table.schemaName, res.table.tableName)
    some r in effective_roles
    grant := data.roles[r].tables[tid]
    any_cap(grant)
}

# ── Grant tingkat-SCHEMA (union dgn grant per-tabel di atas) ─────────────────
# Operasi per-tabel (non-SELECT) yang dicakup cap.
allow if {
    cap := op_caps[op]
    res := input.action.resource
    some r in effective_roles
    sg := schema_grant(r, res.table.schemaName)
    sg[cap] == true
}

# SELECT per-kolom pada tabel di schema yang di-grant.
allow if {
    op == "SELECTFROMCOLUMNS"
    res := input.action.resource
    res.table.catalogName in data_catalogs
    cols := res.table.columns
    count(cols) > 0
    some r in effective_roles
    sg := schema_grant(r, res.table.schemaName)
    sg.select == true
    cols_all_ok(sg.columns, cols)
}

# SELECT tanpa kolom eksplisit (count(*)) pada schema yang di-grant.
allow if {
    op == "SELECTFROMCOLUMNS"
    res := input.action.resource
    res.table.catalogName in data_catalogs
    count(res.table.columns) == 0
    some r in effective_roles
    sg := schema_grant(r, res.table.schemaName)
    sg.select == true
}

# Listing tabel: terlihat bila schema punya grant apapun.
allow if {
    op == "FILTERTABLES"
    res := input.action.resource
    res.table.catalogName in data_catalogs
    some r in effective_roles
    sg := schema_grant(r, res.table.schemaName)
    any_cap(sg)
}

# FilterColumns (visibilitas kolom) utk schema yang di-grant.
allow if {
    op == "FILTERCOLUMNS"
    res := input.action.resource
    some r in effective_roles
    sg := schema_grant(r, res.table.schemaName)
    sg.select == true
    sg.columns == "*"
}
allow if {
    op == "FILTERCOLUMNS"
    res := input.action.resource
    cols := res.table.columns
    count(cols) > 0
    some r in effective_roles
    sg := schema_grant(r, res.table.schemaName)
    sg.select == true
    cols_all_ok(sg.columns, cols)
}

allow if {
    op == "SetSystemSessionProperty"
}

# ── Blok luas: role dengan schema-grant 'write' = OPERASI APA PUN di schema ──
# Dulu HARDCODE (role data-engineer + schema bronze/silver/gold via write_schemas).
# Kini DATA-DRIVEN dari authz.yml `roles.<role>.schemas` — eksplisit & terport.
#
# Cakupan sengaja LEBIH LUAS dari least-privilege per-tabel, karena:
#   1. Menangkap OPERASI APA PUN di schema itu — termasuk operasi level-kolom yang
#      dikirim Trino untuk UPDATE/MERGE dan TIDAK punya entri di op_caps. Tanpa ini
#      `MERGE` (upsert) gagal: "Access Denied: Cannot update columns [...]".
#   2. Berlaku untuk tabel apa pun di schema itu, termasuk tabel sementara
#      `<nama>_tmp` (Sling) dan tabel baru → tak perlu grant per-tabel.
# Hanya bila schema-grant punya 'write' — role 'select'-only (mis. data-analyst)
# TIDAK terpengaruh (tetap least-privilege).
allow if {
    res := input.action.resource
    some r in effective_roles
    sg := schema_grant(r, res.table.schemaName)
    sg.write == true
}