This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new f12bcdbb8a [python] Add permission endpoints to the REST mock server
(#10503)
f12bcdbb8a is described below
commit f12bcdbb8a9de73e494c0652493db1f4231b6716
Author: Jiajia Li <[email protected]>
AuthorDate: Fri Oct 16 21:57:32 2026 +0800
[python] Add permission endpoints to the REST mock server (#10503)
---
.../tests/rest/rest_permission_server_test.py | 219 ++++++++++++++++
paimon-python/pypaimon/tests/rest/rest_server.py | 278 ++++++++++++++++++++-
.../pypaimon/tests/sql_permission_test.py | 161 ++++++++++++
3 files changed, 648 insertions(+), 10 deletions(-)
diff --git a/paimon-python/pypaimon/tests/rest/rest_permission_server_test.py
b/paimon-python/pypaimon/tests/rest/rest_permission_server_test.py
new file mode 100644
index 0000000000..51db1276d6
--- /dev/null
+++ b/paimon-python/pypaimon/tests/rest/rest_permission_server_test.py
@@ -0,0 +1,219 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.api.api_response import ConfigResponse
+from pypaimon.api.auth import BearTokenAuthProvider
+from pypaimon.api.rest_exception import (AlreadyExistsException,
BadRequestException,
+ NoSuchResourceException)
+from pypaimon.catalog.catalog_exception import IllegalArgumentError,
TableNoPermissionException
+from pypaimon.common.identifier import Identifier
+from pypaimon.management.list_permissions_request import ListPermissionsRequest
+from pypaimon.management.permission_assignment import PermissionAssignment
+from pypaimon.management.permission_columns import PermissionColumns
+from pypaimon.management.permission_resource import PermissionResource
+from pypaimon.management.resource_type import ResourceType
+from pypaimon.schema.schema_change import SchemaChange
+from pypaimon.tests.rest.rest_function_test import _mock_function
+from pypaimon.tests.rest.rest_server import RESTCatalogServer
+
+USERS = Identifier.create('default', 'users')
+MEMBERS = Identifier.create('default', 'members')
+COLUMNS = PermissionResource(ResourceType.COLUMN, 'default', 'users')
+ARROW_SCHEMA = pa.schema([('id', pa.int64()), ('email', pa.string()),
('phone', pa.string())])
+
+
+def _schema(query_auth=True):
+ return Schema.from_pyarrow_schema(ARROW_SCHEMA,
options={'query-auth.enabled': 'true'} if query_auth else {})
+
+
[email protected]
+def rest(tmp_path):
+ server = RESTCatalogServer(str(tmp_path), BearTokenAuthProvider('token'),
+ ConfigResponse(defaults={'prefix': 'mock'}),
'warehouse')
+ server.start()
+ try:
+ catalog = CatalogFactory.create({
+ 'metastore': 'rest', 'uri': server.get_url(), 'warehouse':
'warehouse',
+ 'token.provider': 'bear', 'token': 'token', 'data-token.enabled':
'false'})
+ catalog.create_database('default', False)
+ catalog.create_table('default.users', _schema(), False)
+ yield catalog, server
+ finally:
+ server.shutdown()
+
+
+def _table(name):
+ return PermissionResource(ResourceType.TABLE, 'default', name)
+
+
+def _grants(catalog, name, resource_type=ResourceType.TABLE):
+ grants = catalog.permission_management().list_permissions(
+ ListPermissionsRequest(resource_type, 'default', name)).elements
+ return [(grant.get_access(), grant.get_principal()) for grant in grants]
+
+
+def _column_grants(catalog, name='users'):
+ grants = catalog.permission_management().list_permissions(
+ ListPermissionsRequest(ResourceType.COLUMN, 'default', name)).elements
+ return [(grant.get_columns().get_column_names(),
grant.get_columns().get_excluded_column_names())
+ for grant in grants]
+
+
+def test_table_and_column_grants_follow_rename_and_drop(rest):
+ catalog, server = rest
+ permissions = catalog.permission_management()
+ permissions.grant_permission(PermissionAssignment(_table('users'),
'SELECT', 'analyst'))
+ permissions.grant_permission(PermissionAssignment(
+ COLUMNS, 'SELECT', 'analyst',
PermissionColumns(excluded_column_names=['email'])))
+ server.set_query_principals('analyst')
+
+ catalog.rename_table('default.users', 'default.members')
+ assert _grants(catalog, 'members') == [('SELECT', 'analyst')]
+ assert _column_grants(catalog, 'members') == [(None, ('email',))]
+ with pytest.raises(TableNoPermissionException):
+ catalog.auth_table_query(MEMBERS, ['email'])
+ catalog.create_table('default.users', _schema(), False)
+ assert [_grants(catalog, 'users'), _column_grants(catalog)] == [[], []]
+ catalog.auth_table_query(USERS, ['email'])
+
+ catalog.drop_table('default.members')
+ catalog.create_table('default.members', _schema(), False)
+ assert [_grants(catalog, 'members'), _column_grants(catalog, 'members')]
== [[], []]
+ catalog.auth_table_query(MEMBERS, ['email'])
+
+
+def test_dropping_a_database_or_function_removes_its_grants(rest):
+ catalog, server = rest
+ area = Identifier.create('default', 'area')
+ catalog.create_function(area, _mock_function(area), False)
+ catalog.create_database('sales', False)
+ for resource, access in ((PermissionResource(ResourceType.DATABASE,
'sales'), 'DESCRIBE'),
+ (PermissionResource(ResourceType.FUNCTION,
'default', function='area'), 'SELECT'),
+ (PermissionResource(ResourceType.DATABASE,
'default'), 'DESCRIBE')):
+
catalog.permission_management().grant_permission(PermissionAssignment(resource,
access, 'analyst'))
+
+ catalog.drop_database('sales')
+ catalog.drop_function(area, False)
+ assert [(grant.get_resource().get_type(),
grant.get_resource().get_database())
+ for grant in server.permission_store.values()] ==
[(ResourceType.DATABASE, 'default')]
+
+
[email protected]('expire_time, expired', [
+ (None, False),
+ ('2099-01-01T00:00:00.123000000Z', False),
+ ('2099-01-01t00:00:00.5z', False),
+ ('+12000-01-01T00:00:00Z', False),
+ ('2020-01-01T24:00:00.000000000Z', True),
+ ('-0001-01-01T00:00:00.999Z', True),
+])
+def test_column_grants_limit_the_selected_columns_until_they_expire(rest,
expire_time, expired):
+ catalog, server = rest
+ catalog.permission_management().grant_permission(PermissionAssignment(
+ COLUMNS, 'SELECT', 'analyst',
PermissionColumns(excluded_column_names=['email']), expire_time))
+
+ server.set_query_principals('analyst')
+ catalog.auth_table_query(USERS, ['id'])
+ for select in (['email'], None):
+ if expired:
+ catalog.auth_table_query(USERS, select)
+ else:
+ with pytest.raises(TableNoPermissionException):
+ catalog.auth_table_query(USERS, select)
+
+
[email protected]('columns', [PermissionColumns(column_names=['id']),
+
PermissionColumns(excluded_column_names=['email', 'phone'])])
+def test_column_grants_restrict_only_their_principal(rest, columns):
+ catalog, server = rest
+
catalog.permission_management().grant_permission(PermissionAssignment(COLUMNS,
'SELECT', 'analyst', columns))
+
+ server.set_query_principals('auditor')
+ catalog.auth_table_query(USERS, None)
+ server.set_query_principals('analyst')
+ catalog.auth_table_query(USERS, ['id'])
+ for select in (['email'], ['id', 'phone'], None):
+ with pytest.raises(TableNoPermissionException):
+ catalog.auth_table_query(USERS, select)
+
+
+def test_column_grants_need_query_auth_and_existing_columns(rest):
+ catalog, _ = rest
+ catalog.create_table('default.plain', _schema(query_auth=False), False)
+ with pytest.raises(AlreadyExistsException):
+ catalog.permission_management().grant_permission(PermissionAssignment(
+ PermissionResource(ResourceType.COLUMN, 'default', 'plain'),
'SELECT', 'analyst',
+ PermissionColumns(column_names=['id'])))
+ with pytest.raises(NoSuchResourceException):
+ catalog.permission_management().grant_permission(PermissionAssignment(
+ COLUMNS, 'SELECT', 'analyst',
PermissionColumns(column_names=['id', 'mobile'])))
+ assert [_column_grants(catalog, 'plain'), _column_grants(catalog)] == [[],
[]]
+
+ catalog.create_table('default.flagged', Schema.from_pyarrow_schema(
+ ARROW_SCHEMA, options={'query-auth.enabled': True}), False)
+ catalog.permission_management().grant_permission(PermissionAssignment(
+ PermissionResource(ResourceType.COLUMN, 'default', 'flagged'),
'SELECT', 'analyst',
+ PermissionColumns(column_names=['id'])))
+
+
+def test_column_grants_follow_column_renames(rest):
+ catalog, _ = rest
+ catalog.permission_management().grant_permission(PermissionAssignment(
+ COLUMNS, 'SELECT', 'analyst', PermissionColumns(column_names=['id',
'email'])))
+
+ catalog.alter_table('default.users', [SchemaChange.rename_column('email',
'contact')])
+ catalog.alter_table('default.users', [SchemaChange.drop_column('id')])
+ for change in (SchemaChange.drop_column('contact'),
SchemaChange.set_option('query-auth.enabled', 'false')):
+ with pytest.raises(BadRequestException):
+ catalog.alter_table('default.users', [change])
+
+ table = catalog.get_table('default.users')
+ for schema in (table.table_schema, table.schema_manager.latest()):
+ assert [field.name for field in schema.fields] == ['contact', 'phone']
+ assert schema.options['query-auth.enabled'] == 'true'
+ assert _column_grants(catalog) == [(('contact',), None)]
+
+
+def test_permission_listings_are_paged(rest):
+ catalog, _ = rest
+ for principal in ('analyst', 'auditor'):
+
catalog.permission_management().grant_permission(PermissionAssignment(_table('users'),
'SELECT', principal))
+
+ first = catalog.permission_management().list_permissions(
+ ListPermissionsRequest(ResourceType.TABLE, 'default', 'users',
max_results=1))
+ second =
catalog.permission_management().list_permissions(ListPermissionsRequest(
+ ResourceType.TABLE, 'default', 'users',
page_token=first.next_page_token, max_results=1))
+ assert [len(first.elements), len(second.elements), second.next_page_token]
== [1, 1, None]
+ assert [first.elements[0].get_principal(),
second.elements[0].get_principal()] == ['analyst', 'auditor']
+
+
+def test_tables_with_branches_cannot_have_column_grants(rest):
+ catalog, _ = rest
+ grant = PermissionAssignment(COLUMNS, 'SELECT', 'analyst',
PermissionColumns(excluded_column_names=['email']))
+ catalog.create_branch(USERS, 'b1')
+ with pytest.raises(BadRequestException):
+ catalog.permission_management().grant_permission(grant)
+ assert _column_grants(catalog) == []
+
+ catalog.drop_table('default.users')
+ catalog.create_table('default.users', _schema(), False)
+ catalog.permission_management().grant_permission(grant)
+ with pytest.raises(IllegalArgumentError):
+ catalog.create_branch(USERS, 'b1')
diff --git a/paimon-python/pypaimon/tests/rest/rest_server.py
b/paimon-python/pypaimon/tests/rest/rest_server.py
index f8ce233d69..db5a88c0a2 100755
--- a/paimon-python/pypaimon/tests/rest/rest_server.py
+++ b/paimon-python/pypaimon/tests/rest/rest_server.py
@@ -34,8 +34,10 @@ from pypaimon.api.api_request import (AlterDatabaseRequest,
AlterTableRequest,
CreateDatabaseRequest,
CreatePartitionsRequest,
CreateTableRequest, CreateTagRequest,
+ GrantPermissionRequest,
RenameBranchRequest,
- RenameTableRequest)
+ RenameTableRequest,
+ RevokePermissionRequest)
from pypaimon.api.api_response import (ConfigResponse,
CreatePartitionsResponse,
GetDatabaseResponse,
GetFunctionResponse,
@@ -45,7 +47,8 @@ from pypaimon.api.api_response import (ConfigResponse,
CreatePartitionsResponse,
ListFunctionDetailsResponse,
ListFunctionsGloballyResponse,
ListFunctionsResponse,
- ListPartitionsResponse,
ListTablesResponse,
+ ListPartitionsResponse,
+ ListPermissionsResponse,
ListTablesResponse,
ListTagsResponse,
PagedList, Partition,
RESTResponse, ErrorResponse)
@@ -68,6 +71,13 @@ from pypaimon.catalog.rest.table_metadata import
TableMetadata
from pypaimon.common.identifier import Identifier
from pypaimon.api.typedef import RESTAuthParameter
from pypaimon.common.json_util import JSON
+from pypaimon.common.options.core_options import CoreOptions
+from pypaimon.management.permission_access import PermissionAccess
+from pypaimon.management.permission_assignment import (PermissionAssignment,
_INSTANT,
+ _days_from_civil)
+from pypaimon.management.permission_columns import PermissionColumns
+from pypaimon.management.permission_resource import PermissionResource
+from pypaimon.management.resource_type import ResourceType
from pypaimon import Schema
from pypaimon.schema.schema_change import Actions, SchemaChange
from pypaimon.schema.schema_manager import SchemaManager
@@ -198,6 +208,66 @@ def _dict_to_schema_change(change_dict: dict) ->
SchemaChange:
raise ValueError(f"Unknown schema change action: {action}")
+def _cursor_key(*parts: Optional[str]) -> str:
+ return "".join("-1:" if part is None else "{}:{}".format(len(part), part)
for part in parts)
+
+
+def _permission_key(resource: PermissionResource, access: str, principal: str)
-> str:
+ return _cursor_key(resource.get_type().name, resource.get_database(),
resource.get_table(),
+ resource.get_function(), resource.get_view(),
+ PermissionAccess.canonicalize_for(resource, access),
principal)
+
+
+def _assignment_key(assignment: PermissionAssignment) -> str:
+ return _permission_key(
+ assignment.get_resource(), assignment.get_access(),
assignment.get_principal())
+
+
+def _resource_name(resource: PermissionResource) -> str:
+ name = resource.get_table() or resource.get_function() or
resource.get_view()
+ return ".".join(part for part in (resource.get_database(), name) if part)
or "catalog"
+
+
+def _granted_columns(assignment: PermissionAssignment) -> List[str]:
+ columns = assignment.get_columns()
+ names = columns.get_column_names()
+ return list(names if names is not None else
columns.get_excluded_column_names())
+
+
+def _expired(assignment: PermissionAssignment) -> bool:
+ expire_time = assignment.get_expire_time()
+ if expire_time is None:
+ return False
+ sign, digits, month, day, hour, minute, second, fraction =
_INSTANT.fullmatch(expire_time).groups()
+ hour, minute, second = int(hour), int(minute), int(second)
+ days = _days_from_civil(-int(digits) if sign == "-" else int(digits),
int(month), int(day))
+ if hour == 24:
+ days, hour = days + 1, 0
+ second = min(second, 59)
+ epoch_millis = ((days * 86400 + hour * 3600 + minute * 60 + second) * 1000
+ + int((fraction or "").ljust(9, "0")) // 1000000)
+ return epoch_millis <= int(time.time() * 1000)
+
+
+def _renamed_columns(previous: TableSchema, current: TableSchema) -> Dict[str,
Optional[str]]:
+ current_names = {field.id: field.name for field in current.fields}
+ return {field.name: current_names.get(field.id) for field in
previous.fields}
+
+
+def _query_auth_enabled(schema: TableSchema) -> bool:
+ return CoreOptions.from_dict(schema.options).query_auth_enabled
+
+
+def _on_table(resource: PermissionResource, identifier: Identifier) -> bool:
+ return resource.get_type() in (ResourceType.TABLE, ResourceType.COLUMN) \
+ and resource.get_database() == identifier.get_database_name() \
+ and resource.get_table() == identifier.get_table_name()
+
+
+def _matches(parameters: Dict[str, str], name: str, value: Optional[str]) ->
bool:
+ return name not in parameters or parameters[name] == value
+
+
class RESTCatalogServer:
"""Mock REST server for testing"""
@@ -226,6 +296,8 @@ class RESTCatalogServer:
self.no_permission_tables: List[str] = []
self.table_token_store: Dict[str, "RESTToken"] = {}
self.table_query_auth_store = {}
+ self.permission_store: Dict[str, PermissionAssignment] = {}
+ self.query_principals: set = set()
# Initialize mock catalog (simplified)
self.data_path = data_path
@@ -399,12 +471,18 @@ class RESTCatalogServer:
self.table_latest_snapshot_store,
self.table_partitions_store):
if source_name in store:
store[destination_name] = store.pop(source_name)
+ self._rename_table_access(source_table, destination_table)
return self._mock_response("", 200)
# Global functions endpoint (catalog-scoped)
if resource_path == self.resource_paths.functions() and method ==
"GET":
return self._functions_globally_handle(parameters)
+ if resource_path in (self.resource_paths.permissions(),
+ self.resource_paths.grant_permission(),
+ self.resource_paths.revoke_permission()):
+ return self._permissions_handle(method, resource_path, data,
parameters)
+
database = resource_path.split("/")[4]
# Database-specific endpoints
if
resource_path.startswith(self.resource_paths.database(database)):
@@ -591,6 +669,133 @@ class RESTCatalogServer:
return self._mock_response(ErrorResponse(None, None, "Not Found",
404), 404)
return self._mock_response(ErrorResponse(None, None, "Not Found",
404), 404)
+ def set_query_principals(self, *principals: str) -> None:
+ self.query_principals = set(principals)
+
+ def _permissions_handle(self, method: str, resource_path: str, data: str,
+ parameters: Dict[str, str]) -> Tuple[str, int]:
+ if method == "GET" and resource_path ==
self.resource_paths.permissions():
+ target = PermissionResource(
+ parameters.get("resourceType"), parameters.get("database"),
+ parameters.get("table"), parameters.get("function"),
parameters.get("view"))
+ missing = self._missing_permission_resource(target)
+ if missing is not None:
+ return missing
+ matched = [assignment for assignment in
self.permission_store.values()
+ if assignment.get_resource() == target
+ and _matches(parameters, "principal",
assignment.get_principal())
+ and _matches(parameters, "access",
assignment.get_access())]
+ return self._mock_response(ListPermissionsResponse(
+ *self._management_page({_assignment_key(grant): grant for
grant in matched}, parameters)), 200)
+ if method == "POST" and resource_path ==
self.resource_paths.grant_permission():
+ assignment = JSON.from_json(data,
GrantPermissionRequest).assignment()
+ missing =
self._missing_permission_resource(assignment.get_resource())
+ if missing is not None:
+ return missing
+ invalid = self._invalid_column_grant(assignment)
+ if invalid is not None:
+ return invalid
+ self.permission_store[_assignment_key(assignment)] = assignment
+ return self._mock_response("", 200)
+ if method == "POST" and resource_path ==
self.resource_paths.revoke_permission():
+ request = JSON.from_json(data, RevokePermissionRequest)
+ missing = self._missing_permission_resource(request.get_resource())
+ if missing is not None:
+ return missing
+ self.permission_store.pop(_permission_key(
+ request.get_resource(), request.get_access(),
request.get_principal()), None)
+ return self._mock_response("", 200)
+ return self._mock_response(ErrorResponse(None, None, "Not Found",
404), 404)
+
+ def _missing_permission_resource(
+ self, resource: PermissionResource) -> Optional[Tuple[str, int]]:
+ resource_type = resource.get_type()
+ database = resource.get_database()
+ if resource_type in (ResourceType.CATALOG, ResourceType.CATALOG_ALL):
+ return None
+ exists = database in self.database_store
+ if resource_type in (ResourceType.TABLE, ResourceType.COLUMN):
+ exists = exists and Identifier.create(
+ database, resource.get_table()).get_full_name() in
self.table_metadata_store
+ elif resource_type == ResourceType.FUNCTION:
+ exists = exists and "{}.{}".format(
+ database, resource.get_function()) in self.function_store
+ elif resource_type == ResourceType.VIEW:
+ exists = False
+ if exists:
+ return None
+ return self._mock_response(ErrorResponse(
+ resource_type.name, _resource_name(resource),
+ "Permission resource does not exist.", 404), 404)
+
+ def _invalid_column_grant(self, assignment: PermissionAssignment) ->
Optional[Tuple[str, int]]:
+ resource = assignment.get_resource()
+ if resource.get_type() != ResourceType.COLUMN:
+ return None
+ identifier = Identifier.create(resource.get_database(),
resource.get_table())
+ schema = self.table_metadata_store[identifier.get_full_name()].schema
+ if self.branch_store.get(identifier.get_full_name()):
+ return self._mock_response(ErrorResponse(
+ ErrorResponse.RESOURCE_TYPE_TABLE, _resource_name(resource),
+ "Column permissions are not supported on tables with
branches.", 400), 400)
+ if not _query_auth_enabled(schema):
+ return self._mock_response(ErrorResponse(
+ ErrorResponse.RESOURCE_TYPE_TABLE, _resource_name(resource),
+ "Column permissions require query-auth.enabled=true.", 409),
409)
+ names = {field.name for field in schema.fields}
+ for column in _granted_columns(assignment):
+ if column not in names:
+ return self._mock_response(ErrorResponse(
+ ErrorResponse.RESOURCE_TYPE_COLUMN, column,
+ "Permission column does not exist.", 404), 404)
+ return None
+
+ def _restricted(self, identifier: Identifier) -> bool:
+ return bool(self._column_grants(identifier))
+
+ def _column_grants(self, identifier: Identifier) ->
List[PermissionAssignment]:
+ return [assignment for assignment in self.permission_store.values()
+ if assignment.get_resource().get_type() == ResourceType.COLUMN
+ and _on_table(assignment.get_resource(), identifier)]
+
+ def _remove_permissions(self, matches) -> None:
+ for key in [key for key, assignment in self.permission_store.items()
+ if matches(assignment.get_resource())]:
+ del self.permission_store[key]
+
+ def _drop_table_access(self, identifier: Identifier) -> None:
+ for branch in self.branch_store.pop(identifier.get_full_name(), ()):
+ manager = self._schema_manager(identifier, branch)
+ manager.file_io.delete(manager.branch_path, recursive=True)
+ self._remove_permissions(lambda resource: _on_table(resource,
identifier))
+
+ def _rename_table_access(self, source: Identifier, destination:
Identifier) -> None:
+ moved = [assignment for assignment in self.permission_store.values()
+ if _on_table(assignment.get_resource(), source)]
+ self._remove_permissions(lambda resource: _on_table(resource, source))
+ for assignment in moved:
+ renamed = PermissionAssignment(
+ PermissionResource(assignment.get_resource().get_type(),
+ destination.get_database_name(),
destination.get_table_name()),
+ assignment.get_access(), assignment.get_principal(),
+ assignment.get_columns(), assignment.get_expire_time())
+ self.permission_store[_assignment_key(renamed)] = renamed
+
+ def _check_column_grants(self, identifier: Identifier, select:
Optional[List[str]]) -> None:
+ fields = [field.name for field in
self.table_metadata_store[identifier.get_full_name()].schema.fields]
+ grants = [grant for grant in self._column_grants(identifier)
+ if grant.get_principal() in self.query_principals and not
_expired(grant)]
+ if grants:
+ included = set(fields)
+ for grant in grants:
+ columns = set(_granted_columns(grant))
+ if grant.get_columns().get_column_names() is not None:
+ included &= columns
+ else:
+ included -= columns
+ if any(column not in included for column in (fields if select is
None else select)):
+ raise TableNoPermissionException(identifier)
+
# ======================= Function Handlers ===============================
def _functions_handle(self, method: str, data: str, database_name: str,
@@ -651,6 +856,10 @@ class RESTCatalogServer:
if key not in self.function_store:
raise FunctionNotExistException(identifier)
del self.function_store[key]
+ self._remove_permissions(
+ lambda resource: resource.get_type() == ResourceType.FUNCTION
+ and resource.get_database() == identifier.get_database_name()
+ and resource.get_function() == identifier.get_object_name())
return self._mock_response("", 200)
return self._mock_response(ErrorResponse(None, None, "Method Not
Allowed", 405), 405)
@@ -910,17 +1119,19 @@ class RESTCatalogServer:
if request.branch in store:
raise BranchAlreadyExistException(request.branch)
- schema_id =
self.table_metadata_store[identifier.get_full_name()].schema.id
+ metadata = self.table_metadata_store[identifier.get_full_name()]
+ schema_id, snapshot = metadata.schema.id, None
if request.from_tag is not None:
tags = self.tag_store.get(identifier.get_full_name(), {})
if request.from_tag not in tags:
raise TagNotExistException(request.from_tag)
snapshot = tags[request.from_tag].snapshot
- if snapshot is not None:
- schema_id = snapshot.schema_id
- self._write_snapshot_files(
- identifier, snapshot, None, request.branch)
+ schema_id = schema_id if snapshot is None else
snapshot.schema_id
+ if self._restricted(identifier):
+ raise ValueError("Tables with column permissions cannot have
branches.")
manager = self._schema_manager(identifier)
+ if snapshot is not None:
+ self._write_snapshot_files(identifier, snapshot, None,
request.branch)
target = manager.copy_with_branch(request.branch)
for schema in manager.list_all():
if schema.id <= schema_id:
@@ -1033,6 +1244,7 @@ class RESTCatalogServer:
elif method == "DELETE":
del self.database_store[database_name]
+ self._remove_permissions(lambda resource: resource.get_database()
== database_name)
return self._mock_response("", 200)
return self._mock_response(ErrorResponse(None, None, "Method Not
Allowed", 405), 405)
@@ -1091,6 +1303,7 @@ class RESTCatalogServer:
raise TableNotExistException(identifier)
else:
del self.table_metadata_store[identifier.get_full_name()]
+ self._drop_table_access(identifier)
if identifier.get_full_name() in self.table_latest_snapshot_store:
del
self.table_latest_snapshot_store[identifier.get_full_name()]
if identifier.get_full_name() in self.table_partitions_store:
@@ -1105,10 +1318,12 @@ class RESTCatalogServer:
return self._mock_response(ErrorResponse(None, None, "Method Not
Allowed", 405), 405)
if identifier.get_full_name() not in self.table_metadata_store:
raise TableNotExistException(identifier)
+ select = json.loads(data).get('select')
+ self._check_column_grants(identifier, select)
from pypaimon.api.api_response import AuthTableQueryResponse
response = self.table_query_auth_store.get(identifier.get_full_name(),
AuthTableQueryResponse())
if callable(response):
- response = response(json.loads(data).get('select'))
+ response = response(select)
return self._mock_response(response, 200)
def _table_token_handle(self, method: str, identifier: Identifier) ->
Tuple[str, int]:
@@ -1439,6 +1654,13 @@ class RESTCatalogServer:
raise ValueError(f"Invalid maxResults value:
{max_results_str}")
return DEFAULT_MAX_RESULTS
+ def _management_page(self, entities: Dict[str, Any],
+ parameters: Dict[str, str]) -> Tuple[List[Any],
Optional[str]]:
+ token = parameters.get(PAGE_TOKEN)
+ keys = sorted(key for key in entities if not token or key > token)
+ page = keys[:self._get_max_results(parameters)]
+ return [entities[key] for key in page], page[-1] if len(keys) >
len(page) else None
+
def _build_paged_entities(self, entities: List[Any], max_results: int,
page_token: Optional[str], desc: bool = False)
-> PagedList:
"""Build paged entities"""
@@ -1537,6 +1759,11 @@ class RESTCatalogServer:
schema_changes.append(change)
schema_manager = self._schema_manager(identifier, branch)
+ main = branch is None or branch == 'main'
+ base = Identifier.create(identifier.get_database_name(),
identifier.get_table_name())
+ if main:
+ self._validate_access_for_schema(
+ base, table_metadata,
schema_manager._generate_table_schema(table_metadata.schema, schema_changes))
new_schema = schema_manager.commit_changes(schema_changes)
updated_metadata = TableMetadata(
@@ -1544,10 +1771,41 @@ class RESTCatalogServer:
is_external=table_metadata.is_external,
uuid=table_metadata.uuid
)
- if branch is None or branch == 'main':
- base = Identifier.create(identifier.get_database_name(),
identifier.get_table_name())
+ if main:
+ self._evolve_column_grants(base, table_metadata.schema, new_schema)
self.table_metadata_store[base.get_full_name()] = updated_metadata
+ def _validate_access_for_schema(self, identifier: Identifier, metadata:
TableMetadata,
+ schema: TableSchema) -> None:
+ name = identifier.get_full_name()
+ grants = self._column_grants(identifier)
+ if grants and not _query_auth_enabled(schema):
+ raise ValueError(
+ "Cannot disable query-auth.enabled while table {} has column
permissions.".format(name))
+ renamed = _renamed_columns(metadata.schema, schema)
+ for grant in grants:
+ if grant.get_columns().get_column_names() is not None \
+ and not any(renamed.get(column) for column in
_granted_columns(grant)):
+ raise ValueError(
+ "Cannot drop every allowed column while table {} has
column permissions."
+ .format(name))
+
+ def _evolve_column_grants(self, identifier: Identifier, previous:
TableSchema,
+ current: TableSchema) -> None:
+ renamed = _renamed_columns(previous, current)
+ for grant in self._column_grants(identifier):
+ del self.permission_store[_assignment_key(grant)]
+ evolved = [renamed[column] for column in _granted_columns(grant)
if renamed.get(column)]
+ if not evolved:
+ continue
+ columns = (PermissionColumns(column_names=evolved)
+ if grant.get_columns().get_column_names() is not None
+ else PermissionColumns(excluded_column_names=evolved))
+ evolved_grant = PermissionAssignment(
+ grant.get_resource(), grant.get_access(),
grant.get_principal(),
+ columns, grant.get_expire_time())
+ self.permission_store[_assignment_key(evolved_grant)] =
evolved_grant
+
def _get_file_io(self):
"""Get FileIO instance for SchemaManager"""
from pypaimon.common.file_io import FileIO
diff --git a/paimon-python/pypaimon/tests/sql_permission_test.py
b/paimon-python/pypaimon/tests/sql_permission_test.py
new file mode 100644
index 0000000000..158f17f72a
--- /dev/null
+++ b/paimon-python/pypaimon/tests/sql_permission_test.py
@@ -0,0 +1,161 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.api.api_response import ConfigResponse
+from pypaimon.api.auth import BearTokenAuthProvider
+from pypaimon.catalog.catalog_exception import TableNoPermissionException
+from pypaimon.management.list_permissions_request import ListPermissionsRequest
+from pypaimon.management.permission_assignment import PermissionAssignment
+from pypaimon.management.permission_columns import PermissionColumns
+from pypaimon.management.permission_resource import PermissionResource
+from pypaimon.management.resource_type import ResourceType
+from pypaimon.read.native_plan import native_runtime_available
+from pypaimon.schema.schema_change import SchemaChange
+from pypaimon.tests.rest.rest_server import RESTCatalogServer
+
+
+def _rust_version():
+ if not native_runtime_available():
+ return 0, 0
+ from importlib.metadata import version
+ return tuple(int(part) for part in version('pypaimon-rust').split('.')[:2])
+
+
+RUST_VERSION = _rust_version()
+pytestmark = [pytest.mark.native_plan, pytest.mark.skipif(
+ RUST_VERSION < (0, 4), reason='pypaimon-rust 0.4 or later required')]
+
+TOKEN = 'test-token'
+ANALYST = 'analyst'
+ROWS = [
+ {'id': 1, 'region': 'eu', 'email': '[email protected]'},
+ {'id': 2, 'region': 'us', 'email': '[email protected]'},
+ {'id': 3, 'region': 'eu', 'email': '[email protected]'},
+]
+ORDERS = PermissionResource(ResourceType.TABLE, 'default', 'orders')
+
+
[email protected]
+def clients(tmp_path):
+ from pypaimon_rust.datafusion import SQLContext
+
+ server = RESTCatalogServer(str(tmp_path), BearTokenAuthProvider(TOKEN),
+ ConfigResponse(defaults={'prefix':
'permission-test'}),
+ 'warehouse')
+ server.start()
+ try:
+ options = {'metastore': 'rest', 'uri': server.get_url(), 'warehouse':
'warehouse',
+ 'token.provider': 'bear', 'token': TOKEN,
'data-token.enabled': 'false'}
+ catalog = CatalogFactory.create(options)
+ catalog.create_database('default', True)
+ context = SQLContext()
+ context.register_catalog('paimon', options)
+ yield catalog, context, server
+ finally:
+ server.shutdown()
+
+
+def _create_orders(catalog, rows=()):
+ schema = pa.schema([('id', pa.int64()), ('region', pa.string()), ('email',
pa.string())])
+ catalog.create_table('default.orders', Schema.from_pyarrow_schema(schema),
False)
+ if rows:
+ table = catalog.get_table('default.orders')
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.Table.from_pylist(list(rows), schema=schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ catalog.alter_table('default.orders',
[SchemaChange.set_option('query-auth.enabled', 'true')])
+
+
+def _sql(context, sql):
+ return [row for batch in context.sql(sql) for row in batch.to_pylist()]
+
+
+def _call_ok(context, sql):
+ assert _sql(context, sql) == [{'result': 'OK'}], sql
+
+
+def _python_rows(catalog, columns=('id', 'region', 'email')):
+ table = catalog.get_table('default.orders')
+ read_builder = table.new_read_builder().with_projection(list(columns))
+ splits = read_builder.new_scan().plan().splits()
+ return sorted(read_builder.new_read().to_arrow(splits).to_pylist(),
key=lambda row: row['id'])
+
+
+def _sql_rows(context, columns=('id', 'region', 'email')):
+ return _sql(context, 'SELECT {} FROM paimon.default.orders ORDER BY
id'.format(', '.join(columns)))
+
+
+def _grants(catalog, resource):
+ page =
catalog.permission_management().list_permissions(ListPermissionsRequest(
+ resource.get_type(), resource.get_database(), resource.get_table()))
+ return [(grant.get_access(), grant.get_principal(),
grant.get_expire_time(), grant.get_columns())
+ for grant in page.elements]
+
+
+def test_sql_grants_are_seen_by_pypaimon_and_back(clients):
+ catalog, context, _ = clients
+ _create_orders(catalog)
+ permissions = catalog.permission_management()
+
+ _call_ok(context, "CALL sys.grant_permission(resource_type => 'TABLE',
access => 'select', "
+ "principal => 'analyst', database => 'default', table =>
'orders', "
+ "expire_time => '2030-01-01T00:00:00Z')")
+ assert _grants(catalog, ORDERS) == [('SELECT', ANALYST,
'2030-01-01T00:00:00Z', None)]
+
+ columns = PermissionResource(ResourceType.COLUMN, 'default', 'orders')
+ permissions.grant_permission(PermissionAssignment(
+ columns, 'SELECT', ANALYST,
PermissionColumns(excluded_column_names=['email'])))
+ column_grant = {
+ 'resource_type': 'COLUMN', 'database': 'default', 'table': 'orders',
'function': None,
+ 'view': None, 'access': 'SELECT', 'principal': ANALYST,
'column_names': None,
+ 'excluded_column_names': 'email', 'expire_time': None,
'next_page_token': None}
+ assert _sql(context, "CALL sys.list_permissions(resource_type => 'COLUMN',
"
+ "database => 'default', table => 'orders')") ==
[column_grant]
+
+ _call_ok(context, "CALL sys.revoke_permission(resource_type => 'TABLE',
access => 'SELECT', "
+ "principal => 'analyst', database => 'default', table =>
'orders')")
+ permissions.revoke_permission(columns, 'SELECT', ANALYST)
+ assert _grants(catalog, ORDERS) == []
+ assert _sql(context, "CALL sys.list_permissions(resource_type => 'COLUMN',
"
+ "database => 'default', table => 'orders')") == []
+
+
[email protected](RUST_VERSION < (0, 5), reason='pypaimon-rust 0.5 or later
required')
+def test_column_grants_restrict_both_readers(clients):
+ catalog, context, server = clients
+ _create_orders(catalog, ROWS)
+ catalog.permission_management().grant_permission(PermissionAssignment(
+ PermissionResource(ResourceType.COLUMN, 'default', 'orders'),
'SELECT', ANALYST,
+ PermissionColumns(excluded_column_names=['email']),
'2099-01-01T00:00:00.123000000Z'))
+
+ server.set_query_principals(ANALYST)
+ allowed = [{'id': row['id'], 'region': row['region']} for row in ROWS]
+ assert _python_rows(catalog, ('id', 'region')) == allowed
+ assert _sql_rows(context, ('id', 'region')) == allowed
+ with pytest.raises(TableNoPermissionException):
+ _python_rows(catalog)
+ with pytest.raises(Exception, match='(?i)permission'):
+ _sql_rows(context)
