update lakekeeper/0.11.0

- chart 0.8.1 → 0.11.0 (appVersion 0.10.4 → 0.12.2)
- deps: postgres 1.5.8 → 1.5.13, openfga 0.2.44 → 0.2.62
- ingress: Kong → APISIX (use-regex + path /.*, cluster-issuer)
- openfga.playground 비활성화 (preshared 인증 패닉 방지)
- breaking=false, custom-values 키 전부 호환

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
wbsong111
2026-06-15 09:13:31 +09:00
parent 1ab3d4d31c
commit f4d287abef
217 changed files with 23584 additions and 0 deletions
@@ -0,0 +1,79 @@
# This file contains all configurations for the Lakekeeper OPA bridge.
# regal ignore:directory-package-mismatch
package configuration
import data.catalogs
env := opa.runtime().env
# ------------- Lakekeeper Configuration -------------
# Define lakekeeper instances.
#
# The first instance can be configured via environment variables.
# Each configuration must contain the following fields:
#
# - id: Id of this lakekeeper instance.
# Has no relevance except as an internal OPA identifier.
# Used to reference this instance in the `trino_catalog` configuration below.
# - url: The URL where OPA can reach the Lakekeeper instance.
# - openid_token_endpoint: The URL of the token endpoint of the identity provider.
# Example: http://localhost:30080/realms/iceberg/protocol/openid-connect/token
# - client_id: The client ID used for authentication with the IdP (Client Credentials Flow)
# - client_secret: The client secret used for authentication with the IdP (Client Credentials Flow)
# - scope: The scope specified in the client credentials flow
# - max_batch_check_size: Maximum number of checks per batch-check request.
# Lakekeeper server limit is 1000. Default: 1000
#
# A default project is pre-defined and can be configured via environment variables.
# Additional projects can be added below.
lakekeeper := [{
"id": "default",
"url": trim_right(object.get(env, "LAKEKEEPER_URL", "http://localhost:8181"), "/"),
"openid_token_endpoint": env.LAKEKEEPER_TOKEN_ENDPOINT,
"client_id": env.LAKEKEEPER_CLIENT_ID,
"client_secret": env.LAKEKEEPER_CLIENT_SECRET,
"scope": object.get(env, "LAKEKEEPER_SCOPE", "lakekeeper"),
"max_batch_check_size": to_number(object.get(env, "LAKEKEEPER_MAX_BATCH_CHECK_SIZE", "1000")),
}]
# ------------- Trino Configuration -------------
# Mapping of trino catalogs to Lakekeeper warehouses.
# Add additional entries for additional trino catalogs.
# Each configuration must contain the following fields:
#
# - name: The name of the catalog in Trino.
# - lakekeeper_id: Id of the Lakekeeper configuration. (Reference to "id" in the "lakekeeper" array)
# - lakekeeper_warehouse: The name of the warehouse in Lakekeeper.
# Allow access to unmanaged catalogs (catalogs not in the trino_catalog array).
# When Trino has multiple authorizers configured, ALL authorizers must allow an action for it to succeed.
# If Trino uses catalogs managed by other authorizers (not Lakekeeper), such as a connected PostgreSQL catalog,
# set this to true to allow this OPA bridge to permit access to those catalogs.
# Default: false
default trino_allow_unmanaged_catalogs := false
trino_allow_unmanaged_catalogs if {
object.get(env, "TRINO_ALLOW_UNMANAGED_CATALOGS", "false") == "true"
}
# ------------- Trino Admin Users -------------
# Users with full access to all system schemas/tables across all catalogs,
# including FilterViewQueryOwnedBy for all users.
# Specify Trino user IDs (typically OIDC subject).
# Can be set via the TRINO_ADMIN_USERS environment variable as a comma-separated list.
default trino_admin_users := []
trino_admin_users := admin_users if {
admin_csv := object.get(env, "TRINO_ADMIN_USERS", "")
admin_csv != ""
admin_users := [trimmed |
some entry in split(admin_csv, ",")
trimmed := trim_space(entry)
trimmed != ""
]
}
# Trino catalog mappings are generated by the Helm chart from OPABridge.catalogs values.
# They are provided via the catalogs.rego file in the same package.
trino_catalog := data.catalogs.trino_catalog
@@ -0,0 +1,36 @@
package lakekeeper
# Retrieve an access token from the identity provider.
# Caches the token for 150 seconds.
# Token is specific to a `lakekeeper_id`.
access_token[lakekeeper_id] := access_token if {
some lakekeeper_id, this in config_by_id
value := http.send({
"method": "POST",
"headers": {"Content-type": "application/x-www-form-urlencoded"},
"url": this.openid_token_endpoint,
"force_cache": true,
"force_cache_duration_seconds": 150,
"caching_mode": "deserialized",
"raw_body": sprintf(
"grant_type=client_credentials&client_id=%v&client_secret=%v&scope=%v",
[this.client_id, this.client_secret, this.scope],
),
}).body
access_token := value.access_token
}
# Send an authenticated HTTP request to the lakekeeper service
authenticated_http_send(lakekeeper_id, method, path, body) := response if {
this := config_by_id[lakekeeper_id]
url := concat("/", [this.url, trim_left(path, "/")])
response := http.send({
"method": method,
"url": url,
"headers": {
"Authorization": sprintf("Bearer %v", [access_token[lakekeeper_id]]),
"Content-Type": "application/json",
},
"body": body,
})
}
@@ -0,0 +1,255 @@
package lakekeeper
import data.configuration
# Get a lakekeeper project by its name
config_by_id[lakekeeper_id] := lakekeeper if {
some lakekeeper in configuration.lakekeeper
lakekeeper_id := lakekeeper.id
}
# Check access to a warehouse
require_warehouse_access_simple(lakekeeper_id, warehouse_name, user, action) if {
value := authenticated_http_send(
lakekeeper_id,
"POST", "/management/v1/action/batch-check",
{
"error-on-not-found": false,
"checks": [{
"operation": {"warehouse": {
"action": {"action": action},
"warehouse-id": warehouse_id_for_name(lakekeeper_id, warehouse_name),
}},
"identity": {"user": user},
}],
},
).body
value.results[0].allowed == true
count(value.results) == 1
}
# Check access to a warehouse
require_warehouse_create_namespace_access(lakekeeper_id, warehouse_name, user, properties, name) if {
value := authenticated_http_send(
lakekeeper_id,
"POST", "/management/v1/action/batch-check",
{
"error-on-not-found": false,
"checks": [{
"operation": {"warehouse": {
"action": {"action": "create_namespace", "properties": properties, "name": name},
"warehouse-id": warehouse_id_for_name(lakekeeper_id, warehouse_name),
}},
"identity": {"user": user},
}],
},
).body
value.results[0].allowed == true
count(value.results) == 1
}
# Check access to a namespace
require_namespace_access_simple(lakekeeper_id, warehouse_name, namespace_name, user, action) if {
value := authenticated_http_send(
lakekeeper_id,
"POST", "/management/v1/action/batch-check",
{
"error-on-not-found": false,
"checks": [{
"operation": {"namespace": {
"action": {"action": action},
"warehouse-id": warehouse_id_for_name(lakekeeper_id, warehouse_name),
"namespace": namespace_name,
}},
"identity": {"user": user},
}],
},
).body
value.results[0].allowed == true
count(value.results) == 1
}
require_namespace_access_create(lakekeeper_id, warehouse_name, namespace_name, user, action, properties, name) if {
value := authenticated_http_send(
lakekeeper_id,
"POST", "/management/v1/action/batch-check",
{
"error-on-not-found": false,
"checks": [{
"operation": {"namespace": {
"action": {"action": action, "properties": properties, "name": name},
"warehouse-id": warehouse_id_for_name(lakekeeper_id, warehouse_name),
"namespace": namespace_name,
}},
"identity": {"user": user},
}],
},
).body
value.results[0].allowed == true
count(value.results) == 1
}
require_namespace_access_update_properties(
lakekeeper_id,
warehouse_name,
namespace_name,
user,
removed_properties,
updated_properties,
) if {
value := authenticated_http_send(
lakekeeper_id,
"POST", "/management/v1/action/batch-check",
{
"error-on-not-found": false,
"checks": [{
"operation": {"namespace": {
"action": {
"action": "update_properties",
"removed_properties": removed_properties,
"updated_properties": updated_properties,
},
"warehouse-id": warehouse_id_for_name(lakekeeper_id, warehouse_name),
"namespace": namespace_name,
}},
"identity": {"user": user},
}],
},
).body
value.results[0].allowed == true
count(value.results) == 1
}
# Check access to a table
require_table_access_simple(lakekeeper_id, warehouse_name, namespace_name, table_name, user, action) if {
value := authenticated_http_send(
lakekeeper_id,
"POST", "/management/v1/action/batch-check",
{
"error-on-not-found": false,
"checks": [{
"operation": {"table": {
"action": {"action": action},
"warehouse-id": warehouse_id_for_name(lakekeeper_id, warehouse_name),
"namespace": namespace_name,
"table": table_name,
}},
"identity": {"user": user},
}],
},
).body
value.results[0].allowed == true
count(value.results) == 1
}
# Check access to a table
require_table_access_commit(
lakekeeper_id, warehouse_name, namespace_name,
table_name, user, updated_properties, removed_properties,
) if {
value := authenticated_http_send(
lakekeeper_id,
"POST", "/management/v1/action/batch-check",
{
"error-on-not-found": false,
"checks": [{
"operation": {"table": {
"action": {
"action": "commit",
"updated_properties": updated_properties,
"removed_properties": removed_properties,
},
"warehouse-id": warehouse_id_for_name(lakekeeper_id, warehouse_name),
"namespace": namespace_name,
"table": table_name,
}},
"identity": {"user": user},
}],
},
).body
value.results[0].allowed == true
count(value.results) == 1
}
# Execute a batch-check, automatically chunking into multiple HTTP requests
# if the number of checks exceeds the instance's max_batch_check_size.
# Returns a flat array of results in the same order as the input checks.
batch_check_results(lakekeeper_id, checks) := all_results if {
count(checks) > 0
_raw_max := object.get(config_by_id[lakekeeper_id], "max_batch_check_size", 1000)
max_size := max([_raw_max, 1])
num_batches := ceil(count(checks) / max_size)
all_results := [result |
some batch_idx in numbers.range(0, num_batches - 1)
start := batch_idx * max_size
batch := array.slice(checks, start, start + max_size)
chunk_results := _batch_check_http(lakekeeper_id, batch)
some result in chunk_results
]
}
# Send a single batch-check HTTP request.
_batch_check_http(lakekeeper_id, checks) := value.results if {
value := authenticated_http_send(
lakekeeper_id,
"POST", "/management/v1/action/batch-check",
{
"error-on-not-found": false,
"checks": checks,
},
).body
}
# Build a single check object for a table action
build_table_check(warehouse_id, namespace_name, table_name, user, action) := {
"operation": {"table": {
"action": {"action": action},
"warehouse-id": warehouse_id,
"namespace": namespace_name,
"table": table_name,
}},
"identity": {"user": user},
}
# Build a single check object for a view action
build_view_check(warehouse_id, namespace_name, view_name, user, action) := {
"operation": {"view": {
"action": {"action": action},
"warehouse-id": warehouse_id,
"namespace": namespace_name,
"table": view_name,
}},
"identity": {"user": user},
}
# Build a single check object for a namespace action
build_namespace_check(warehouse_id, namespace_name, user, action) := {
"operation": {"namespace": {
"action": {"action": action},
"warehouse-id": warehouse_id,
"namespace": namespace_name,
}},
"identity": {"user": user},
}
# Check access to a view
require_view_access_simple(lakekeeper_id, warehouse_name, namespace_name, view_name, user, action) if {
value := authenticated_http_send(
lakekeeper_id,
"POST", "/management/v1/action/batch-check",
{
"error-on-not-found": false,
"checks": [{
"operation": {"view": {
"action": {"action": action},
"warehouse-id": warehouse_id_for_name(lakekeeper_id, warehouse_name),
"namespace": namespace_name,
"table": view_name,
}},
"identity": {"user": user},
}],
},
).body
value.results[0].allowed == true
count(value.results) == 1
}
@@ -0,0 +1,16 @@
package lakekeeper
# Translate a warehouse name to a warehouse ID.
# Cache the result for 1 hour to avoid repeated calls to the Lakekeeper API.
warehouse_id_for_name(lakekeeper_id, warehouse_name) := warehouse_id if {
this := config_by_id[lakekeeper_id]
url := concat("/", [this.url, sprintf("catalog/v1/config?warehouse=%s", [urlquery.encode(warehouse_name)])])
warehouse_id := http.send({
"method": "GET",
"url": url,
"headers": {"Authorization": sprintf("Bearer %v", [access_token[lakekeeper_id]])},
"force_cache": true,
"force_cache_duration_seconds": 3600,
"caching_mode": "deserialized",
}).body.defaults.prefix
}
@@ -0,0 +1,21 @@
package trino
allow_catalog if {
allow_catalog_management
}
allow_catalog if {
allow_catalog_access
}
allow_catalog_management if {
input.action.operation in ["CreateCatalog", "DropCatalog"]
catalog := input.action.resource.catalog.name
require_catalog_access_simple(catalog, "delete")
}
allow_catalog_access if {
input.action.operation in ["AccessCatalog", "FilterCatalogs"]
catalog := input.action.resource.catalog.name
require_catalog_access_simple(catalog, "get_config")
}
@@ -0,0 +1,182 @@
package trino
import data.configuration
# Tables allowed in information_schema across all catalogs.
# Re-used in allow_schema.rego for Lakekeeper catalogs.
allowed_information_schema_tables := ["columns", "schemata", "tables", "views"]
# System schemas in Lakekeeper catalogs (not user-created, handled separately)
lakekeeper_system_schemas := ["information_schema", "schema_discovery", "system"]
# Tables allowed in Lakekeeper catalog system schemas
allowed_schema_discovery_tables := ["discovery", "shallow_discovery"]
allowed_system_schema_tables := ["iceberg_tables"]
# Tables allowed in system.metadata - excludes *_authorization and broken_catalog_definitions
allowed_metadata_tables := [
"analyze_properties",
"catalogs",
"column_properties",
"materialized_views",
"schema_properties",
"table_comments",
"table_properties",
]
# Admin check - reusable across policies
is_admin if {
trino_user_id in configuration.trino_admin_users
}
allow_default_access if {
allow_admin_system_access
}
allow_default_access if {
allow_execute_query
}
allow_default_access if {
allow_access_catalog_on_system_catalog
}
allow_default_access if {
allow_jdbc_schema
}
allow_default_access if {
allow_filter_catalogs_for_system_catalog
}
allow_default_access if {
allow_show_schemas_on_system_catalog
}
allow_default_access if {
allow_metadata_schema
}
allow_default_access if {
allow_information_schema
}
allow_default_access if {
allow_runtime_schema
}
allow_default_access if {
allow_view_own_queries
}
allow_default_access if {
allow_read_system_information
}
# Every authenticated user can execute queries.
allow_execute_query if {
input.action.operation == "ExecuteQuery"
}
allow_access_catalog_on_system_catalog if {
input.action.operation == "AccessCatalog"
input.action.resource.catalog.name == "system"
}
allow_show_schemas_on_system_catalog if {
input.action.operation == "ShowSchemas"
input.action.resource.catalog.name == "system"
}
# Allow access for JDBC Clients, which require access to the "system.jdbc" schema
allow_jdbc_schema if {
input.action.operation == "FilterSchemas"
input.action.resource.schema.catalogName == "system"
input.action.resource.schema.schemaName == "jdbc"
}
allow_jdbc_schema if {
input.action.operation in ["SelectFromColumns", "FilterTables", "FilterColumns"]
input.action.resource.table.catalogName == "system"
input.action.resource.table.schemaName == "jdbc"
}
# Metadata schema used by starburst UI
allow_metadata_schema if {
input.action.operation == "FilterSchemas"
input.action.resource.schema.catalogName == "system"
input.action.resource.schema.schemaName == "metadata"
}
allow_metadata_schema if {
input.action.operation in ["SelectFromColumns", "FilterTables", "FilterColumns"]
input.action.resource.table.catalogName == "system"
input.action.resource.table.schemaName == "metadata"
input.action.resource.table.tableName in allowed_metadata_tables
}
allow_information_schema if {
input.action.operation == "FilterSchemas"
input.action.resource.schema.catalogName == "system"
input.action.resource.schema.schemaName == "information_schema"
}
allow_information_schema if {
input.action.operation in ["SelectFromColumns", "FilterTables", "FilterColumns"]
input.action.resource.table.catalogName == "system"
input.action.resource.table.schemaName == "information_schema"
input.action.resource.table.tableName in allowed_information_schema_tables
}
# Public access to system.runtime - only the "queries" table
allow_runtime_schema if {
input.action.operation == "FilterSchemas"
input.action.resource.schema.catalogName == "system"
input.action.resource.schema.schemaName == "runtime"
}
allow_runtime_schema if {
input.action.operation in ["SelectFromColumns", "FilterTables", "FilterColumns"]
input.action.resource.table.catalogName == "system"
input.action.resource.table.schemaName == "runtime"
input.action.resource.table.tableName == "queries"
}
allow_filter_catalogs_for_system_catalog if {
input.action.operation == "FilterCatalogs"
input.action.resource.catalog.name == "system"
}
# ------------- Admin Access -------------
# Admins get full access to all schemas and tables in the system catalog
# (no table filtering on metadata, information_schema, etc.)
allow_admin_system_access if {
is_admin
input.action.operation in ["FilterSchemas", "ShowSchemas"]
input.action.resource.schema.catalogName == "system"
}
allow_admin_system_access if {
is_admin
input.action.operation in ["SelectFromColumns", "FilterTables", "FilterColumns"]
input.action.resource.table.catalogName == "system"
}
# Admins can view queries owned by any user
allow_admin_system_access if {
is_admin
input.action.operation in ["FilterViewQueryOwnedBy", "ViewQueryOwnedBy"]
}
# Non-admins can only view their own queries
allow_view_own_queries if {
input.action.operation in ["FilterViewQueryOwnedBy", "ViewQueryOwnedBy"]
input.action.resource.user.user == input.context.identity.user
}
# Allow access to ReadSystemInformation, is required
# for Trino in order to allow access to the API /metrics endpoint
# see: https://trino.io/docs/current/admin/openmetrics.html
allow_read_system_information if {
input.action.operation == "ReadSystemInformation"
}
@@ -0,0 +1,157 @@
package trino
allow_schema if {
allow_schema_create
}
allow_schema if {
allow_schema_drop
}
allow_schema if {
allow_schema_rename
}
allow_schema if {
allow_show_schemas
}
allow_schema if {
allow_tables_in_system_schemas
}
allow_schema if {
allow_filter_schemas
}
allow_schema if {
allow_show_create_schemas
}
allow_schema if {
allow_show_tables_in_schema
}
allow_schema if {
allow_filter_system_schemas
}
allow_schema if {
allow_admin_system_schemas
}
allow_schema_create if {
input.action.operation == "CreateSchema"
schema := input.action.resource.schema.schemaName
is_nested_schema(schema) == false
catalog := input.action.resource.schema.catalogName
properties := object.get(input.action.resource.schema, "properties", {})
flattended_properties := flatten_properties(properties)
require_catalog_create_namespace_access(catalog, flattended_properties, schema)
}
allow_schema_create if {
input.action.operation == "CreateSchema"
schema := input.action.resource.schema.schemaName
is_nested_schema(schema) == true
catalog := input.action.resource.schema.catalogName
properties := object.get(input.action.resource.schema, "properties", {})
flattended_properties := flatten_properties(properties)
require_schema_access_create(
catalog,
parent_schema(schema),
"create_namespace",
flattended_properties,
child_schema_name(schema),
)
}
allow_schema_drop if {
input.action.operation == "DropSchema"
catalog := input.action.resource.schema.catalogName
schema := input.action.resource.schema.schemaName
require_schema_access_simple(catalog, schema, "delete")
}
# renameNamespace is not supported for Iceberg REST catalog in trino.
# Lakekeeper supports renaming schemas, please use the UI or management API
# to rename schemas. (namespaces)
default allow_schema_rename := false
allow_show_schemas if {
input.action.operation == "ShowSchemas"
catalog := input.action.resource.catalog.name
require_catalog_access_simple(catalog, "list_namespaces")
}
allow_filter_system_schemas if {
input.action.operation == "FilterSchemas"
schema := input.action.resource.schema.schemaName
schema in lakekeeper_system_schemas
catalog := input.action.resource.schema.catalogName
require_catalog_access_simple(catalog, "get_config")
}
# Table-level access for Lakekeeper system schemas.
# Each schema has its own allowed table list defined in allow_default_access.rego.
allow_tables_in_system_schemas if {
input.action.operation in ["SelectFromColumns", "FilterTables", "FilterColumns"]
input.action.resource.table.schemaName == "information_schema"
input.action.resource.table.tableName in allowed_information_schema_tables
catalog := input.action.resource.table.catalogName
require_catalog_access_simple(catalog, "get_config")
}
allow_tables_in_system_schemas if {
input.action.operation in ["SelectFromColumns", "FilterTables", "FilterColumns"]
input.action.resource.table.schemaName == "schema_discovery"
input.action.resource.table.tableName in allowed_schema_discovery_tables
catalog := input.action.resource.table.catalogName
require_catalog_access_simple(catalog, "get_config")
}
allow_tables_in_system_schemas if {
input.action.operation in ["SelectFromColumns", "FilterTables", "FilterColumns"]
input.action.resource.table.schemaName == "system"
input.action.resource.table.tableName in allowed_system_schema_tables
catalog := input.action.resource.table.catalogName
require_catalog_access_simple(catalog, "get_config")
}
allow_filter_schemas if {
input.action.operation == "FilterSchemas"
schema := input.action.resource.schema.schemaName
not schema in lakekeeper_system_schemas
catalog := input.action.resource.schema.catalogName
require_schema_access_simple(catalog, schema, "get_metadata")
}
allow_show_create_schemas if {
input.action.operation == "ShowCreateSchema"
schema := input.action.resource.schema.schemaName
not schema in lakekeeper_system_schemas
catalog := input.action.resource.schema.catalogName
require_schema_access_simple(catalog, schema, "get_metadata")
}
allow_show_tables_in_schema if {
input.action.operation == "ShowTables"
catalog := input.action.resource.schema.catalogName
schema := input.action.resource.schema.schemaName
require_schema_access_simple(catalog, schema, "get_metadata")
}
# ------------- Admin Access -------------
# Admins get full access to all tables in Lakekeeper system schemas
# (no table filtering on information_schema, schema_discovery, system)
allow_admin_system_schemas if {
is_admin
input.action.operation == "FilterSchemas"
input.action.resource.schema.schemaName in lakekeeper_system_schemas
}
allow_admin_system_schemas if {
is_admin
input.action.operation in ["SelectFromColumns", "FilterTables", "FilterColumns"]
input.action.resource.table.schemaName in lakekeeper_system_schemas
}
@@ -0,0 +1,133 @@
package trino
allow_table if {
allow_table_create
}
allow_table if {
allow_table_drop
}
allow_table if {
allow_table_rename
}
allow_table if {
allow_table_modify
}
allow_table if {
allow_table_metadata
}
allow_table if {
allow_table_read
}
allow_table if {
allow_table_procedure
}
allow_table if {
allow_table_metadata_read
}
allow_table_create if {
input.action.operation == "CreateTable"
table := input.action.resource.table.tableName
is_metadata_table(table) == false
catalog := input.action.resource.table.catalogName
schema := input.action.resource.table.schemaName
properties := object.get(input.action.resource.table, "properties", {})
flattened_properties := flatten_properties(properties)
require_schema_access_create(catalog, schema, "create_table", flattened_properties, table)
}
allow_table_drop if {
input.action.operation == "DropTable"
table := input.action.resource.table.tableName
is_metadata_table(table) == false
catalog := input.action.resource.table.catalogName
schema := input.action.resource.table.schemaName
require_table_access_simple(catalog, schema, table, "drop")
}
allow_table_rename if {
input.action.operation == "RenameTable"
source_table := input.action.resource.table.tableName
is_metadata_table(source_table) == false
source_catalog := input.action.resource.table.catalogName
source_schema := input.action.resource.table.schemaName
require_table_access_simple(source_catalog, source_schema, source_table, "rename")
target_catalog := input.action.targetResource.table.catalogName
target_schema := input.action.targetResource.table.schemaName
require_schema_access_simple(target_catalog, target_schema, "create_table")
}
allow_table_modify if {
input.action.operation in [
"SetTableComment", "SetColumnComment",
"AddColumn", "AlterColumn", "DropColumn", "RenameColumn",
"InsertIntoTable", "DeleteFromTable", "TruncateTable",
"UpdateTableColumns",
]
table := input.action.resource.table.tableName
is_metadata_table(table) == false
catalog := input.action.resource.table.catalogName
schema := input.action.resource.table.schemaName
require_table_access_simple(catalog, schema, table, "write_data")
}
allow_table_modify if {
input.action.operation == "SetTableProperties"
table := input.action.resource.table.tableName
is_metadata_table(table) == false
catalog := input.action.resource.table.catalogName
schema := input.action.resource.table.schemaName
properties := object.get(input.action.resource.table, "properties", {})
flattened_properties := flatten_properties(properties)
require_table_access_commit(catalog, schema, table, flattened_properties, [])
}
allow_table_metadata if {
input.action.operation in ["FilterTables", "ShowColumns", "FilterColumns", "ShowCreateTable"]
table := input.action.resource.table.tableName
is_metadata_table(table) == false
catalog := input.action.resource.table.catalogName
schema := input.action.resource.table.schemaName
require_table_access_simple(catalog, schema, table, "get_metadata")
}
allow_table_read if {
input.action.operation in ["SelectFromColumns", "CreateViewWithSelectFromColumns"]
table := input.action.resource.table.tableName
is_metadata_table(table) == false
catalog := input.action.resource.table.catalogName
schema := input.action.resource.table.schemaName
require_table_access_simple(catalog, schema, table, "read_data")
}
allow_table_metadata_read if {
input.action.operation == "SelectFromColumns"
catalog := input.action.resource.table.catalogName
schema := input.action.resource.table.schemaName
table := split_metadata_table_name(input.action.resource.table.tableName)
require_table_access_simple(catalog, schema, table, "get_metadata")
}
allow_table_procedure if {
input.action.operation == "ExecuteTableProcedure"
input.action.resource.function.functionName in [
"OPTIMIZE",
"OPTIMIZE_MANIFESTS",
"EXPIRE_SNAPSHOTS",
"DROP_EXTENDED_STATS",
"REMOVE_ORPHAN_FILES",
"ADD_FILES",
]
table := input.action.resource.table.tableName
is_metadata_table(table) == false
catalog := input.action.resource.table.catalogName
schema := input.action.resource.table.schemaName
require_table_access_simple(catalog, schema, table, "write_data")
}
@@ -0,0 +1,78 @@
package trino
allow_view if {
allow_view_modify
}
allow_view if {
allow_view_create
}
allow_view if {
allow_view_rename
}
allow_view if {
allow_view_drop
}
allow_view if {
allow_view_metadata
}
allow_view if {
allow_view_read
}
allow_view_create if {
input.action.operation in ["CreateView", "CreateMaterializedView"]
catalog := input.action.resource.table.catalogName
schema := input.action.resource.table.schemaName
view := input.action.resource.table.tableName
properties := object.get(input.action.resource.table, "properties", {})
flattened_properties := flatten_properties(properties)
require_schema_access_create(catalog, schema, "create_view", flattened_properties, view)
}
allow_view_modify if {
input.action.operation == "SetViewComment"
catalog := input.action.resource.table.catalogName
schema := input.action.resource.table.schemaName
table := input.action.resource.table.tableName
require_view_access_simple(catalog, schema, table, "write_data")
}
allow_view_rename if {
input.action.operation == "RenameView"
source_catalog := input.action.resource.table.catalogName
source_schema := input.action.resource.table.schemaName
source_table := input.action.resource.table.tableName
require_view_access_simple(source_catalog, source_schema, source_table, "rename")
target_catalog := input.action.targetResource.table.catalogName
target_schema := input.action.targetResource.table.schemaName
require_schema_access_simple(target_catalog, target_schema, "create_view")
}
allow_view_drop if {
input.action.operation == "DropView"
catalog := input.action.resource.table.catalogName
schema := input.action.resource.table.schemaName
table := input.action.resource.table.tableName
require_view_access_simple(catalog, schema, table, "drop")
}
allow_view_metadata if {
input.action.operation in ["FilterTables", "ShowColumns", "FilterColumns"]
catalog := input.action.resource.table.catalogName
schema := input.action.resource.table.schemaName
table := input.action.resource.table.tableName
require_view_access_simple(catalog, schema, table, "get_metadata")
}
allow_view_read if {
input.action.operation == "SelectFromColumns"
catalog := input.action.resource.table.catalogName
schema := input.action.resource.table.schemaName
table := input.action.resource.table.tableName
require_view_access_simple(catalog, schema, table, "get_metadata")
}
@@ -0,0 +1,163 @@
# Contains convenience wrappers around Lakekeeper functions and rules
# to reduce the number inputs required to call them
package trino
import data.configuration
import data.lakekeeper
catalog_config_by_name[catalog_name] := trino_catalog if {
some trino_catalog in configuration.trino_catalog
catalog_name := trino_catalog.name
}
# flatten properties by merging extra_properties into the main properties object
flatten_properties(properties) := flattened if {
extra := object.get(properties, "extra_properties", {})
base := object.remove(properties, ["extra_properties"])
flattened := object.union(base, extra)
}
# Iceberg REST Namespaces are multi part identifiers (arrays).
# Trino schemas are strings separated by dots.
namespace_for_schema(schema_name) := namespace_name if {
namespace_name := split(schema_name, ".")
count(namespace_name) > 0
}
is_nested_schema(schema_name) := is_nested if {
namespace_name := namespace_for_schema(schema_name)
is_nested := count(namespace_name) > 1
}
parent_schema(schema_name) := parent_schema if {
namespace_name := namespace_for_schema(schema_name)
parent_namespace := array.slice(namespace_name, 0, count(namespace_name) - 1)
parent_schema := concat(".", parent_namespace)
}
require_catalog_access_simple(catalog_name, action) if {
trino_catalog := catalog_config_by_name[catalog_name]
lakekeeper.require_warehouse_access_simple(
trino_catalog.lakekeeper_id,
trino_catalog.lakekeeper_warehouse,
lakekeeper_user_id,
action,
)
}
require_catalog_create_namespace_access(catalog_name, properties, name) if {
trino_catalog := catalog_config_by_name[catalog_name]
lakekeeper.require_warehouse_create_namespace_access(
trino_catalog.lakekeeper_id,
trino_catalog.lakekeeper_warehouse,
lakekeeper_user_id,
properties,
name,
)
}
require_schema_access_simple(catalog_name, schema_name, action) if {
trino_catalog := catalog_config_by_name[catalog_name]
namespace_name := namespace_for_schema(schema_name)
lakekeeper.require_namespace_access_simple(
trino_catalog.lakekeeper_id,
trino_catalog.lakekeeper_warehouse,
namespace_name,
lakekeeper_user_id,
action,
)
}
require_schema_access_create(catalog_name, schema_name, action, properties, name) if {
trino_catalog := catalog_config_by_name[catalog_name]
namespace_name := namespace_for_schema(schema_name)
lakekeeper.require_namespace_access_create(
trino_catalog.lakekeeper_id,
trino_catalog.lakekeeper_warehouse,
namespace_name,
lakekeeper_user_id,
action,
properties,
name,
)
}
# Not used yet
require_schema_access_update_properties(catalog_name, schema_name, removed_properties, updated_properties) if {
trino_catalog := catalog_config_by_name[catalog_name]
namespace_name := namespace_for_schema(schema_name)
lakekeeper.require_namespace_access_update_properties(
trino_catalog.lakekeeper_id,
trino_catalog.lakekeeper_warehouse,
namespace_name,
lakekeeper_user_id,
removed_properties,
updated_properties,
)
}
require_table_access_simple(catalog_name, schema_name, table_name, action) if {
trino_catalog := catalog_config_by_name[catalog_name]
namespace_name := namespace_for_schema(schema_name)
lakekeeper.require_table_access_simple(
trino_catalog.lakekeeper_id,
trino_catalog.lakekeeper_warehouse,
namespace_name,
table_name,
lakekeeper_user_id,
action,
)
}
require_table_access_commit(catalog_name, schema_name, table_name, updated_properties, removed_properties) if {
trino_catalog := catalog_config_by_name[catalog_name]
namespace_name := namespace_for_schema(schema_name)
lakekeeper.require_table_access_commit(
trino_catalog.lakekeeper_id,
trino_catalog.lakekeeper_warehouse,
namespace_name,
table_name,
lakekeeper_user_id,
updated_properties,
removed_properties,
)
}
require_view_access_simple(catalog_name, schema_name, view_name, action) if {
trino_catalog := catalog_config_by_name[catalog_name]
namespace_name := namespace_for_schema(schema_name)
lakekeeper.require_view_access_simple(
trino_catalog.lakekeeper_id,
trino_catalog.lakekeeper_warehouse,
namespace_name,
view_name,
lakekeeper_user_id,
action,
)
}
child_schema_name(schema_name) := child_name if {
parts := split(schema_name, ".")
child_name := parts[count(parts) - 1]
}
is_metadata_table(table_name) if {
table_name_suffixes := [
"$properties", "$history", "$metadata_log_entries",
"$snapshots", "$manifests", "$all_manifests",
"$partitions", "$files", "$entries", "$all_entries", "$refs",
]
some suffix in table_name_suffixes
endswith(table_name, suffix)
} else := false
split_metadata_table_name(table_name) := base_table_name if {
table_name_suffixes := [
"$properties", "$history", "$metadata_log_entries",
"$snapshots", "$manifests", "$all_manifests",
"$partitions", "$files", "$entries", "$all_entries", "$refs",
]
some suffix in table_name_suffixes
endswith(table_name, suffix)
base_table_name := substring(table_name, 0, count(table_name) - count(suffix))
}
@@ -0,0 +1,143 @@
# Optimized batch filtering for Lakekeeper-managed catalogs.
# Instead of calling `allow` per resource (which makes 1+ HTTP calls each),
# this collects all checks into a single Lakekeeper batch-check HTTP request.
package trino
import data.lakekeeper
# Lakekeeper action for each Trino batch operation
_batch_lakekeeper_actions := {
"FilterTables": "get_metadata",
"FilterColumns": "get_metadata",
"SelectFromColumns": "read_data",
}
# _managed_catalog_names is defined in main.rego
# --- Table/View batch ---
# Collect non-system-schema table resource indices for managed catalogs.
_lakekeeper_batch_table_indices contains i if {
input.action.operation in ["FilterTables", "FilterColumns", "SelectFromColumns"]
some i, raw_resource in input.action.filterResources
raw_resource.table.catalogName in _managed_catalog_names
not raw_resource.table.schemaName in lakekeeper_system_schemas
not is_metadata_table(raw_resource.table.tableName)
}
# Build checks, execute batch-check, and return allowed indices per catalog.
# Each resource generates two checks (table + view) since Trino doesn't distinguish.
# Checks are automatically chunked to stay within Lakekeeper's batch-check limit.
# regal ignore:rule-length
_lakekeeper_batch_allowed[catalog_name] := allowed_indices if {
input.action.operation in ["FilterTables", "FilterColumns", "SelectFromColumns"]
some catalog_name in _managed_catalog_names
action := _batch_lakekeeper_actions[input.action.operation]
trino_catalog := catalog_config_by_name[catalog_name]
warehouse_id := lakekeeper.warehouse_id_for_name(trino_catalog.lakekeeper_id, trino_catalog.lakekeeper_warehouse)
# Build an ordered list of (index, resource) from the input array (deterministic order).
# Sets are unordered in Rego, so we iterate the array directly to guarantee
# that checks and ordered_indices are aligned.
catalog_resources := [{"idx": i, "res": raw_resource} |
some i, raw_resource in input.action.filterResources
i in _lakekeeper_batch_table_indices
raw_resource.table.catalogName == catalog_name
]
checks := [check |
some entry in catalog_resources
namespace := namespace_for_schema(entry.res.table.schemaName)
some check in [
lakekeeper.build_table_check(warehouse_id, namespace, entry.res.table.tableName, lakekeeper_user_id, action),
lakekeeper.build_view_check(warehouse_id, namespace, entry.res.table.tableName, lakekeeper_user_id, action),
]
]
count(checks) > 0
ordered_indices := [entry.idx |
some entry in catalog_resources
some _ in [0, 1] # two checks per resource (table + view)
]
results := lakekeeper.batch_check_results(trino_catalog.lakekeeper_id, checks)
# A resource is allowed if ANY of its checks (table or view) returned true.
# Each resource has two consecutive checks (table at j, view at j+1).
allowed_indices := {idx |
some j, idx in ordered_indices
j % 2 == 0 # only process even positions (first of each pair)
some type_offset in [0, 1]
results[j + type_offset].allowed == true
}
}
# --- Schema batch ---
_lakekeeper_batch_schema_indices contains i if {
input.action.operation == "FilterSchemas"
some i, raw_resource in input.action.filterResources
raw_resource.schema.catalogName in _managed_catalog_names
not raw_resource.schema.schemaName in lakekeeper_system_schemas
}
# regal ignore:rule-length
_lakekeeper_batch_schema_allowed[catalog_name] := allowed_indices if {
input.action.operation == "FilterSchemas"
some catalog_name in _managed_catalog_names
trino_catalog := catalog_config_by_name[catalog_name]
warehouse_id := lakekeeper.warehouse_id_for_name(trino_catalog.lakekeeper_id, trino_catalog.lakekeeper_warehouse)
catalog_resources := [{"idx": i, "res": raw_resource} |
some i, raw_resource in input.action.filterResources
i in _lakekeeper_batch_schema_indices
raw_resource.schema.catalogName == catalog_name
]
checks := [check |
some entry in catalog_resources
namespace := namespace_for_schema(entry.res.schema.schemaName)
check := lakekeeper.build_namespace_check(warehouse_id, namespace, lakekeeper_user_id, "get_metadata")
]
count(checks) > 0
ordered_indices := [entry.idx |
some entry in catalog_resources
]
results := lakekeeper.batch_check_results(trino_catalog.lakekeeper_id, checks)
allowed_indices := {idx |
some j, idx in ordered_indices
results[j].allowed == true
}
}
# --- Batch rules (grouped together to avoid messy-rule) ---
batch contains i if {
some catalog_name in _managed_catalog_names
some i in _lakekeeper_batch_allowed[catalog_name]
}
batch contains i if {
some catalog_name in _managed_catalog_names
some i in _lakekeeper_batch_schema_allowed[catalog_name]
}
# FilterColumns with a single table + columns array on a managed catalog:
# Lakekeeper authorizes at table level, so we check the table once via the
# batch path and return all column indices if allowed — no per-column evaluation.
batch contains i if {
input.action.operation == "FilterColumns"
count(input.action.filterResources) == 1
raw_resource := input.action.filterResources[0]
count(raw_resource.table.columns) > 0
raw_resource.table.catalogName in _managed_catalog_names
not raw_resource.table.schemaName in lakekeeper_system_schemas
not is_metadata_table(raw_resource.table.tableName)
0 in _lakekeeper_batch_allowed[raw_resource.table.catalogName]
some i in numbers.range(0, count(raw_resource.table.columns) - 1)
}
@@ -0,0 +1,150 @@
package trino
import data.configuration
# METADATA
# entrypoint: true
default allow := false
default allow_managed := false
default allow_unmanaged := false
# Blanket allow for unmanaged catalogs when enabled via env var or configuration.
# This is useful when Trino has multiple authorizers and this OPA bridge
# should not block access to catalogs managed by other authorizers.
allow_unmanaged if {
configuration.trino_allow_unmanaged_catalogs == true
_resource_catalog_name
not _resource_catalog_name in _managed_catalog_names
}
# Extract catalog name from the current resource (works for all resource types)
_resource_catalog_name := input.action.resource.catalog.name
_resource_catalog_name := input.action.resource.table.catalogName
_resource_catalog_name := input.action.resource.schema.catalogName
_resource_catalog_name := input.action.resource.function.catalogName
# Pre-compute managed catalog names (evaluated once)
_managed_catalog_names contains cat.name if {
some cat in configuration.trino_catalog
}
# --- allow rules ---
# Default access (system catalog, ExecuteQuery, etc.) - always applies
# regal ignore:messy-rule
allow if {
allow_default_access
}
# Managed catalog rules (Lakekeeper)
allow if {
allow_catalog
}
allow if {
allow_schema
}
allow if {
allow_table
}
allow if {
allow_view
}
# Extension point for managed catalogs.
# Create policies/trino/allow_managed.rego with rules that set allow_managed to true.
allow if {
allow_managed
}
# Extension point for catalogs not listed in configuration.trino_catalog.
# Create policies/trino/allow_unmanaged.rego with rules that set allow_unmanaged to true.
# When TRINO_ALLOW_UNMANAGED_CATALOGS=true, all access to unmanaged catalogs is permitted.
allow if {
allow_unmanaged
}
# --- batch rules ---
# Operations with dedicated batch handling via check_batch.rego
_batch_operations := {"FilterTables", "FilterColumns", "SelectFromColumns", "FilterSchemas"}
# Extract catalog name from a batch resource
_batch_resource_catalog(raw_resource) := raw_resource.table.catalogName
_batch_resource_catalog(raw_resource) := raw_resource.schema.catalogName
_batch_resource_catalog(raw_resource) := raw_resource.catalog.name
# Extract schema name from a batch resource
_batch_resource_schema(raw_resource) := raw_resource.table.schemaName
_batch_resource_schema(raw_resource) := raw_resource.schema.schemaName
# Fast path for unmanaged catalogs: only default_access + allow_unmanaged
_allow_unmanaged if allow_default_access
_allow_unmanaged if allow_unmanaged
# Unmanaged catalogs in batch operations: fast path (no Lakekeeper evaluation)
batch contains i if {
input.action.operation in _batch_operations
some i, raw_resource in input.action.filterResources
not _batch_resource_catalog(raw_resource) in _managed_catalog_names
# regal ignore:with-outside-test-context
_allow_unmanaged with input.action.resource as raw_resource
}
# System schema resources in managed catalogs still need per-resource evaluation
# (information_schema, schema_discovery, system are excluded from Lakekeeper batch)
batch contains i if {
input.action.operation in _batch_operations
some i, raw_resource in input.action.filterResources
_batch_resource_catalog(raw_resource) in _managed_catalog_names
_batch_resource_schema(raw_resource) in lakekeeper_system_schemas
# regal ignore:with-outside-test-context
allow with input.action.resource as raw_resource
}
# Metadata tables in managed catalogs need per-resource evaluation
# (excluded from Lakekeeper batch since they resolve to base table permissions)
batch contains i if {
input.action.operation in {"FilterTables", "FilterColumns", "SelectFromColumns"}
some i, raw_resource in input.action.filterResources
_batch_resource_catalog(raw_resource) in _managed_catalog_names
not _batch_resource_schema(raw_resource) in lakekeeper_system_schemas
is_metadata_table(raw_resource.table.tableName)
# regal ignore:with-outside-test-context
allow with input.action.resource as raw_resource
}
# Non-batch operations: per-resource allow evaluation as before
batch contains i if {
not input.action.operation in _batch_operations
some i, raw_resource in input.action.filterResources
# regal ignore:with-outside-test-context
allow with input.action.resource as raw_resource
}
# Corner case: filtering columns is done with a single table item, and many columns inside
# We cannot use our normal logic in other parts of the policy as they are based on sets
# and we need to retain order
batch contains i if {
input.action.operation == "FilterColumns"
count(input.action.filterResources) == 1
raw_resource := input.action.filterResources[0]
count(raw_resource.table.columns) > 0
new_resources := [
object.union(raw_resource, {"table": object.union(raw_resource.table, {"column": column_name})}) |
some column_name in raw_resource.table.columns
]
some i, resource in new_resources
# regal ignore:with-outside-test-context
allow with input.action.resource as resource
}
@@ -0,0 +1,5 @@
package trino
# regal ignore:rule-name-repeats-package
trino_user_id := input.context.identity.user
lakekeeper_user_id := concat("", ["oidc~", trino_user_id])