diff --git a/paimon-api/src/main/java/org/apache/paimon/management/ColumnMask.java b/paimon-api/src/main/java/org/apache/paimon/management/ColumnMask.java new file mode 100644 index 000000000000..c2dbadbbdb7b --- /dev/null +++ b/paimon-api/src/main/java/org/apache/paimon/management/ColumnMask.java @@ -0,0 +1,77 @@ +/* + * 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. + */ + +package org.apache.paimon.management; + +import org.apache.paimon.annotation.Experimental; + +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty; + +import java.beans.ConstructorProperties; +import java.nio.charset.StandardCharsets; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** Protected column and serialized Paimon transform for a column-mask policy. */ +@Experimental +@JsonIgnoreProperties(ignoreUnknown = true) +public class ColumnMask { + + public static final int MAX_TRANSFORM_BYTES = 60 * 1024; + + private static final String FIELD_ON_COLUMN = "onColumn"; + private static final String FIELD_TRANSFORM = "transform"; + + @JsonProperty(FIELD_ON_COLUMN) + private final String onColumn; + + @JsonProperty(FIELD_TRANSFORM) + private final String transform; + + @JsonCreator + @ConstructorProperties({FIELD_ON_COLUMN, FIELD_TRANSFORM}) + public ColumnMask( + @JsonProperty(FIELD_ON_COLUMN) String onColumn, + @JsonProperty(FIELD_TRANSFORM) String transform) { + checkArgument(!isBlank(onColumn), "onColumn cannot be empty."); + checkArgument(!isBlank(transform), "transform cannot be empty."); + checkArgument( + transform.getBytes(StandardCharsets.UTF_8).length <= MAX_TRANSFORM_BYTES, + "transform must not exceed %s UTF-8 bytes.", + MAX_TRANSFORM_BYTES); + this.onColumn = onColumn; + this.transform = transform; + } + + @JsonGetter(FIELD_ON_COLUMN) + public String getOnColumn() { + return onColumn; + } + + @JsonGetter(FIELD_TRANSFORM) + public String getTransform() { + return transform; + } + + private static boolean isBlank(String value) { + return value == null || value.trim().isEmpty(); + } +} diff --git a/paimon-api/src/main/java/org/apache/paimon/management/DataPolicy.java b/paimon-api/src/main/java/org/apache/paimon/management/DataPolicy.java new file mode 100644 index 000000000000..08fe3fb0357d --- /dev/null +++ b/paimon-api/src/main/java/org/apache/paimon/management/DataPolicy.java @@ -0,0 +1,120 @@ +/* + * 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. + */ + +package org.apache.paimon.management; + +import org.apache.paimon.annotation.Experimental; + +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonInclude; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty; + +import javax.annotation.Nullable; + +import java.beans.ConstructorProperties; + +import static org.apache.paimon.utils.Preconditions.checkArgument; +import static org.apache.paimon.utils.Preconditions.checkNotNull; + +/** + * Principal-scoped row-filter or column-mask policy attached to one table. + * + *

A principal has at most one row filter per table and at most one mask per table column. When + * policies are enforced, applicable row filters are combined with logical AND and multiple + * effective masks for one column fail closed. Invalid predicates or transforms also fail closed. + */ +@Experimental +@JsonIgnoreProperties(ignoreUnknown = true) +public class DataPolicy { + + private static final String FIELD_RESOURCE = "resource"; + private static final String FIELD_ROW_FILTER = "rowFilter"; + private static final String FIELD_COLUMN_MASK = "columnMask"; + private static final String FIELD_PRINCIPAL = "principal"; + + @JsonProperty(FIELD_RESOURCE) + private final PermissionResource resource; + + @Nullable + @JsonProperty(FIELD_ROW_FILTER) + @JsonInclude(JsonInclude.Include.NON_NULL) + private final RowFilter rowFilter; + + @Nullable + @JsonProperty(FIELD_COLUMN_MASK) + @JsonInclude(JsonInclude.Include.NON_NULL) + private final ColumnMask columnMask; + + @JsonProperty(FIELD_PRINCIPAL) + private final String principal; + + @JsonCreator + @ConstructorProperties({FIELD_RESOURCE, FIELD_ROW_FILTER, FIELD_COLUMN_MASK, FIELD_PRINCIPAL}) + public DataPolicy( + @JsonProperty(FIELD_RESOURCE) PermissionResource resource, + @Nullable @JsonProperty(FIELD_ROW_FILTER) RowFilter rowFilter, + @Nullable @JsonProperty(FIELD_COLUMN_MASK) ColumnMask columnMask, + @JsonProperty(FIELD_PRINCIPAL) String principal) { + this.resource = checkNotNull(resource, "resource cannot be null"); + resource.validatePolicyAttachment(); + checkArgument( + (rowFilter == null) != (columnMask == null), + "A policy must contain exactly one of rowFilter and columnMask."); + this.rowFilter = rowFilter; + this.columnMask = columnMask; + this.principal = PermissionAssignment.validatePrincipal(principal); + } + + public static DataPolicy rowFilter( + PermissionResource resource, RowFilter rowFilter, String principal) { + return new DataPolicy(resource, rowFilter, null, principal); + } + + public static DataPolicy columnMask( + PermissionResource resource, ColumnMask columnMask, String principal) { + return new DataPolicy(resource, null, columnMask, principal); + } + + @JsonGetter(FIELD_RESOURCE) + public PermissionResource getResource() { + return resource; + } + + @Nullable + @JsonGetter(FIELD_ROW_FILTER) + public RowFilter getRowFilter() { + return rowFilter; + } + + @Nullable + @JsonGetter(FIELD_COLUMN_MASK) + public ColumnMask getColumnMask() { + return columnMask; + } + + public PolicyType type() { + return rowFilter == null ? PolicyType.COLUMN_MASKING : PolicyType.ROW_FILTER; + } + + @JsonGetter(FIELD_PRINCIPAL) + public String getPrincipal() { + return principal; + } +} diff --git a/paimon-api/src/main/java/org/apache/paimon/management/ListPoliciesRequest.java b/paimon-api/src/main/java/org/apache/paimon/management/ListPoliciesRequest.java new file mode 100644 index 000000000000..e79186be3d3e --- /dev/null +++ b/paimon-api/src/main/java/org/apache/paimon/management/ListPoliciesRequest.java @@ -0,0 +1,102 @@ +/* + * 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. + */ + +package org.apache.paimon.management; + +import org.apache.paimon.annotation.Experimental; + +import javax.annotation.Nullable; + +import static org.apache.paimon.utils.Preconditions.checkArgument; +import static org.apache.paimon.utils.Preconditions.checkNotNull; + +/** Filters for listing policies attached to an exact table resource. */ +@Experimental +public class ListPoliciesRequest { + + private final PermissionResource resource; + @Nullable private final PolicyType type; + @Nullable private final String principal; + @Nullable private final String column; + @Nullable private final String pageToken; + @Nullable private final Integer maxResults; + + public ListPoliciesRequest( + PermissionResource resource, + @Nullable PolicyType type, + @Nullable String principal, + @Nullable String column, + @Nullable String pageToken, + @Nullable Integer maxResults) { + this.resource = checkNotNull(resource, "resource cannot be null"); + resource.validatePolicyAttachment(); + if (!isBlank(principal)) { + PermissionAssignment.validatePrincipal(principal); + } + checkArgument(maxResults == null || maxResults > 0, "maxResults must be greater than 0."); + checkArgument( + maxResults == null || maxResults <= ListPermissionsRequest.MAX_PAGE_SIZE, + "maxResults must be at most %s.", + ListPermissionsRequest.MAX_PAGE_SIZE); + this.type = type; + this.principal = isBlank(principal) ? null : principal; + checkArgument( + isBlank(column) || type == PolicyType.COLUMN_MASKING, + "column filter requires type COLUMN_MASKING."); + this.column = isBlank(column) ? null : column; + this.pageToken = pageToken; + this.maxResults = maxResults; + } + + public PermissionResource getResource() { + return resource; + } + + @Nullable + public PolicyType getType() { + return type; + } + + @Nullable + public String getPrincipal() { + return principal; + } + + @Nullable + public String getColumn() { + return column; + } + + @Nullable + public String getPageToken() { + return pageToken; + } + + @Nullable + public Integer getMaxResults() { + return maxResults; + } + + public ListPoliciesRequest withPageToken(@Nullable String newPageToken) { + return new ListPoliciesRequest(resource, type, principal, column, newPageToken, maxResults); + } + + private static boolean isBlank(@Nullable String value) { + return value == null || value.trim().isEmpty(); + } +} diff --git a/paimon-api/src/main/java/org/apache/paimon/management/PermissionResource.java b/paimon-api/src/main/java/org/apache/paimon/management/PermissionResource.java index f0100af10096..b3d15373b821 100644 --- a/paimon-api/src/main/java/org/apache/paimon/management/PermissionResource.java +++ b/paimon-api/src/main/java/org/apache/paimon/management/PermissionResource.java @@ -122,6 +122,13 @@ public String getView() { return view; } + /** Validates that this resource can carry a data policy in the current contract. */ + public void validatePolicyAttachment() { + checkArgument( + type == ResourceType.TABLE, + "Policies can currently be attached only to TABLE resources."); + } + @Override public boolean equals(Object o) { if (this == o) { diff --git a/paimon-api/src/main/java/org/apache/paimon/management/PolicyManagement.java b/paimon-api/src/main/java/org/apache/paimon/management/PolicyManagement.java new file mode 100644 index 000000000000..6b33195b0629 --- /dev/null +++ b/paimon-api/src/main/java/org/apache/paimon/management/PolicyManagement.java @@ -0,0 +1,70 @@ +/* + * 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. + */ + +package org.apache.paimon.management; + +import org.apache.paimon.PagedList; +import org.apache.paimon.annotation.Experimental; + +import javax.annotation.Nullable; + +/** Control-plane contract for row-filter and column-masking policies. */ +@Experimental +public interface PolicyManagement { + + PagedList listPolicies(ListPoliciesRequest request); + + void createPolicy(DataPolicy policy) throws PolicyAlreadyExistException; + + void dropPolicy( + PermissionResource resource, + PolicyType type, + String principal, + @Nullable String column, + boolean ignoreIfNotExists); + + /** Exception for trying to create a policy that already exists. */ + class PolicyAlreadyExistException extends Exception { + + private final DataPolicy policy; + + public PolicyAlreadyExistException(DataPolicy policy) { + this(policy, null); + } + + public PolicyAlreadyExistException(DataPolicy policy, Throwable cause) { + super(message(policy), cause); + this.policy = policy; + } + + public DataPolicy policy() { + return policy; + } + + private static String message(DataPolicy policy) { + String target = policy.type().name(); + if (policy.getColumnMask() != null) { + target += "(" + policy.getColumnMask().getOnColumn() + ")"; + } + PermissionResource resource = policy.getResource(); + return String.format( + "%s policy for principal '%s' already exists on table '%s.%s'.", + target, policy.getPrincipal(), resource.getDatabase(), resource.getTable()); + } + } +} diff --git a/paimon-api/src/main/java/org/apache/paimon/management/PolicyType.java b/paimon-api/src/main/java/org/apache/paimon/management/PolicyType.java new file mode 100644 index 000000000000..1312a2632fed --- /dev/null +++ b/paimon-api/src/main/java/org/apache/paimon/management/PolicyType.java @@ -0,0 +1,37 @@ +/* + * 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. + */ + +package org.apache.paimon.management; + +import org.apache.paimon.annotation.Experimental; + +import javax.annotation.Nullable; + +import java.util.Locale; + +/** Fine-grained data policy types. */ +@Experimental +public enum PolicyType { + ROW_FILTER, + COLUMN_MASKING; + + @Nullable + public static PolicyType fromString(@Nullable String value) { + return value == null ? null : valueOf(value.toUpperCase(Locale.ROOT)); + } +} diff --git a/paimon-api/src/main/java/org/apache/paimon/management/RowFilter.java b/paimon-api/src/main/java/org/apache/paimon/management/RowFilter.java new file mode 100644 index 000000000000..4af32042b593 --- /dev/null +++ b/paimon-api/src/main/java/org/apache/paimon/management/RowFilter.java @@ -0,0 +1,64 @@ +/* + * 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. + */ + +package org.apache.paimon.management; + +import org.apache.paimon.annotation.Experimental; + +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty; + +import java.beans.ConstructorProperties; +import java.nio.charset.StandardCharsets; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** Serialized Paimon predicate for a row-filter policy. */ +@Experimental +@JsonIgnoreProperties(ignoreUnknown = true) +public class RowFilter { + + public static final int MAX_PREDICATE_BYTES = 60 * 1024; + + private static final String FIELD_PREDICATE = "predicate"; + + @JsonProperty(FIELD_PREDICATE) + private final String predicate; + + @JsonCreator + @ConstructorProperties({FIELD_PREDICATE}) + public RowFilter(@JsonProperty(FIELD_PREDICATE) String predicate) { + checkArgument(!isBlank(predicate), "predicate cannot be empty."); + checkArgument( + predicate.getBytes(StandardCharsets.UTF_8).length <= MAX_PREDICATE_BYTES, + "predicate must not exceed %s UTF-8 bytes.", + MAX_PREDICATE_BYTES); + this.predicate = predicate; + } + + @JsonGetter(FIELD_PREDICATE) + public String getPredicate() { + return predicate; + } + + private static boolean isBlank(String value) { + return value == null || value.trim().isEmpty(); + } +} diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java b/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java index e2f5cc370259..b4b110cfec20 100644 --- a/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java +++ b/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java @@ -26,9 +26,12 @@ import org.apache.paimon.catalog.Identifier; import org.apache.paimon.consumer.ConsumerInfo; import org.apache.paimon.function.FunctionChange; +import org.apache.paimon.management.DataPolicy; import org.apache.paimon.management.ListPermissionsRequest; +import org.apache.paimon.management.ListPoliciesRequest; import org.apache.paimon.management.PermissionAssignment; import org.apache.paimon.management.PermissionResource; +import org.apache.paimon.management.PolicyType; import org.apache.paimon.options.Options; import org.apache.paimon.partition.Partition; import org.apache.paimon.partition.PartitionStatistics; @@ -51,11 +54,13 @@ import org.apache.paimon.rest.requests.CreateTagRequest; import org.apache.paimon.rest.requests.CreateViewRequest; import org.apache.paimon.rest.requests.DropPartitionsRequest; +import org.apache.paimon.rest.requests.DropPolicyRequest; import org.apache.paimon.rest.requests.ForwardBranchRequest; import org.apache.paimon.rest.requests.GrantPermissionRequest; import org.apache.paimon.rest.requests.ListPartitionsByFilterRequest; import org.apache.paimon.rest.requests.ListPartitionsByNamesRequest; import org.apache.paimon.rest.requests.MarkDonePartitionsRequest; +import org.apache.paimon.rest.requests.PolicyRequest; import org.apache.paimon.rest.requests.RegisterTableRequest; import org.apache.paimon.rest.requests.RenameTableRequest; import org.apache.paimon.rest.requests.ReplaceTableRequest; @@ -86,6 +91,7 @@ import org.apache.paimon.rest.responses.ListFunctionsResponse; import org.apache.paimon.rest.responses.ListPartitionsResponse; import org.apache.paimon.rest.responses.ListPermissionsResponse; +import org.apache.paimon.rest.responses.ListPoliciesResponse; import org.apache.paimon.rest.responses.ListSnapshotsResponse; import org.apache.paimon.rest.responses.ListTableDetailsResponse; import org.apache.paimon.rest.responses.ListTablesGloballyResponse; @@ -125,6 +131,7 @@ import static org.apache.paimon.rest.RESTUtil.extractPrefixMap; import static org.apache.paimon.rest.auth.AuthProviderFactory.createAuthProvider; import static org.apache.paimon.utils.Preconditions.checkArgument; +import static org.apache.paimon.utils.Preconditions.checkNotNull; /** * REST API for REST Catalog. @@ -880,6 +887,57 @@ public void revokePermission(PermissionResource resource, String access, String restAuthFunction); } + /** Lists policies attached to an exact table resource. */ + @Experimental + public ListPoliciesResponse listPolicies(ListPoliciesRequest request) { + Map queryParams = Maps.newHashMap(); + if (request.getType() != null) { + putQueryParameter(queryParams, "type", request.getType().name()); + } + putQueryParameter(queryParams, "principal", request.getPrincipal()); + putQueryParameter(queryParams, "column", request.getColumn()); + if (request.getMaxResults() != null) { + queryParams.put(MAX_RESULTS, request.getMaxResults().toString()); + } + putQueryParameter(queryParams, PAGE_TOKEN, request.getPageToken()); + return client.get( + resourcePaths.policies(request.getResource()), + queryParams, + ListPoliciesResponse.class, + restAuthFunction); + } + + /** Creates a principal policy on its attachment resource. */ + @Experimental + public void createPolicy(DataPolicy policy) { + client.post( + resourcePaths.policies(policy.getResource()), + new PolicyRequest(policy), + restAuthFunction); + } + + /** Drops a principal policy from its exact attachment resource. */ + @Experimental + public void dropPolicy( + PermissionResource resource, + PolicyType type, + String principal, + @Nullable String column, + boolean ignoreIfNotExists) { + checkNotNull(resource, "resource cannot be null").validatePolicyAttachment(); + try { + client.post( + resourcePaths.dropPolicy(resource), + new DropPolicyRequest(type, principal, column), + restAuthFunction); + } catch (NoSuchResourceException e) { + if (!ignoreIfNotExists + || !ErrorResponse.RESOURCE_TYPE_POLICY.equals(e.resourceType())) { + throw e; + } + } + } + /** * Drop table. * diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/RESTPolicyManagement.java b/paimon-api/src/main/java/org/apache/paimon/rest/RESTPolicyManagement.java new file mode 100644 index 000000000000..45a42f6cf86c --- /dev/null +++ b/paimon-api/src/main/java/org/apache/paimon/rest/RESTPolicyManagement.java @@ -0,0 +1,72 @@ +/* + * 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. + */ + +package org.apache.paimon.rest; + +import org.apache.paimon.PagedList; +import org.apache.paimon.annotation.Experimental; +import org.apache.paimon.management.DataPolicy; +import org.apache.paimon.management.ListPoliciesRequest; +import org.apache.paimon.management.PermissionResource; +import org.apache.paimon.management.PolicyManagement; +import org.apache.paimon.management.PolicyManagement.PolicyAlreadyExistException; +import org.apache.paimon.management.PolicyType; +import org.apache.paimon.rest.exceptions.AlreadyExistsException; +import org.apache.paimon.rest.responses.ErrorResponse; +import org.apache.paimon.rest.responses.ListPoliciesResponse; + +import javax.annotation.Nullable; + +/** REST implementation of data policy management for a configured catalog prefix. */ +@Experimental +public class RESTPolicyManagement implements PolicyManagement { + + private final RESTApi api; + + public RESTPolicyManagement(RESTApi api) { + this.api = api; + } + + @Override + public PagedList listPolicies(ListPoliciesRequest request) { + ListPoliciesResponse response = api.listPolicies(request); + return new PagedList<>(response.getPolicies(), response.getNextPageToken()); + } + + @Override + public void createPolicy(DataPolicy policy) throws PolicyAlreadyExistException { + try { + api.createPolicy(policy); + } catch (AlreadyExistsException e) { + if (ErrorResponse.RESOURCE_TYPE_POLICY.equals(e.resourceType())) { + throw new PolicyAlreadyExistException(policy, e); + } + throw e; + } + } + + @Override + public void dropPolicy( + PermissionResource resource, + PolicyType type, + String principal, + @Nullable String column, + boolean ignoreIfNotExists) { + api.dropPolicy(resource, type, principal, column, ignoreIfNotExists); + } +} diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java b/paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java index 6ae7ebf8c3ec..4cef311061a5 100644 --- a/paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java +++ b/paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java @@ -19,6 +19,7 @@ package org.apache.paimon.rest; import org.apache.paimon.annotation.Experimental; +import org.apache.paimon.management.PermissionResource; import org.apache.paimon.options.Options; import org.apache.paimon.shade.guava30.com.google.common.base.Joiner; @@ -44,6 +45,7 @@ public class ResourcePaths { protected static final String FUNCTIONS = "functions"; protected static final String FUNCTION_DETAILS = "function-details"; protected static final String PERMISSIONS = "permissions"; + protected static final String POLICIES = "policies"; protected static final String ID = "id"; private static final Joiner SLASH = Joiner.on("/").skipNulls(); @@ -77,6 +79,19 @@ public String revokePermission() { return SLASH.join(permissions(), "revoke"); } + /** Policy collection nested below its attachment resource. */ + @Experimental + public String policies(PermissionResource resource) { + resource.validatePolicyAttachment(); + return SLASH.join(table(resource.getDatabase(), resource.getTable()), POLICIES); + } + + /** Action endpoint for dropping one policy from its attachment resource. */ + @Experimental + public String dropPolicy(PermissionResource resource) { + return SLASH.join(policies(resource), "drop"); + } + public String databases() { return SLASH.join(V1, prefix, DATABASES); } diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/requests/DropPolicyRequest.java b/paimon-api/src/main/java/org/apache/paimon/rest/requests/DropPolicyRequest.java new file mode 100644 index 000000000000..64b7b83c0ef1 --- /dev/null +++ b/paimon-api/src/main/java/org/apache/paimon/rest/requests/DropPolicyRequest.java @@ -0,0 +1,99 @@ +/* + * 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. + */ + +package org.apache.paimon.rest.requests; + +import org.apache.paimon.annotation.Experimental; +import org.apache.paimon.management.PermissionAssignment; +import org.apache.paimon.management.PolicyType; +import org.apache.paimon.rest.RESTRequest; + +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonInclude; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty; + +import javax.annotation.Nullable; + +import java.beans.ConstructorProperties; + +import static org.apache.paimon.utils.Preconditions.checkArgument; +import static org.apache.paimon.utils.Preconditions.checkNotNull; + +/** Request for dropping one principal's row filter or column mask. */ +@Experimental +@JsonIgnoreProperties(ignoreUnknown = true) +public class DropPolicyRequest implements RESTRequest { + + private static final String FIELD_TYPE = "type"; + private static final String FIELD_PRINCIPAL = "principal"; + private static final String FIELD_COLUMN = "column"; + + private final PolicyType type; + private final String principal; + @Nullable private final String column; + + @JsonCreator + @ConstructorProperties({FIELD_TYPE, FIELD_PRINCIPAL, FIELD_COLUMN}) + public DropPolicyRequest( + @JsonProperty(FIELD_TYPE) PolicyType type, + @JsonProperty(FIELD_PRINCIPAL) String principal, + @Nullable @JsonProperty(FIELD_COLUMN) String column) { + this.type = checkNotNull(type, "policy type cannot be null"); + this.principal = validatePrincipal(principal); + if (type == PolicyType.ROW_FILTER) { + checkArgument(isBlank(column), "ROW_FILTER identity cannot contain a column."); + this.column = null; + } else { + checkArgument(!isBlank(column), "column is required for COLUMN_MASKING identity."); + this.column = column; + } + } + + @JsonGetter(FIELD_TYPE) + public PolicyType getType() { + return type; + } + + @JsonGetter(FIELD_PRINCIPAL) + public String getPrincipal() { + return principal; + } + + @Nullable + @JsonGetter(FIELD_COLUMN) + @JsonInclude(JsonInclude.Include.NON_NULL) + public String getColumn() { + return column; + } + + private static String validatePrincipal(String principal) { + checkArgument( + principal != null && !principal.trim().isEmpty(), "principal cannot be empty."); + checkArgument( + principal.length() <= PermissionAssignment.MAX_PRINCIPAL_LENGTH, + "principal must contain at most %s characters.", + PermissionAssignment.MAX_PRINCIPAL_LENGTH); + return principal; + } + + private static boolean isBlank(@Nullable String value) { + return value == null || value.trim().isEmpty(); + } +} diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/requests/PolicyRequest.java b/paimon-api/src/main/java/org/apache/paimon/rest/requests/PolicyRequest.java new file mode 100644 index 000000000000..f1e9a0ed53e5 --- /dev/null +++ b/paimon-api/src/main/java/org/apache/paimon/rest/requests/PolicyRequest.java @@ -0,0 +1,96 @@ +/* + * 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. + */ + +package org.apache.paimon.rest.requests; + +import org.apache.paimon.annotation.Experimental; +import org.apache.paimon.management.ColumnMask; +import org.apache.paimon.management.DataPolicy; +import org.apache.paimon.management.PermissionResource; +import org.apache.paimon.management.RowFilter; +import org.apache.paimon.rest.RESTRequest; + +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnore; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonInclude; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty; + +import javax.annotation.Nullable; + +import java.beans.ConstructorProperties; + +/** Create payload for a principal policy whose table is identified by the request path. */ +@Experimental +@JsonIgnoreProperties(ignoreUnknown = true) +public class PolicyRequest implements RESTRequest { + + private static final String FIELD_ROW_FILTER = "rowFilter"; + private static final String FIELD_COLUMN_MASK = "columnMask"; + private static final String FIELD_PRINCIPAL = "principal"; + + @Nullable private final RowFilter rowFilter; + @Nullable private final ColumnMask columnMask; + private final String principal; + + public PolicyRequest(DataPolicy policy) { + this(policy.getRowFilter(), policy.getColumnMask(), policy.getPrincipal()); + } + + @JsonCreator + @ConstructorProperties({FIELD_ROW_FILTER, FIELD_COLUMN_MASK, FIELD_PRINCIPAL}) + public PolicyRequest( + @Nullable @JsonProperty(FIELD_ROW_FILTER) RowFilter rowFilter, + @Nullable @JsonProperty(FIELD_COLUMN_MASK) ColumnMask columnMask, + @JsonProperty(FIELD_PRINCIPAL) String principal) { + this.rowFilter = rowFilter; + this.columnMask = columnMask; + this.principal = principal; + } + + public DataPolicy policy(PermissionResource resource) { + return new DataPolicy(resource, rowFilter, columnMask, principal); + } + + /** Creating a principal policy cannot be replayed after an ambiguous server response. */ + @JsonIgnore + @Override + public boolean isRetrySafe() { + return false; + } + + @Nullable + @JsonGetter(FIELD_ROW_FILTER) + @JsonInclude(JsonInclude.Include.NON_NULL) + public RowFilter getRowFilter() { + return rowFilter; + } + + @Nullable + @JsonGetter(FIELD_COLUMN_MASK) + @JsonInclude(JsonInclude.Include.NON_NULL) + public ColumnMask getColumnMask() { + return columnMask; + } + + @JsonGetter(FIELD_PRINCIPAL) + public String getPrincipal() { + return principal; + } +} diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/responses/ErrorResponse.java b/paimon-api/src/main/java/org/apache/paimon/rest/responses/ErrorResponse.java index 4eb52308dd0c..4fe52b1d0b22 100644 --- a/paimon-api/src/main/java/org/apache/paimon/rest/responses/ErrorResponse.java +++ b/paimon-api/src/main/java/org/apache/paimon/rest/responses/ErrorResponse.java @@ -51,6 +51,8 @@ public class ErrorResponse implements RESTResponse { public static final String RESOURCE_TYPE_FUNCTION = "FUNCTION"; + public static final String RESOURCE_TYPE_POLICY = "POLICY"; + public static final String RESOURCE_TYPE_DEFINITION = "DEFINITION"; private static final String FIELD_MESSAGE = "message"; diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/responses/ListPoliciesResponse.java b/paimon-api/src/main/java/org/apache/paimon/rest/responses/ListPoliciesResponse.java new file mode 100644 index 000000000000..95595f066d59 --- /dev/null +++ b/paimon-api/src/main/java/org/apache/paimon/rest/responses/ListPoliciesResponse.java @@ -0,0 +1,72 @@ +/* + * 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. + */ + +package org.apache.paimon.rest.responses; + +import org.apache.paimon.annotation.Experimental; +import org.apache.paimon.management.DataPolicy; + +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonInclude; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty; + +import javax.annotation.Nullable; + +import java.beans.ConstructorProperties; +import java.util.List; + +/** Response for listing data policies. */ +@Experimental +@JsonIgnoreProperties(ignoreUnknown = true) +public class ListPoliciesResponse implements PagedResponse { + + private static final String FIELD_POLICIES = "policies"; + private static final String FIELD_NEXT_PAGE_TOKEN = "nextPageToken"; + + private final List policies; + @Nullable private final String nextPageToken; + + @JsonCreator + @ConstructorProperties({FIELD_POLICIES, FIELD_NEXT_PAGE_TOKEN}) + public ListPoliciesResponse( + @JsonProperty(FIELD_POLICIES) List policies, + @Nullable @JsonProperty(FIELD_NEXT_PAGE_TOKEN) String nextPageToken) { + this.policies = policies; + this.nextPageToken = nextPageToken; + } + + @JsonGetter(FIELD_POLICIES) + public List getPolicies() { + return policies; + } + + @Override + @Nullable + @JsonGetter(FIELD_NEXT_PAGE_TOKEN) + @JsonInclude(JsonInclude.Include.NON_NULL) + public String getNextPageToken() { + return nextPageToken; + } + + @Override + public List data() { + return policies; + } +} diff --git a/paimon-api/src/test/java/org/apache/paimon/management/PolicyManagementJsonTest.java b/paimon-api/src/test/java/org/apache/paimon/management/PolicyManagementJsonTest.java new file mode 100644 index 000000000000..53c1905144dd --- /dev/null +++ b/paimon-api/src/test/java/org/apache/paimon/management/PolicyManagementJsonTest.java @@ -0,0 +1,176 @@ +/* + * 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. + */ + +package org.apache.paimon.management; + +import org.apache.paimon.rest.RESTApi; +import org.apache.paimon.rest.requests.DropPolicyRequest; +import org.apache.paimon.rest.requests.PolicyRequest; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** JSON and validation tests for data-policy management contracts. */ +public class PolicyManagementJsonTest { + + private static final String PREDICATE_JSON = + "{\"kind\":\"LEAF\",\"transform\":{\"name\":\"FIELD_REF\"," + + "\"fieldRef\":{\"index\":0,\"name\":\"region\",\"type\":\"STRING\"}}," + + "\"function\":\"EQUAL\",\"literals\":[\"APAC\"]}"; + private static final String TRANSFORM_JSON = + "{\"name\":\"CONCAT\",\"inputs\":[{\"index\":0,\"name\":\"region\"," + + "\"type\":\"STRING\"},\"****\"]}"; + + @Test + void testPolicyDefinitionsRoundTripWithShadedAndExternalJackson() throws Exception { + DataPolicy policy = + DataPolicy.columnMask( + tableResource(), new ColumnMask("email", TRANSFORM_JSON), "analyst"); + String json = RESTApi.toJson(policy); + + DataPolicy shaded = RESTApi.fromJson(json, DataPolicy.class); + DataPolicy external = + new com.fasterxml.jackson.databind.ObjectMapper().readValue(json, DataPolicy.class); + for (DataPolicy roundTrip : Arrays.asList(shaded, external)) { + assertThat(roundTrip.type()).isEqualTo(PolicyType.COLUMN_MASKING); + assertThat(roundTrip.getResource()).isEqualTo(tableResource()); + assertThat(roundTrip.getColumnMask().getOnColumn()).isEqualTo("email"); + assertThat(roundTrip.getColumnMask().getTransform()).isEqualTo(TRANSFORM_JSON); + assertThat(roundTrip.getPrincipal()).isEqualTo("analyst"); + } + + PolicyRequest request = new PolicyRequest(policy); + assertThat(request.isRetrySafe()).isFalse(); + Map wire = RESTApi.fromJson(RESTApi.toJson(request), Map.class); + assertThat(wire).hasSize(2); + assertThat(wire.get("columnMask")).isNotNull(); + assertThat(wire.get("principal")).isEqualTo("analyst"); + assertThat(request.policy(tableResource()).getResource()).isEqualTo(tableResource()); + } + + @Test + void testRowFilterRoundTrip() throws Exception { + DataPolicy policy = + DataPolicy.rowFilter(tableResource(), new RowFilter(PREDICATE_JSON), "analyst"); + + DataPolicy roundTrip = RESTApi.fromJson(RESTApi.toJson(policy), DataPolicy.class); + assertThat(roundTrip.type()).isEqualTo(PolicyType.ROW_FILTER); + assertThat(roundTrip.getRowFilter().getPredicate()).isEqualTo(PREDICATE_JSON); + assertThat(roundTrip.getColumnMask()).isNull(); + } + + @Test + void testPolicyValidationAndPayloadBounds() { + assertThatThrownBy(() -> new RowFilter(" ")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("predicate"); + assertThatThrownBy(() -> new ColumnMask("email", " ")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("transform"); + assertThatThrownBy( + () -> + new DataPolicy( + tableResource(), + new RowFilter(PREDICATE_JSON), + new ColumnMask("email", TRANSFORM_JSON), + "analyst")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("exactly one"); + assertThatThrownBy( + () -> + DataPolicy.columnMask( + catalogResource(), + new ColumnMask("email", TRANSFORM_JSON), + "analyst")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("TABLE"); + assertThatThrownBy(() -> new RowFilter(repeat('p', RowFilter.MAX_PREDICATE_BYTES + 1))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("UTF-8 bytes"); + assertThatThrownBy( + () -> + new ColumnMask( + "email", repeat('t', ColumnMask.MAX_TRANSFORM_BYTES + 1))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("UTF-8 bytes"); + } + + @Test + void testDropPolicyRequestRoundTripAndIdentityValidation() throws Exception { + DropPolicyRequest request = + new DropPolicyRequest(PolicyType.COLUMN_MASKING, "analyst", "email"); + DropPolicyRequest roundTrip = + RESTApi.fromJson(RESTApi.toJson(request), DropPolicyRequest.class); + + assertThat(roundTrip.getType()).isEqualTo(PolicyType.COLUMN_MASKING); + assertThat(roundTrip.getPrincipal()).isEqualTo("analyst"); + assertThat(roundTrip.getColumn()).isEqualTo("email"); + assertThatThrownBy(() -> new DropPolicyRequest(PolicyType.COLUMN_MASKING, "analyst", null)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("column is required"); + assertThatThrownBy(() -> new DropPolicyRequest(PolicyType.ROW_FILTER, "analyst", "email")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("cannot contain a column"); + } + + @Test + void testListPoliciesValidationAndOpaquePageToken() { + ListPoliciesRequest request = + new ListPoliciesRequest( + tableResource(), PolicyType.COLUMN_MASKING, "analyst", "email", null, 25); + + assertThat(request.withPageToken(" \t").getPageToken()).isEqualTo(" \t"); + assertThatThrownBy( + () -> + new ListPoliciesRequest( + tableResource(), null, null, "email", null, 25)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("COLUMN_MASKING"); + assertThatThrownBy( + () -> + new ListPoliciesRequest( + tableResource(), null, null, null, null, 1001)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("at most 1000"); + assertThatThrownBy( + () -> + new ListPoliciesRequest( + catalogResource(), null, null, null, null, 25)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("TABLE"); + } + + private static PermissionResource catalogResource() { + return new PermissionResource(ResourceType.CATALOG, null, null, null, null); + } + + private static PermissionResource tableResource() { + return new PermissionResource(ResourceType.TABLE, "sales", "orders", null, null); + } + + private static String repeat(char value, int length) { + char[] values = new char[length]; + Arrays.fill(values, value); + return new String(values); + } +} diff --git a/paimon-api/src/test/java/org/apache/paimon/rest/RESTPolicyManagementTest.java b/paimon-api/src/test/java/org/apache/paimon/rest/RESTPolicyManagementTest.java new file mode 100644 index 000000000000..a09e0e902a50 --- /dev/null +++ b/paimon-api/src/test/java/org/apache/paimon/rest/RESTPolicyManagementTest.java @@ -0,0 +1,251 @@ +/* + * 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. + */ + +package org.apache.paimon.rest; + +import org.apache.paimon.PagedList; +import org.apache.paimon.management.ColumnMask; +import org.apache.paimon.management.DataPolicy; +import org.apache.paimon.management.ListPoliciesRequest; +import org.apache.paimon.management.PermissionResource; +import org.apache.paimon.management.PolicyManagement; +import org.apache.paimon.management.PolicyManagement.PolicyAlreadyExistException; +import org.apache.paimon.management.PolicyType; +import org.apache.paimon.management.ResourceType; +import org.apache.paimon.options.Options; +import org.apache.paimon.rest.exceptions.AlreadyExistsException; +import org.apache.paimon.rest.exceptions.NoSuchResourceException; +import org.apache.paimon.utils.JsonSerdeUtil; + +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpServer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import static org.apache.paimon.rest.RESTCatalogInternalOptions.PREFIX; +import static org.apache.paimon.rest.RESTCatalogOptions.TOKEN; +import static org.apache.paimon.rest.RESTCatalogOptions.TOKEN_PROVIDER; +import static org.apache.paimon.rest.RESTCatalogOptions.URI; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Behavioral tests for REST data-policy management. */ +public class RESTPolicyManagementTest { + + private static final String COLLECTION_PATH = + "/v1/catalog+id/databases/sales/tables/orders/policies"; + private static final String DROP_PATH = COLLECTION_PATH + "/drop"; + + private HttpServer server; + private PolicyManagement management; + private final AtomicReference listQuery = new AtomicReference<>(); + private final AtomicReference createBody = new AtomicReference<>(); + private final AtomicReference createError = new AtomicReference<>(); + private final AtomicReference dropBody = new AtomicReference<>(); + private final AtomicReference dropError = new AtomicReference<>(); + private final AtomicInteger dropCalls = new AtomicInteger(); + + @BeforeEach + void setUp() throws Exception { + server = HttpServer.create(new InetSocketAddress(0), 0); + server.createContext( + "/v1/", + exchange -> { + String path = exchange.getRequestURI().getRawPath(); + String method = exchange.getRequestMethod(); + if (COLLECTION_PATH.equals(path) && "GET".equals(method)) { + listQuery.set(exchange.getRequestURI().getRawQuery()); + respond(exchange, 200, listResponse()); + } else if (COLLECTION_PATH.equals(path) && "POST".equals(method)) { + createBody.set(readBody(exchange)); + if (createError.get() == null) { + respond(exchange, 200, null); + } else { + respond(exchange, 409, createError.get()); + } + } else if (DROP_PATH.equals(path) && "POST".equals(method)) { + dropBody.set(readBody(exchange)); + dropCalls.incrementAndGet(); + if (dropError.get() == null) { + respond(exchange, 200, null); + } else { + respond(exchange, 404, dropError.get()); + } + } else { + respond(exchange, 404, "{\"message\":\"missing\",\"code\":404}"); + } + }); + server.start(); + + Options options = new Options(); + options.set(URI, "http://127.0.0.1:" + server.getAddress().getPort()); + options.set(TOKEN_PROVIDER, "bear"); + options.set(TOKEN, "secret"); + options.set(PREFIX, "catalog id"); + management = new RESTPolicyManagement(new RESTApi(options, false)); + } + + @AfterEach + void tearDown() { + if (server != null) { + server.stop(0); + } + } + + @Test + void testListUsesResourceNestedPathAndIdentityFilters() { + PagedList policies = + management.listPolicies( + new ListPoliciesRequest( + tableResource(), + PolicyType.COLUMN_MASKING, + "analyst", + "email", + "start", + 25)); + + assertThat(policies.getElements()).hasSize(1); + assertThat(policies.getNextPageToken()).isEqualTo("next"); + assertThat(listQuery.get()) + .contains("type=COLUMN_MASKING") + .contains("principal=analyst") + .contains("column=email") + .contains("maxResults=25") + .contains("pageToken=start"); + } + + @Test + void testCreateAndDropUsePostEndpoints() throws Exception { + DataPolicy policy = policy(); + management.createPolicy(policy); + management.dropPolicy( + policy.getResource(), + policy.type(), + policy.getPrincipal(), + policy.getColumnMask().getOnColumn(), + false); + + assertThat(createBody.get()).contains("\"principal\":\"analyst\""); + assertThat(createBody.get()).doesNotContain("\"resource\""); + assertThat(dropBody.get()) + .contains("\"type\":\"COLUMN_MASKING\"") + .contains("\"column\":\"email\""); + assertThat(dropCalls).hasValue(1); + } + + @Test + void testCreateMapsOnlyPolicyConflict() { + DataPolicy policy = policy(); + createError.set( + "{\"resourceType\":\"POLICY\",\"resourceName\":" + + "\"COLUMN_MASKING:analyst:email\"," + + "\"message\":\"already exists\",\"code\":409}"); + + assertThatThrownBy(() -> management.createPolicy(policy)) + .isInstanceOf(PolicyAlreadyExistException.class) + .hasMessageContaining("COLUMN_MASKING(email)") + .hasCauseInstanceOf(AlreadyExistsException.class); + + createError.set( + "{\"resourceType\":\"TABLE\",\"resourceName\":\"orders\"," + + "\"message\":\"table conflict\",\"code\":409}"); + assertThatThrownBy(() -> management.createPolicy(policy)) + .isInstanceOf(AlreadyExistsException.class) + .hasMessageContaining("table conflict"); + } + + @Test + void testDropIfExistsOnlyIgnoresMissingPolicy() { + DataPolicy policy = policy(); + dropError.set( + "{\"resourceType\":\"POLICY\",\"resourceName\":" + + "\"COLUMN_MASKING:analyst:email\"," + + "\"message\":\"missing\",\"code\":404}"); + + management.dropPolicy( + policy.getResource(), + policy.type(), + policy.getPrincipal(), + policy.getColumnMask().getOnColumn(), + true); + + dropError.set( + "{\"resourceType\":\"TABLE\",\"resourceName\":\"orders\"," + + "\"message\":\"missing table\",\"code\":404}"); + assertThatThrownBy( + () -> + management.dropPolicy( + policy.getResource(), + policy.type(), + policy.getPrincipal(), + policy.getColumnMask().getOnColumn(), + true)) + .isInstanceOf(NoSuchResourceException.class) + .hasMessageContaining("missing table"); + } + + private static DataPolicy policy() { + return DataPolicy.columnMask( + tableResource(), + new ColumnMask( + "email", + "{\"name\":\"FIELD_REF\",\"fieldRef\":{\"index\":0," + + "\"name\":\"region\",\"type\":\"STRING\"}}"), + "analyst"); + } + + private static PermissionResource tableResource() { + return new PermissionResource(ResourceType.TABLE, "sales", "orders", null, null); + } + + private static String listResponse() { + return "{\"policies\":[" + policyJson() + "],\"nextPageToken\":\"next\"}"; + } + + private static String policyJson() { + return JsonSerdeUtil.toFlatJson(policy()); + } + + private static String readBody(HttpExchange exchange) throws IOException { + byte[] data = new byte[8192]; + int read = exchange.getRequestBody().read(data); + return read < 0 ? "" : new String(data, 0, read, StandardCharsets.UTF_8); + } + + private static void respond(HttpExchange exchange, int code, String body) throws IOException { + if (body == null) { + exchange.sendResponseHeaders(code, 0); + exchange.getResponseBody().close(); + } else { + byte[] bytes = body.getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(code, bytes.length); + try (OutputStream output = exchange.getResponseBody()) { + output.write(bytes); + } + exchange.close(); + } + } +} diff --git a/paimon-api/src/test/java/org/apache/paimon/rest/requests/RequestJacksonCompatibilityTest.java b/paimon-api/src/test/java/org/apache/paimon/rest/requests/RequestJacksonCompatibilityTest.java index 30ebab546325..9e645db3b79f 100644 --- a/paimon-api/src/test/java/org/apache/paimon/rest/requests/RequestJacksonCompatibilityTest.java +++ b/paimon-api/src/test/java/org/apache/paimon/rest/requests/RequestJacksonCompatibilityTest.java @@ -169,7 +169,9 @@ public class RequestJacksonCompatibilityTest { CreatePartitionsRequest.class, CreateTableRequest.class, CreateViewRequest.class, + DropPolicyRequest.class, GrantPermissionRequest.class, + PolicyRequest.class, RegisterTableRequest.class, RenameTableRequest.class, ReplaceTableRequest.class, diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java index 3c668aeacb12..9a21f3fa41e2 100644 --- a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java @@ -41,6 +41,7 @@ import org.apache.paimon.function.Function; import org.apache.paimon.function.FunctionChange; import org.apache.paimon.management.PermissionManagement; +import org.apache.paimon.management.PolicyManagement; import org.apache.paimon.options.Options; import org.apache.paimon.partition.Partition; import org.apache.paimon.partition.PartitionStatistics; @@ -144,6 +145,11 @@ public PermissionManagement permissionManagement() { return new RESTPermissionManagement(api); } + @Experimental + public PolicyManagement policyManagement() { + return new RESTPolicyManagement(api); + } + @Override public List listDatabases() { return api.listDatabases(); diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/ResourcePathsTest.java b/paimon-core/src/test/java/org/apache/paimon/rest/ResourcePathsTest.java index 432ea8f77622..b6f0f39f38f5 100644 --- a/paimon-core/src/test/java/org/apache/paimon/rest/ResourcePathsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/rest/ResourcePathsTest.java @@ -18,9 +18,13 @@ package org.apache.paimon.rest; +import org.apache.paimon.management.PermissionResource; +import org.apache.paimon.management.ResourceType; + import org.junit.Test; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; /** Test for {@link ResourcePaths}. */ public class ResourcePathsTest { @@ -47,4 +51,24 @@ public void testPermissionManagementUsesPrefix() { assertEquals("/v1/catalog%2Fid/permissions/grant", resourcePaths.grantPermission()); assertEquals("/v1/catalog%2Fid/permissions/revoke", resourcePaths.revokePermission()); } + + @Test + public void testPoliciesAreNestedUnderAttachmentResource() { + ResourcePaths paths = new ResourcePaths("catalog/id"); + PermissionResource catalog = + new PermissionResource(ResourceType.CATALOG, null, null, null, null); + PermissionResource database = + new PermissionResource(ResourceType.DATABASE, "sales db", null, null, null); + PermissionResource table = + new PermissionResource(ResourceType.TABLE, "sales db", "orders/all", null, null); + + assertThrows(IllegalArgumentException.class, () -> paths.policies(catalog)); + assertThrows(IllegalArgumentException.class, () -> paths.policies(database)); + assertEquals( + "/v1/catalog%2Fid/databases/sales+db/tables/orders%2Fall/policies", + paths.policies(table)); + assertEquals( + "/v1/catalog%2Fid/databases/sales+db/tables/orders%2Fall/policies/drop", + paths.dropPolicy(table)); + } }