roryqi commented on code in PR #12850: URL: https://github.com/apache/gravitino/pull/12850#discussion_r3922718968
########## server/src/main/java/org/apache/gravitino/server/web/filter/authorization/LineageAuthorizationExecutor.java: ########## @@ -0,0 +1,187 @@ +/* + * 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.gravitino.server.web.filter.authorization; + +import static org.apache.gravitino.server.web.filter.ParameterUtil.extractFromParameters; + +import com.google.common.base.Preconditions; +import io.openlineage.server.OpenLineage.Dataset; +import io.openlineage.server.OpenLineage.DatasetFacet; +import io.openlineage.server.OpenLineage.DatasetFacets; +import io.openlineage.server.OpenLineage.RunEvent; +import java.lang.reflect.Parameter; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import org.apache.commons.lang3.StringUtils; +import org.apache.gravitino.Entity; +import org.apache.gravitino.MetadataObject; +import org.apache.gravitino.MetadataObjects; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.authorization.AuthorizationRequestContext; +import org.apache.gravitino.server.authorization.expression.AuthorizationExpressionEvaluator; +import org.apache.gravitino.utils.MetadataObjectUtil; +import org.apache.gravitino.utils.NameIdentifierUtil; + +/** Authorization executor for every input and output dataset in an OpenLineage event. */ +public class LineageAuthorizationExecutor implements AuthorizationExecutor { + + private static final String DATASET_TYPE_FACET = "datasetType"; + + private final Parameter[] parameters; + private final Object[] args; + private final String expression; + + /** + * Creates an authorization executor for an OpenLineage event. + * + * @param parameters parameters of the intercepted REST method + * @param args arguments passed to the intercepted REST method + * @param expression authorization expression to evaluate for every dataset + */ + public LineageAuthorizationExecutor(Parameter[] parameters, Object[] args, String expression) { + this.parameters = parameters; + this.args = args; + this.expression = expression; + } + + @Override + public Set<String> getAuthorizationMetalakes() { + Object request = extractFromParameters(parameters, args); + if (!(request instanceof RunEvent)) { + return Set.of(); + } + + RunEvent event = (RunEvent) request; + if (!hasValidDatasetIdentifiers(event.getInputs()) + || !hasValidDatasetIdentifiers(event.getOutputs())) { + return Set.of(); + } + + Set<String> metalakes = new LinkedHashSet<>(); + addDatasetMetalakes(event.getInputs(), metalakes); + addDatasetMetalakes(event.getOutputs(), metalakes); + return metalakes; + } + + @Override + public boolean execute(AuthorizationRequestContext context) { + Object request = extractFromParameters(parameters, args); + if (!(request instanceof RunEvent)) { + // Request validation is owned by LineageOperations and must still run when authorization is + // disabled. Let it return HTTP 400 for a null or malformed event. + return true; + } + + RunEvent event = (RunEvent) request; + if (!hasValidDatasetIdentifiers(event.getInputs()) + || !hasValidDatasetIdentifiers(event.getOutputs())) { + return true; + } + + AuthorizationExpressionEvaluator evaluator = new AuthorizationExpressionEvaluator(expression); + context.setOriginalAuthorizationExpression(expression); + + return authorizeDatasets(event.getInputs(), context, evaluator) + && authorizeDatasets(event.getOutputs(), context, evaluator); + } + + static MetadataObject.Type getMetadataType(Dataset dataset) { + DatasetFacets facets = dataset.getFacets(); + if (facets == null) { + return MetadataObject.Type.TABLE; + } + + DatasetFacet datasetTypeFacet = facets.getAdditionalProperties().get(DATASET_TYPE_FACET); + if (datasetTypeFacet == null) { + return MetadataObject.Type.TABLE; + } + + Object datasetType = datasetTypeFacet.getAdditionalProperties().get(DATASET_TYPE_FACET); + Preconditions.checkArgument( + datasetType instanceof String && StringUtils.isNotBlank((String) datasetType), + "The datasetType facet must contain a non-blank datasetType"); + + return switch (((String) datasetType).toUpperCase(Locale.ROOT)) { + case "TABLE" -> MetadataObject.Type.TABLE; + case "VIEW" -> MetadataObject.Type.VIEW; + case "FILE", "FILESET" -> MetadataObject.Type.FILESET; + case "MODEL", "MODEL_VERSION" -> MetadataObject.Type.MODEL; + case "TOPIC" -> MetadataObject.Type.TOPIC; + default -> throw new IllegalArgumentException("Unsupported dataset type: " + datasetType); + }; + } + + private static boolean hasValidDatasetIdentifiers(List<? extends Dataset> datasets) { + if (datasets == null) { + return true; + } + + return datasets.stream() + .allMatch( + dataset -> + dataset != null + && StringUtils.isNotBlank(dataset.getNamespace()) + && StringUtils.isNotBlank(dataset.getName())); + } + + private static void addDatasetMetalakes(List<? extends Dataset> datasets, Set<String> metalakes) { + if (datasets != null) { + datasets.forEach(dataset -> metalakes.add(dataset.getNamespace())); + } + } + + private static boolean authorizeDatasets( + List<? extends Dataset> datasets, + AuthorizationRequestContext context, + AuthorizationExpressionEvaluator evaluator) { + if (datasets == null) { + return true; + } + + for (Dataset dataset : datasets) { + if (!authorizeDataset(dataset, context, evaluator)) { + return false; + } + } + return true; + } + + private static boolean authorizeDataset( + Dataset dataset, + AuthorizationRequestContext context, + AuthorizationExpressionEvaluator evaluator) { + String metalake = dataset.getNamespace(); + try { + MetadataObject.Type metadataType = getMetadataType(dataset); + MetadataObject metadataObject = MetadataObjects.parse(dataset.getName(), metadataType); + Entity.EntityType entityType = MetadataObjectUtil.toEntityType(metadataType); + NameIdentifier identifier = MetadataObjectUtil.toEntityIdent(metalake, metadataObject); + Map<Entity.EntityType, NameIdentifier> metadataContext = + NameIdentifierUtil.splitNameIdentifier(metalake, entityType, identifier); + return evaluator.evaluate(metadataContext, Map.of(), context, Optional.of(entityType.name())); + } catch (IllegalArgumentException exception) { Review Comment: Fixed in e7f6ccfa2. All dataset types, names, namespaces, and metadata contexts are resolved before authorization. Only target-preparation IllegalArgumentException is mapped to 400; ordinary denial remains 403, while evaluator or authorizer failures propagate to the interceptor 500 path. Tests cover all three mappings. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
