This is an automated email from the ASF dual-hosted git repository.
roryqi pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 954bc1fca4 [Cherry-pick to branch-1.3] [#12840] fix(lineage): Validate
and authorize lineage events (#12850) (#12911)
954bc1fca4 is described below
commit 954bc1fca4f484f8969d99a749de9a04745b9fad
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Sep 4 19:03:58 2026 +0800
[Cherry-pick to branch-1.3] [#12840] fix(lineage): Validate and authorize
lineage events (#12850) (#12911)
**Cherry-pick Information:**
- Original commit: c0871fa58ef602782e38d87cea39cf68fae8d358
- Target branch: `branch-1.3`
- Status: ✅ **Conflicts resolved**
The cherry-pick conflicts have been manually resolved and validated.
---------
Co-authored-by: roryqi <[email protected]>
Co-authored-by: roryqi <[email protected]>
---
docs/open-api/lineage.yaml | 29 +-
lineage/build.gradle.kts | 1 +
.../lineage/source/rest/LineageEventValidator.java | 73 ++++
.../lineage/source/rest/LineageOperations.java | 24 +-
.../lineage/source/TestLineageOperations.java | 31 +-
.../source/rest/TestLineageEventValidator.java | 153 +++++++
.../annotations/AuthorizationRequest.java | 1 +
server/build.gradle.kts | 3 +
.../web/filter/GravitinoInterceptionService.java | 132 ++++--
.../authorization/AuthorizationExecutor.java | 10 +
.../authorization/AuthorizeExecutorFactory.java | 1 +
.../LineageAuthorizationExecutor.java | 188 ++++++++
.../filter/TestGravitinoInterceptionService.java | 36 ++
.../TestLineageAuthorizationExecutor.java | 478 +++++++++++++++++++++
14 files changed, 1117 insertions(+), 43 deletions(-)
diff --git a/docs/open-api/lineage.yaml b/docs/open-api/lineage.yaml
index 30074bf05b..187f3e56ef 100644
--- a/docs/open-api/lineage.yaml
+++ b/docs/open-api/lineage.yaml
@@ -21,7 +21,17 @@ paths:
/lineage:
post:
summary: Post runEvent
- description: Updates a run state for a job.
+ description: |
+ Updates a run state for a job. When authorization is enabled,
`job.namespace`
+ identifies the Gravitino metalake (organization), and every input and
output
+ dataset namespace must match it. Dataset names must be Gravitino
metadata full
+ names. The optional `datasetType` facet defaults to `TABLE`; supported
values are
+ `TABLE`, `VIEW`, `FILE`, `FILESET`, `MODEL`, and `TOPIC`.
+ Inputs and outputs both require metadata visibility because this
endpoint records
+ producer-reported lineage rather than attesting underlying read or
write execution.
+ RunEvents without datasets require only metalake membership.
Unsupported or external
+ dataset identifiers are rejected when authorization is enabled. When
authorization
+ is disabled, generic OpenLineage namespaces remain supported.
operationId: postRunEvent
tags:
- lineage
@@ -29,13 +39,18 @@ paths:
content:
application/json:
schema:
- oneOf:
- - $ref: '#/components/schemas/RunEvent'
- - $ref: '#/components/schemas/DatasetEvent'
- - $ref: '#/components/schemas/JobEvent'
+ $ref: '#/components/schemas/RunEvent'
responses:
- "200":
- description: OK
+ "201":
+ description: Created
+ "400":
+ $ref: "./openapi.yaml#/components/responses/BadRequestErrorResponse"
+ "403":
+ description: Forbidden - The caller cannot access at least one input
or output dataset.
+ content:
+ application/vnd.gravitino.v1+json:
+ schema:
+ $ref: "./openapi.yaml#/components/schemas/ErrorModel"
"429":
description: Too Many Requests
"5xx":
diff --git a/lineage/build.gradle.kts b/lineage/build.gradle.kts
index 81f3442064..26951eab2e 100644
--- a/lineage/build.gradle.kts
+++ b/lineage/build.gradle.kts
@@ -22,6 +22,7 @@ plugins {
}
dependencies {
+ implementation(project(":api"))
implementation(project(":common"))
implementation(project(":core"))
implementation(project(":server-common"))
diff --git
a/lineage/src/main/java/org/apache/gravitino/lineage/source/rest/LineageEventValidator.java
b/lineage/src/main/java/org/apache/gravitino/lineage/source/rest/LineageEventValidator.java
new file mode 100644
index 0000000000..db62ef859e
--- /dev/null
+++
b/lineage/src/main/java/org/apache/gravitino/lineage/source/rest/LineageEventValidator.java
@@ -0,0 +1,73 @@
+/*
+ * 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.lineage.source.rest;
+
+import com.google.common.base.Preconditions;
+import io.openlineage.server.OpenLineage.Dataset;
+import io.openlineage.server.OpenLineage.RunEvent;
+import java.util.List;
+import org.apache.commons.lang3.StringUtils;
+
+/** Validates the required fields of an OpenLineage run event. */
+public final class LineageEventValidator {
+
+ private LineageEventValidator() {}
+
+ /**
+ * Validates the required fields of an OpenLineage run event.
+ *
+ * @param event event to validate
+ * @throws IllegalArgumentException if a required field is absent or blank
+ */
+ public static void validate(RunEvent event) {
+ Preconditions.checkArgument(event != null, "Lineage event cannot be null");
+ Preconditions.checkArgument(event.getEventTime() != null, "eventTime is
required");
+ Preconditions.checkArgument(event.getProducer() != null, "producer is
required");
+ Preconditions.checkArgument(event.getSchemaURL() != null, "schemaURL is
required");
+ Preconditions.checkArgument(event.getRun() != null, "run is required");
+ Preconditions.checkArgument(event.getRun().getRunId() != null, "run.runId
is required");
+ Preconditions.checkArgument(event.getJob() != null, "job is required");
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(event.getJob().getNamespace()), "job.namespace
is required");
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(event.getJob().getName()), "job.name is
required");
+
+ validateDatasets(event.getInputs(), "inputs");
+ validateDatasets(event.getOutputs(), "outputs");
+ }
+
+ private static void validateDatasets(List<? extends Dataset> datasets,
String fieldName) {
+ if (datasets == null) {
+ return;
+ }
+
+ for (int index = 0; index < datasets.size(); index++) {
+ Dataset dataset = datasets.get(index);
+ Preconditions.checkArgument(dataset != null, "%s[%s] cannot be null",
fieldName, index);
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(dataset.getNamespace()),
+ "%s[%s].namespace is required",
+ fieldName,
+ index);
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(dataset.getName()), "%s[%s].name is
required", fieldName, index);
+ }
+ }
+}
diff --git
a/lineage/src/main/java/org/apache/gravitino/lineage/source/rest/LineageOperations.java
b/lineage/src/main/java/org/apache/gravitino/lineage/source/rest/LineageOperations.java
index 2f5fa472b6..ff2e6db9a5 100644
---
a/lineage/src/main/java/org/apache/gravitino/lineage/source/rest/LineageOperations.java
+++
b/lineage/src/main/java/org/apache/gravitino/lineage/source/rest/LineageOperations.java
@@ -32,6 +32,9 @@ import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import org.apache.gravitino.lineage.LineageDispatcher;
import org.apache.gravitino.metrics.MetricNames;
+import
org.apache.gravitino.server.authorization.annotations.AuthorizationExpression;
+import
org.apache.gravitino.server.authorization.annotations.AuthorizationRequest;
+import
org.apache.gravitino.server.authorization.expression.AuthorizationExpressionConstants;
import org.apache.gravitino.server.web.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,7 +43,7 @@ import org.slf4j.LoggerFactory;
public class LineageOperations {
private static final Logger LOG =
LoggerFactory.getLogger(LineageOperations.class);
- private LineageDispatcher lineageDispatcher;
+ private final LineageDispatcher lineageDispatcher;
@Context private HttpServletRequest httpRequest;
@@ -53,16 +56,25 @@ public class LineageOperations {
@Produces(MediaType.APPLICATION_JSON)
@Timed(name = "post-lineage." + MetricNames.HTTP_PROCESS_DURATION, absolute
= true)
@ResponseMetered(name = "post-lineage", absolute = true)
- public Response postLineage(OpenLineage.RunEvent event) {
- LOG.info(
- "Open lineage event, run id:{}, job name:{}",
- org.apache.gravitino.lineage.Utils.getRunID(event),
- org.apache.gravitino.lineage.Utils.getJobName(event));
+ @AuthorizationExpression(expression =
AuthorizationExpressionConstants.CAN_ACCESS_METADATA)
+ public Response postLineage(
+ @AuthorizationRequest(type = AuthorizationRequest.RequestType.LINEAGE)
+ OpenLineage.RunEvent event) {
+ try {
+ LineageEventValidator.validate(event);
+ } catch (IllegalArgumentException e) {
+ LOG.warn("Invalid lineage event", e);
+ return Utils.illegalArguments(e.getMessage(), e);
+ }
try {
return Utils.doAs(
httpRequest,
() -> {
+ LOG.info(
+ "Open lineage event, run id:{}, job name:{}",
+ org.apache.gravitino.lineage.Utils.getRunID(event),
+ org.apache.gravitino.lineage.Utils.getJobName(event));
if (lineageDispatcher.dispatchLineageEvent(event)) {
return Utils.created();
} else {
diff --git
a/lineage/src/test/java/org/apache/gravitino/lineage/source/TestLineageOperations.java
b/lineage/src/test/java/org/apache/gravitino/lineage/source/TestLineageOperations.java
index 714f80171b..3fa3e0ccc7 100644
---
a/lineage/src/test/java/org/apache/gravitino/lineage/source/TestLineageOperations.java
+++
b/lineage/src/test/java/org/apache/gravitino/lineage/source/TestLineageOperations.java
@@ -92,7 +92,7 @@ public class TestLineageOperations extends JerseyTest {
@SneakyThrows
@Test
- public void testUpdateLineageSucc() {
+ public void testAuthorizationDisabledAcceptsGenericNamespaces() {
RunEvent runEvent = createRunEvent();
Mockito.when(lineageDispatcher.dispatchLineageEvent(ArgumentMatchers.any())).thenReturn(true);
Response resp =
@@ -103,6 +103,23 @@ public class TestLineageOperations extends JerseyTest {
Assertions.assertEquals(Status.CREATED.getStatusCode(), resp.getStatus());
}
+ @SneakyThrows
+ @Test
+ void testDispatcherIllegalArgumentExceptionReturnsInternalError() {
+ RunEvent runEvent = createRunEvent();
+
Mockito.when(lineageDispatcher.dispatchLineageEvent(ArgumentMatchers.any()))
+ .thenThrow(new IllegalArgumentException("Dispatcher failure"));
+
+ Response resp =
+ target("/lineage")
+ .request(MediaType.APPLICATION_JSON_TYPE)
+ .accept(MediaType.APPLICATION_JSON_TYPE)
+ .post(Entity.entity(runEvent, MediaType.APPLICATION_JSON_TYPE));
+
+ Assertions.assertEquals(Status.INTERNAL_SERVER_ERROR.getStatusCode(),
resp.getStatus());
+
Mockito.verify(lineageDispatcher).dispatchLineageEvent(ArgumentMatchers.any());
+ }
+
@SneakyThrows
@Test
public void testUpdateLineageFailed() {
@@ -116,6 +133,18 @@ public class TestLineageOperations extends JerseyTest {
Assertions.assertEquals(Status.TOO_MANY_REQUESTS.getStatusCode(),
resp.getStatus());
}
+ @Test
+ public void testRejectInvalidLineageEvent() {
+ Response resp =
+ target("/lineage")
+ .request(MediaType.APPLICATION_JSON_TYPE)
+ .accept(MediaType.APPLICATION_JSON_TYPE)
+ .post(Entity.entity("{}", MediaType.APPLICATION_JSON_TYPE));
+
+ Assertions.assertEquals(Status.BAD_REQUEST.getStatusCode(),
resp.getStatus());
+ Mockito.verifyNoInteractions(lineageDispatcher);
+ }
+
private RunEvent createRunEvent() {
URI producer = URI.create("producer");
OpenLineage ol = new OpenLineage(producer);
diff --git
a/lineage/src/test/java/org/apache/gravitino/lineage/source/rest/TestLineageEventValidator.java
b/lineage/src/test/java/org/apache/gravitino/lineage/source/rest/TestLineageEventValidator.java
new file mode 100644
index 0000000000..be648e0158
--- /dev/null
+++
b/lineage/src/test/java/org/apache/gravitino/lineage/source/rest/TestLineageEventValidator.java
@@ -0,0 +1,153 @@
+/*
+ * 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.lineage.source.rest;
+
+import io.openlineage.server.OpenLineage.InputDataset;
+import io.openlineage.server.OpenLineage.Job;
+import io.openlineage.server.OpenLineage.OutputDataset;
+import io.openlineage.server.OpenLineage.Run;
+import io.openlineage.server.OpenLineage.RunEvent;
+import io.openlineage.server.OpenLineage.RunEvent.EventType;
+import java.net.URI;
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
+import java.util.Arrays;
+import java.util.List;
+import java.util.UUID;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class TestLineageEventValidator {
+
+ private static final ZonedDateTime EVENT_TIME =
ZonedDateTime.now(ZoneOffset.UTC);
+ private static final URI PRODUCER =
URI.create("https://gravitino.apache.org/test";);
+ private static final URI SCHEMA_URL =
+
URI.create("https://openlineage.io/spec/2-0-2/OpenLineage.json#/$defs/RunEvent";);
+
+ @Test
+ void testAcceptValidEvent() {
+ Assertions.assertDoesNotThrow(() ->
LineageEventValidator.validate(validEvent()));
+ }
+
+ @Test
+ void testRejectMissingRequiredFields() {
+ assertInvalid("Lineage event cannot be null", null);
+ assertInvalid(
+ "eventTime is required",
+ event(null, PRODUCER, SCHEMA_URL, validRun(), validJob(), null, null));
+ assertInvalid(
+ "producer is required",
+ event(EVENT_TIME, null, SCHEMA_URL, validRun(), validJob(), null,
null));
+ assertInvalid(
+ "schemaURL is required",
+ event(EVENT_TIME, PRODUCER, null, validRun(), validJob(), null, null));
+ assertInvalid(
+ "run is required", event(EVENT_TIME, PRODUCER, SCHEMA_URL, null,
validJob(), null, null));
+ assertInvalid(
+ "run.runId is required",
+ event(EVENT_TIME, PRODUCER, SCHEMA_URL, new Run(null, null),
validJob(), null, null));
+ assertInvalid(
+ "job is required", event(EVENT_TIME, PRODUCER, SCHEMA_URL, validRun(),
null, null, null));
+ assertInvalid(
+ "job.namespace is required",
+ event(EVENT_TIME, PRODUCER, SCHEMA_URL, validRun(), new Job(" ",
"job", null), null, null));
+ assertInvalid(
+ "job.name is required",
+ event(
+ EVENT_TIME,
+ PRODUCER,
+ SCHEMA_URL,
+ validRun(),
+ new Job("namespace", " ", null),
+ null,
+ null));
+ }
+
+ @Test
+ void testRejectInvalidDatasets() {
+ assertInvalid(
+ "inputs[0] cannot be null",
+ event(
+ EVENT_TIME,
+ PRODUCER,
+ SCHEMA_URL,
+ validRun(),
+ validJob(),
+ Arrays.asList((InputDataset) null),
+ null));
+ assertInvalid(
+ "inputs[0].namespace is required",
+ event(
+ EVENT_TIME,
+ PRODUCER,
+ SCHEMA_URL,
+ validRun(),
+ validJob(),
+ List.of(new InputDataset(" ", "catalog.schema.table", null, null)),
+ null));
+ assertInvalid(
+ "outputs[0].name is required",
+ event(
+ EVENT_TIME,
+ PRODUCER,
+ SCHEMA_URL,
+ validRun(),
+ validJob(),
+ null,
+ List.of(new OutputDataset("metalake", " ", null, null))));
+ }
+
+ private static RunEvent validEvent() {
+ return event(
+ EVENT_TIME,
+ PRODUCER,
+ SCHEMA_URL,
+ validRun(),
+ validJob(),
+ List.of(new InputDataset("metalake", "catalog.schema.table", null,
null)),
+ List.of(new OutputDataset("metalake", "catalog.schema.table", null,
null)));
+ }
+
+ private static RunEvent event(
+ ZonedDateTime eventTime,
+ URI producer,
+ URI schemaURL,
+ Run run,
+ Job job,
+ List<InputDataset> inputs,
+ List<OutputDataset> outputs) {
+ return new RunEvent(eventTime, producer, schemaURL, EventType.START, run,
job, inputs, outputs);
+ }
+
+ private static Run validRun() {
+ return new Run(UUID.randomUUID(), null);
+ }
+
+ private static Job validJob() {
+ return new Job("namespace", "job", null);
+ }
+
+ private static void assertInvalid(String message, RunEvent event) {
+ IllegalArgumentException exception =
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
LineageEventValidator.validate(event));
+ Assertions.assertEquals(message, exception.getMessage());
+ }
+}
diff --git
a/server-common/src/main/java/org/apache/gravitino/server/authorization/annotations/AuthorizationRequest.java
b/server-common/src/main/java/org/apache/gravitino/server/authorization/annotations/AuthorizationRequest.java
index 3d84fa84f2..2492fc9976 100644
---
a/server-common/src/main/java/org/apache/gravitino/server/authorization/annotations/AuthorizationRequest.java
+++
b/server-common/src/main/java/org/apache/gravitino/server/authorization/annotations/AuthorizationRequest.java
@@ -33,6 +33,7 @@ public @interface AuthorizationRequest {
ASSOCIATE_TAG,
ASSOCIATE_POLICY,
RUN_JOB,
+ LINEAGE,
LOAD_TABLE,
CREATE_SCHEMA
}
diff --git a/server/build.gradle.kts b/server/build.gradle.kts
index e417e70927..c5627f62b1 100644
--- a/server/build.gradle.kts
+++ b/server/build.gradle.kts
@@ -39,6 +39,9 @@ dependencies {
implementation(libs.jackson.datatype.jsr310)
implementation(libs.jackson.databind)
implementation(libs.metrics.jersey2)
+ implementation(libs.openlineage.java) {
+ isTransitive = false
+ }
// As of Java 9 or newer, the javax.activation package (needed by the jetty
server) is no longer part of the JDK. It was removed because it was part of the
// JavaBeans Activation Framework (JAF) which has been removed from Java SE.
So we need to add it as a dependency. For more,
diff --git
a/server/src/main/java/org/apache/gravitino/server/web/filter/GravitinoInterceptionService.java
b/server/src/main/java/org/apache/gravitino/server/web/filter/GravitinoInterceptionService.java
index 551a443771..d515cec330 100644
---
a/server/src/main/java/org/apache/gravitino/server/web/filter/GravitinoInterceptionService.java
+++
b/server/src/main/java/org/apache/gravitino/server/web/filter/GravitinoInterceptionService.java
@@ -43,8 +43,10 @@ import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.authorization.AuthorizationRequestContext;
import org.apache.gravitino.authorization.AuthorizationUtils;
+import org.apache.gravitino.exceptions.BadRequestException;
import org.apache.gravitino.exceptions.ForbiddenException;
import org.apache.gravitino.exceptions.NoSuchMetalakeException;
+import org.apache.gravitino.lineage.source.rest.LineageOperations;
import
org.apache.gravitino.listener.api.event.server.AuthorizationDenialFailureEvent;
import
org.apache.gravitino.server.authorization.annotations.AuthorizationExpression;
import
org.apache.gravitino.server.authorization.annotations.AuthorizationRequest;
@@ -113,7 +115,8 @@ public class GravitinoInterceptionService implements
InterceptionService {
PolicyOperations.class.getName(),
MetadataObjectPolicyOperations.class.getName(),
JobOperations.class.getName(),
- MetadataObjectCredentialOperations.class.getName()));
+ MetadataObjectCredentialOperations.class.getName(),
+ LineageOperations.class.getName()));
}
@Override
@@ -150,7 +153,7 @@ public class GravitinoInterceptionService implements
InterceptionService {
method.getAnnotation(AuthorizationExpression.class);
try {
- AuthorizationExecutor executor;
+ AuthorizationExecutor executor = null;
if (expressionAnnotation != null) {
String expression = expressionAnnotation.expression();
Object[] args = methodInvocation.getArguments();
@@ -162,34 +165,21 @@ public class GravitinoInterceptionService implements
InterceptionService {
AuthorizationRequestContext authorizationRequestContext =
new AuthorizationRequestContext();
- // Check metalake and user existence before authorization
+ Optional<String> authorizationMetalake = Optional.empty();
NameIdentifier metalakeIdent =
metadataContext.get(Entity.EntityType.METALAKE);
if (metalakeIdent != null) {
- String currentUser = PrincipalUtils.getCurrentUserName();
- try {
- AuthorizationUtils.checkCurrentUser(
- metalakeIdent.name(), currentUser,
authorizationRequestContext);
- } catch (NoSuchMetalakeException e) {
- LOG.warn(
- "Metalake {} does not exist when validating user {}",
metalakeIdent, currentUser);
- // Not a real authz denial — metalake is absent, not forbidden.
Skip event dispatch;
- // HttpAuditFilter will emit a generic HttpRequestFailureEvent
for this 403.
- return buildNoAuthResponse(expressionAnnotation,
metadataContext, method, expression);
- } catch (ForbiddenException ex) {
- LOG.warn(
- "User validation failed - User: {}, Metalake: {}, Reason:
{}",
- currentUser,
- metalakeIdent.name(),
- ex.getMessage());
- dispatchAuthzDenialEvent(currentUser, metalakeIdent,
method.getName(), expression);
- return Utils.forbidden(ex.getMessage(), ex);
- } catch (Exception ex) {
- LOG.error(
- "Unexpected error during user validation - User: {},
Metalake: {}",
- currentUser,
- metalakeIdent.name(),
- ex);
- return Utils.internalError("Failed to validate user", ex);
+ authorizationMetalake = Optional.of(metalakeIdent.name());
+ Optional<Response> validationFailure =
+ validateCurrentUser(
+ metalakeIdent,
+ authorizationRequestContext,
+ expressionAnnotation,
+ metadataContext,
+ method,
+ expression,
+ false);
+ if (validationFailure.isPresent()) {
+ return validationFailure.get();
}
}
@@ -211,7 +201,46 @@ public class GravitinoInterceptionService implements
InterceptionService {
args,
secondaryExpression,
secondaryExpressionCondition);
- boolean authorizeResult =
executor.execute(authorizationRequestContext);
+ Optional<String> dynamicMetalake;
+ try {
+ dynamicMetalake = executor.getAuthorizationMetalake();
+ if (dynamicMetalake.isPresent()
+ && authorizationMetalake.isPresent()
+ &&
!dynamicMetalake.get().equals(authorizationMetalake.get())) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Authorization request metalake '%s' does not match
path metalake '%s'",
+ dynamicMetalake.get(), authorizationMetalake.get()));
+ }
+ } catch (IllegalArgumentException exception) {
+ LOG.warn("Invalid authorization request", exception);
+ return Utils.illegalArguments(exception.getMessage(), exception);
+ }
+
+ if (dynamicMetalake.isPresent() &&
authorizationMetalake.isEmpty()) {
+ Optional<Response> validationFailure =
+ validateCurrentUser(
+ NameIdentifier.of(dynamicMetalake.get()),
+ authorizationRequestContext,
+ expressionAnnotation,
+ metadataContext,
+ method,
+ expression,
+ true);
+ if (validationFailure.isPresent()) {
+ return validationFailure.get();
+ }
+ }
+ }
+
+ if (executor != null) {
+ boolean authorizeResult;
+ try {
+ authorizeResult = executor.execute(authorizationRequestContext);
+ } catch (BadRequestException exception) {
+ LOG.warn("Invalid authorization request", exception);
+ return Utils.illegalArguments(exception.getMessage(), exception);
+ }
if (!authorizeResult) {
MetadataObject.Type type =
expressionAnnotation.accessMetadataType();
NameIdentifier accessMetadataName =
@@ -240,6 +269,51 @@ public class GravitinoInterceptionService implements
InterceptionService {
}
}
+ private Optional<Response> validateCurrentUser(
+ NameIdentifier metalakeIdent,
+ AuthorizationRequestContext authorizationRequestContext,
+ AuthorizationExpression expressionAnnotation,
+ Map<Entity.EntityType, NameIdentifier> metadataContext,
+ Method method,
+ String expression,
+ boolean dynamicMetalake) {
+ String currentUser = PrincipalUtils.getCurrentUserName();
+ try {
+ AuthorizationUtils.checkCurrentUser(
+ metalakeIdent.name(), currentUser, authorizationRequestContext);
+ } catch (NoSuchMetalakeException e) {
+ LOG.warn("Metalake {} does not exist when validating user {}",
metalakeIdent, currentUser);
+ if (dynamicMetalake) {
+ return Optional.of(
+ Utils.illegalArguments(
+ String.format(
+ "job.namespace must identify an existing metalake: %s",
metalakeIdent.name()),
+ e));
+ }
+ // Not a real authz denial — metalake is absent, not forbidden. Skip
event dispatch;
+ // HttpAuditFilter will emit a generic HttpRequestFailureEvent for
this 403.
+ return Optional.of(
+ buildNoAuthResponse(expressionAnnotation, metadataContext, method,
expression));
+ } catch (ForbiddenException ex) {
+ LOG.warn(
+ "User validation failed - User: {}, Metalake: {}, Reason: {}",
+ currentUser,
+ metalakeIdent.name(),
+ ex.getMessage());
+ dispatchAuthzDenialEvent(currentUser, metalakeIdent, method.getName(),
expression);
+ return Optional.of(Utils.forbidden(ex.getMessage(), ex));
+ } catch (Exception ex) {
+ LOG.error(
+ "Unexpected error during user validation - User: {}, Metalake: {}",
+ currentUser,
+ metalakeIdent.name(),
+ ex);
+ return Optional.of(Utils.internalError("Failed to validate user", ex));
+ }
+
+ return Optional.empty();
+ }
+
private Response buildNoAuthResponse(
AuthorizationExpression expressionAnnotation,
Map<Entity.EntityType, NameIdentifier> metadataContext,
diff --git
a/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/AuthorizationExecutor.java
b/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/AuthorizationExecutor.java
index d98d448a10..50111c58a9 100644
---
a/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/AuthorizationExecutor.java
+++
b/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/AuthorizationExecutor.java
@@ -17,9 +17,19 @@
package org.apache.gravitino.server.web.filter.authorization;
+import java.util.Optional;
import org.apache.gravitino.authorization.AuthorizationRequestContext;
public interface AuthorizationExecutor {
+ /**
+ * Returns the metalake dynamically resolved from the authorization request.
+ *
+ * @return the metalake that requires user and active-role validation
+ */
+ default Optional<String> getAuthorizationMetalake() {
+ return Optional.empty();
+ }
+
boolean execute(AuthorizationRequestContext authorizationRequestContext)
throws Exception;
}
diff --git
a/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/AuthorizeExecutorFactory.java
b/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/AuthorizeExecutorFactory.java
index 4d2f4c63aa..027f949abc 100644
---
a/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/AuthorizeExecutorFactory.java
+++
b/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/AuthorizeExecutorFactory.java
@@ -46,6 +46,7 @@ public class AuthorizeExecutorFactory {
expression, parameters, args, metadataContext, pathParams,
entityType);
case RUN_JOB -> new RunJobAuthorizationExecutor(
parameters, args, expression, metadataContext, pathParams,
entityType);
+ case LINEAGE -> new LineageAuthorizationExecutor(parameters, args,
expression);
case LOAD_TABLE -> new LoadTableAuthorizationExecutor(
parameters,
args,
diff --git
a/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/LineageAuthorizationExecutor.java
b/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/LineageAuthorizationExecutor.java
new file mode 100644
index 0000000000..5af3a60d62
--- /dev/null
+++
b/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/LineageAuthorizationExecutor.java
@@ -0,0 +1,188 @@
+/*
+ * 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.ArrayList;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+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.exceptions.BadRequestException;
+import org.apache.gravitino.lineage.source.rest.LineageEventValidator;
+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;
+ private List<AuthorizationTarget> authorizationTargets = List.of();
+ private boolean authorizationTargetsResolved;
+
+ /**
+ * 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 Optional<String> getAuthorizationMetalake() {
+ RunEvent event = extractRunEvent();
+ Preconditions.checkArgument(event.getJob() != null, "job is required");
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(event.getJob().getNamespace()), "job.namespace
is required");
+ return Optional.of(event.getJob().getNamespace());
+ }
+
+ @Override
+ public boolean execute(AuthorizationRequestContext context) {
+ if (!authorizationTargetsResolved) {
+ try {
+ resolveAuthorizationTargets(extractRunEvent());
+ } catch (IllegalArgumentException e) {
+ throw new BadRequestException(e, "%s", e.getMessage());
+ }
+ }
+
+ AuthorizationExpressionEvaluator evaluator = new
AuthorizationExpressionEvaluator(expression);
+ context.setOriginalAuthorizationExpression(expression);
+
+ // Dataset-less OpenLineage events are valid; the interceptor has already
validated metalake
+ // membership before this executor runs.
+ for (AuthorizationTarget target : authorizationTargets) {
+ if (!evaluator.evaluate(
+ target.metadataContext, Map.of(), context,
Optional.of(target.entityType.name()))) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ 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" -> MetadataObject.Type.MODEL;
+ case "TOPIC" -> MetadataObject.Type.TOPIC;
+ default -> throw new IllegalArgumentException("Unsupported dataset type:
" + datasetType);
+ };
+ }
+
+ private RunEvent extractRunEvent() {
+ Object request = extractFromParameters(parameters, args);
+ Preconditions.checkArgument(request instanceof RunEvent, "Lineage request
must be a RunEvent");
+ return (RunEvent) request;
+ }
+
+ private void resolveAuthorizationTargets(RunEvent event) {
+ if (authorizationTargetsResolved) {
+ return;
+ }
+
+ LineageEventValidator.validate(event);
+ String metalake = event.getJob().getNamespace();
+ List<AuthorizationTarget> targets = new ArrayList<>();
+ resolveAuthorizationTargets(event.getInputs(), "inputs", metalake,
targets);
+ resolveAuthorizationTargets(event.getOutputs(), "outputs", metalake,
targets);
+ authorizationTargets = List.copyOf(targets);
+ authorizationTargetsResolved = true;
+ }
+
+ private static void resolveAuthorizationTargets(
+ List<? extends Dataset> datasets,
+ String fieldName,
+ String metalake,
+ List<AuthorizationTarget> targets) {
+ if (datasets == null) {
+ return;
+ }
+
+ for (int index = 0; index < datasets.size(); index++) {
+ Dataset dataset = datasets.get(index);
+ Preconditions.checkArgument(
+ metalake.equals(dataset.getNamespace()),
+ "%s[%s].namespace must match job.namespace '%s'",
+ fieldName,
+ index,
+ metalake);
+
+ 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);
+ targets.add(new AuthorizationTarget(metadataContext, entityType));
+ }
+ }
+
+ private static class AuthorizationTarget {
+ private final Map<Entity.EntityType, NameIdentifier> metadataContext;
+ private final Entity.EntityType entityType;
+
+ private AuthorizationTarget(
+ Map<Entity.EntityType, NameIdentifier> metadataContext,
Entity.EntityType entityType) {
+ this.metadataContext = metadataContext;
+ this.entityType = entityType;
+ }
+ }
+}
diff --git
a/server/src/test/java/org/apache/gravitino/server/web/filter/TestGravitinoInterceptionService.java
b/server/src/test/java/org/apache/gravitino/server/web/filter/TestGravitinoInterceptionService.java
index 69d6b17f15..93f3681170 100644
---
a/server/src/test/java/org/apache/gravitino/server/web/filter/TestGravitinoInterceptionService.java
+++
b/server/src/test/java/org/apache/gravitino/server/web/filter/TestGravitinoInterceptionService.java
@@ -47,6 +47,7 @@ import org.apache.gravitino.authorization.AuthorizationUtils;
import org.apache.gravitino.authorization.GravitinoAuthorizer;
import org.apache.gravitino.authorization.Privilege;
import org.apache.gravitino.catalog.ViewDispatcher;
+import org.apache.gravitino.dto.requests.SchemaCreateRequest;
import org.apache.gravitino.dto.responses.ErrorResponse;
import org.apache.gravitino.exceptions.ForbiddenException;
import org.apache.gravitino.exceptions.NoSuchMetalakeException;
@@ -57,6 +58,7 @@ import
org.apache.gravitino.server.authorization.GravitinoAuthorizerProvider;
import
org.apache.gravitino.server.authorization.annotations.AuthorizationExpression;
import
org.apache.gravitino.server.authorization.annotations.AuthorizationMetadata;
import org.apache.gravitino.server.web.Utils;
+import org.apache.gravitino.server.web.rest.SchemaOperations;
import org.apache.gravitino.server.web.rest.ViewOperations;
import org.apache.gravitino.utils.PrincipalUtils;
import org.apache.gravitino.utils.RequestContext;
@@ -77,6 +79,40 @@ public class TestGravitinoInterceptionService {
RequestContext.clear();
}
+ @Test
+ public void testPathMetalakeValidationPrecedesExecutorConstruction() throws
Throwable {
+ Method method =
+ SchemaOperations.class.getMethod(
+ "createSchema", String.class, String.class,
SchemaCreateRequest.class);
+ MethodInvocation invocation = mock(MethodInvocation.class);
+ SchemaCreateRequest malformedRequest = mock(SchemaCreateRequest.class);
+ when(invocation.getMethod()).thenReturn(method);
+ when(invocation.getArguments())
+ .thenReturn(new Object[] {"metalake", "catalog", malformedRequest});
+
+ try (MockedStatic<PrincipalUtils> principalUtils =
mockStatic(PrincipalUtils.class);
+ MockedStatic<AuthorizationUtils> authorizationUtils =
+ mockStatic(AuthorizationUtils.class)) {
+
principalUtils.when(PrincipalUtils::getCurrentUserName).thenReturn("tester");
+ authorizationUtils
+ .when(
+ () ->
+ AuthorizationUtils.checkCurrentUser(
+ ArgumentMatchers.eq("metalake"),
+ ArgumentMatchers.eq("tester"),
+ any(AuthorizationRequestContext.class)))
+ .thenThrow(new ForbiddenException("User tester is not a member"));
+
+ MethodInterceptor interceptor =
+ new
GravitinoInterceptionService().getMethodInterceptors(method).get(0);
+ Response response = (Response) interceptor.invoke(invocation);
+
+ assertEquals(Response.Status.FORBIDDEN.getStatusCode(),
response.getStatus());
+ verify(malformedRequest, never()).getName();
+ verify(invocation, never()).proceed();
+ }
+ }
+
@Test
public void testViewOperationsIsRegisteredForInterception() {
Descriptor descriptor = mock(Descriptor.class);
diff --git
a/server/src/test/java/org/apache/gravitino/server/web/filter/authorization/TestLineageAuthorizationExecutor.java
b/server/src/test/java/org/apache/gravitino/server/web/filter/authorization/TestLineageAuthorizationExecutor.java
new file mode 100644
index 0000000000..2829baf2ca
--- /dev/null
+++
b/server/src/test/java/org/apache/gravitino/server/web/filter/authorization/TestLineageAuthorizationExecutor.java
@@ -0,0 +1,478 @@
+/*
+ * 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.authorization.expression.AuthorizationExpressionConstants.CAN_ACCESS_METADATA;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+import io.openlineage.server.OpenLineage.DatasetFacet;
+import io.openlineage.server.OpenLineage.DatasetFacets;
+import io.openlineage.server.OpenLineage.InputDataset;
+import io.openlineage.server.OpenLineage.Job;
+import io.openlineage.server.OpenLineage.OutputDataset;
+import io.openlineage.server.OpenLineage.Run;
+import io.openlineage.server.OpenLineage.RunEvent;
+import io.openlineage.server.OpenLineage.RunEvent.EventType;
+import java.lang.reflect.Method;
+import java.net.URI;
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import javax.ws.rs.core.Response;
+import org.aopalliance.intercept.MethodInterceptor;
+import org.aopalliance.intercept.MethodInvocation;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.MetadataObject;
+import org.apache.gravitino.MetadataObjects;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.UserPrincipal;
+import org.apache.gravitino.authorization.AuthorizationRequestContext;
+import org.apache.gravitino.authorization.AuthorizationUtils;
+import org.apache.gravitino.authorization.GravitinoAuthorizer;
+import org.apache.gravitino.authorization.Privilege;
+import org.apache.gravitino.dto.responses.ErrorResponse;
+import org.apache.gravitino.exceptions.BadRequestException;
+import org.apache.gravitino.exceptions.ForbiddenException;
+import org.apache.gravitino.exceptions.NoSuchMetalakeException;
+import org.apache.gravitino.lineage.source.rest.LineageOperations;
+import org.apache.gravitino.server.authorization.GravitinoAuthorizerProvider;
+import
org.apache.gravitino.server.authorization.annotations.AuthorizationRequest;
+import
org.apache.gravitino.server.authorization.annotations.ExpressionCondition;
+import org.apache.gravitino.server.web.filter.GravitinoInterceptionService;
+import org.apache.gravitino.utils.PrincipalUtils;
+import org.glassfish.hk2.api.Descriptor;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.MockedStatic;
+
+class TestLineageAuthorizationExecutor {
+
+ private static final String METALAKE = "metalake";
+ private static final String DATASET_NAME = "catalog.schema.object";
+ private static final URI PRODUCER =
URI.create("https://gravitino.apache.org/test";);
+ private static final URI SCHEMA_URL =
+
URI.create("https://openlineage.io/spec/2-0-2/OpenLineage.json#/$defs/RunEvent";);
+
+ @Test
+ void testRegisterWithStandardInterceptionFlow() throws Exception {
+ GravitinoInterceptionService service = new GravitinoInterceptionService();
+ Descriptor descriptor = mock(Descriptor.class);
+
when(descriptor.getImplementation()).thenReturn(LineageOperations.class.getName());
+ Method method = lineageMethod();
+
+ Assertions.assertTrue(service.getDescriptorFilter().matches(descriptor));
+ Assertions.assertEquals(1, service.getMethodInterceptors(method).size());
+ Assertions.assertTrue(
+ service
+ .getMethodInterceptors(method)
+ .get(0)
+ .getClass()
+ .getSimpleName()
+ .contains("MetadataAuthorization"));
+ Assertions.assertInstanceOf(LineageAuthorizationExecutor.class,
createWithFactory(event()));
+ }
+
+ @Test
+ void testExposeEventOrganizationAsAuthorizationMetalake() throws Exception {
+ Assertions.assertEquals(Optional.of(METALAKE),
executor(event()).getAuthorizationMetalake());
+ }
+
+ @Test
+ void testInterceptorRejectsUserOutsideDynamicMetalake() throws Throwable {
+ MethodInvocation invocation = invocation(event());
+
+ try (MockedStatic<PrincipalUtils> principalUtils =
mockStatic(PrincipalUtils.class);
+ MockedStatic<AuthorizationUtils> authorizationUtils =
+ mockStatic(AuthorizationUtils.class)) {
+
principalUtils.when(PrincipalUtils::getCurrentPrincipal).thenReturn(principal());
+
principalUtils.when(PrincipalUtils::getCurrentUserName).thenReturn("tester");
+ authorizationUtils
+ .when(
+ () ->
+ AuthorizationUtils.checkCurrentUser(
+ eq(METALAKE), eq("tester"),
any(AuthorizationRequestContext.class)))
+ .thenThrow(new ForbiddenException("User tester is not a member"));
+
+ Response response = (Response) lineageInterceptor().invoke(invocation);
+
+ Assertions.assertEquals(Response.Status.FORBIDDEN.getStatusCode(),
response.getStatus());
+ verify(invocation, never()).proceed();
+ }
+ }
+
+ @Test
+ void testRejectNonexistentDynamicMetalakeAsBadRequest() throws Throwable {
+ MethodInvocation invocation = invocation(event());
+
+ try (MockedStatic<PrincipalUtils> principalUtils =
mockStatic(PrincipalUtils.class);
+ MockedStatic<AuthorizationUtils> authorizationUtils =
+ mockStatic(AuthorizationUtils.class)) {
+
principalUtils.when(PrincipalUtils::getCurrentPrincipal).thenReturn(principal());
+
principalUtils.when(PrincipalUtils::getCurrentUserName).thenReturn("tester");
+ authorizationUtils
+ .when(
+ () ->
+ AuthorizationUtils.checkCurrentUser(
+ eq(METALAKE), eq("tester"),
any(AuthorizationRequestContext.class)))
+ .thenThrow(new NoSuchMetalakeException("Metalake does not exist"));
+
+ Response response = (Response) lineageInterceptor().invoke(invocation);
+
+ Assertions.assertEquals(Response.Status.BAD_REQUEST.getStatusCode(),
response.getStatus());
+ ErrorResponse errorResponse = (ErrorResponse) response.getEntity();
+ Assertions.assertTrue(
+ errorResponse.getMessage().contains("job.namespace"),
+ "The response should identify the invalid field");
+ verify(invocation, never()).proceed();
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({"inputs,0", "inputs,1", "outputs,0", "outputs,1"})
+ void testRejectCrossOrganizationDatasetAtEveryPosition(String field, int
index) throws Throwable {
+ List<InputDataset> inputs =
+ new ArrayList<>(List.of(input("catalog.schema.input0"),
input("catalog.schema.input1")));
+ List<OutputDataset> outputs =
+ new ArrayList<>(
+ List.of(output("catalog.schema.output0"),
output("catalog.schema.output1")));
+ if (field.equals("inputs")) {
+ inputs.set(
+ index, new InputDataset("anotherMetalake",
inputs.get(index).getName(), null, null));
+ } else {
+ outputs.set(
+ index, new OutputDataset("anotherMetalake",
outputs.get(index).getName(), null, null));
+ }
+
+ assertBadRequest(event(METALAKE, inputs, outputs));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"UNKNOWN", "MODEL_VERSION"})
+ void testRejectUnsupportedDatasetTypeAsBadRequest(String datasetType) throws
Throwable {
+ assertBadRequest(
+ event(METALAKE, List.of(dataset(datasetType, METALAKE, DATASET_NAME)),
List.of()));
+ }
+
+ @Test
+ void testRejectMalformedDatasetNameAsBadRequest() throws Throwable {
+ assertBadRequest(
+ event(METALAKE, List.of(dataset(null, METALAKE, "catalog.object")),
List.of()));
+ }
+
+ @Test
+ void testRejectMalformedEventAsBadRequest() throws Throwable {
+ assertBadRequest(null);
+ }
+
+ @Test
+ void testResolveTargetsOnlyDuringExecution() throws Exception {
+ LineageAuthorizationExecutor executor =
+ executor(event(METALAKE, List.of(input("catalog.object")), List.of()));
+
+ Assertions.assertEquals(Optional.of(METALAKE),
executor.getAuthorizationMetalake());
+ Assertions.assertThrows(
+ BadRequestException.class,
+ () -> execute(executor, mock(GravitinoAuthorizer.class), principal()));
+ }
+
+ @Test
+ void testAllowDatasetlessEvent() throws Exception {
+ GravitinoAuthorizer authorizer = mock(GravitinoAuthorizer.class);
+ LineageAuthorizationExecutor executor = executor(event(METALAKE,
List.of(), List.of()));
+
+ Assertions.assertEquals(Optional.of(METALAKE),
executor.getAuthorizationMetalake());
+ Assertions.assertTrue(execute(executor, authorizer, principal()));
+ verifyNoInteractions(authorizer);
+ }
+
+ @Test
+ void testOutputRequiresMetadataVisibilityOnly() throws Exception {
+ MetadataObject outputObject = MetadataObjects.parse(DATASET_NAME,
MetadataObject.Type.TABLE);
+ GravitinoAuthorizer authorizer = mock(GravitinoAuthorizer.class);
+ when(authorizer.authorize(
+ any(),
+ eq(METALAKE),
+ any(),
+ eq(Privilege.Name.USE_CATALOG),
+ any(AuthorizationRequestContext.class)))
+ .thenReturn(true);
+ when(authorizer.authorize(
+ any(),
+ eq(METALAKE),
+ any(),
+ eq(Privilege.Name.USE_SCHEMA),
+ any(AuthorizationRequestContext.class)))
+ .thenReturn(true);
+ when(authorizer.authorize(
+ any(),
+ eq(METALAKE),
+ eq(outputObject),
+ eq(Privilege.Name.SELECT_TABLE),
+ any(AuthorizationRequestContext.class)))
+ .thenReturn(true);
+
+ Assertions.assertTrue(
+ execute(
+ executor(event(METALAKE, List.of(),
List.of(output(DATASET_NAME)))),
+ authorizer,
+ principal()));
+ verify(authorizer)
+ .authorize(
+ any(),
+ eq(METALAKE),
+ eq(outputObject),
+ eq(Privilege.Name.SELECT_TABLE),
+ any(AuthorizationRequestContext.class));
+ }
+
+ @Test
+ void testPermissionDenialReturnsForbidden() throws Throwable {
+ GravitinoAuthorizer authorizer = mock(GravitinoAuthorizer.class);
+
+ Response response = intercept(event(), authorizer);
+
+ Assertions.assertEquals(Response.Status.FORBIDDEN.getStatusCode(),
response.getStatus());
+ }
+
+ @Test
+ void testAuthorizerIllegalArgumentExceptionReturnsInternalError() throws
Throwable {
+ GravitinoAuthorizer authorizer = mock(GravitinoAuthorizer.class);
+ when(authorizer.isOwner(any(), any(), any(), any()))
+ .thenThrow(new IllegalArgumentException("Authorizer backend failure"));
+
+ Response response = intercept(event(), authorizer);
+
+ Assertions.assertEquals(
+ Response.Status.INTERNAL_SERVER_ERROR.getStatusCode(),
response.getStatus());
+ }
+
+ @Test
+ void testAuthorizeAllInputAndOutputDatasets() throws Exception {
+ GravitinoAuthorizer authorizer = mock(GravitinoAuthorizer.class);
+ when(authorizer.isOwner(any(), any(), any(), any())).thenReturn(true);
+ RunEvent runEvent =
+ event(
+ METALAKE,
+ List.of(input("catalog.schema.input")),
+ List.of(output("catalog.schema.output")));
+
+ Assertions.assertTrue(execute(executor(runEvent, "TABLE::OWNER"),
authorizer, principal()));
+ }
+
+ @ParameterizedTest
+ @CsvSource({"inputs,0", "inputs,1", "outputs,0", "outputs,1"})
+ void testPermissionFailureAtEveryDatasetPosition(String field, int index)
throws Exception {
+ List<InputDataset> inputs =
+ List.of(input("catalog.schema.input0"),
input("catalog.schema.input1"));
+ List<OutputDataset> outputs =
+ List.of(output("catalog.schema.output0"),
output("catalog.schema.output1"));
+ String deniedName =
+ field.equals("inputs") ? inputs.get(index).getName() :
outputs.get(index).getName();
+ MetadataObject denied = MetadataObjects.parse(deniedName,
MetadataObject.Type.TABLE);
+ GravitinoAuthorizer authorizer = mock(GravitinoAuthorizer.class);
+ when(authorizer.isOwner(any(), eq(METALAKE), any(), any()))
+ .thenAnswer(call -> !denied.equals(call.getArgument(2)));
+
+ Assertions.assertFalse(
+ execute(
+ executor(event(METALAKE, inputs, outputs), "TABLE::OWNER"),
authorizer, principal()));
+ verify(authorizer).isOwner(any(), eq(METALAKE), eq(denied), any());
+ }
+
+ @ParameterizedTest
+ @CsvSource({
+ "DEFAULT,TABLE",
+ "TABLE,TABLE",
+ "VIEW,VIEW",
+ "FILE,FILESET",
+ "Fileset,FILESET",
+ "MODEL,MODEL",
+ "TOPIC,TOPIC"
+ })
+ void testResolveExactAuthorizationTarget(String datasetType,
MetadataObject.Type expectedType)
+ throws Exception {
+ String facetType = datasetType.equals("DEFAULT") ? null : datasetType;
+ InputDataset dataset = dataset(facetType, METALAKE, DATASET_NAME);
+ MetadataObject expected = MetadataObjects.parse(DATASET_NAME,
expectedType);
+ GravitinoAuthorizer authorizer = mock(GravitinoAuthorizer.class);
+ when(authorizer.isOwner(any(), eq(METALAKE), any(), any()))
+ .thenAnswer(call -> expected.equals(call.getArgument(2)));
+
+ Assertions.assertEquals(expectedType,
LineageAuthorizationExecutor.getMetadataType(dataset));
+ Assertions.assertTrue(
+ execute(
+ executor(event(METALAKE, List.of(dataset), List.of()),
expectedType.name() + "::OWNER"),
+ authorizer,
+ principal()));
+ verify(authorizer).isOwner(any(), eq(METALAKE), eq(expected), any());
+ }
+
+ private static void assertBadRequest(RunEvent event) throws Throwable {
+ MethodInvocation invocation = invocation(event);
+ try (MockedStatic<PrincipalUtils> principalUtils =
mockStatic(PrincipalUtils.class);
+ MockedStatic<AuthorizationUtils> authorizationUtils =
+ mockStatic(AuthorizationUtils.class)) {
+
principalUtils.when(PrincipalUtils::getCurrentPrincipal).thenReturn(principal());
+
principalUtils.when(PrincipalUtils::getCurrentUserName).thenReturn("tester");
+
+ Response response = (Response) lineageInterceptor().invoke(invocation);
+
+ Assertions.assertEquals(Response.Status.BAD_REQUEST.getStatusCode(),
response.getStatus());
+ verify(invocation, never()).proceed();
+ if (event == null) {
+ authorizationUtils.verifyNoInteractions();
+ } else {
+ authorizationUtils.verify(
+ () ->
+ AuthorizationUtils.checkCurrentUser(
+ eq(METALAKE), eq("tester"),
any(AuthorizationRequestContext.class)));
+ }
+ }
+ }
+
+ private static Response intercept(RunEvent event, GravitinoAuthorizer
authorizer)
+ throws Throwable {
+ MethodInvocation invocation = invocation(event);
+ try (MockedStatic<PrincipalUtils> principalUtils =
mockStatic(PrincipalUtils.class);
+ MockedStatic<AuthorizationUtils> authorizationUtils =
mockStatic(AuthorizationUtils.class);
+ MockedStatic<GravitinoAuthorizerProvider> providerStatic =
+ mockStatic(GravitinoAuthorizerProvider.class)) {
+
principalUtils.when(PrincipalUtils::getCurrentPrincipal).thenReturn(principal());
+
principalUtils.when(PrincipalUtils::getCurrentUserName).thenReturn("tester");
+ GravitinoAuthorizerProvider provider =
mock(GravitinoAuthorizerProvider.class);
+
providerStatic.when(GravitinoAuthorizerProvider::getInstance).thenReturn(provider);
+ when(provider.getGravitinoAuthorizer()).thenReturn(authorizer);
+
+ Response response = (Response) lineageInterceptor().invoke(invocation);
+ verify(invocation, never()).proceed();
+ return response;
+ }
+ }
+
+ private static boolean execute(
+ LineageAuthorizationExecutor executor,
+ GravitinoAuthorizer authorizer,
+ UserPrincipal principal)
+ throws Exception {
+ try (MockedStatic<GravitinoAuthorizerProvider> providerStatic =
+ mockStatic(GravitinoAuthorizerProvider.class)) {
+ GravitinoAuthorizerProvider provider =
mock(GravitinoAuthorizerProvider.class);
+
providerStatic.when(GravitinoAuthorizerProvider::getInstance).thenReturn(provider);
+ when(provider.getGravitinoAuthorizer()).thenReturn(authorizer);
+ return PrincipalUtils.doAs(
+ principal, () -> executor.execute(new
AuthorizationRequestContext()));
+ }
+ }
+
+ private static LineageAuthorizationExecutor executor(RunEvent event) throws
Exception {
+ return executor(event, CAN_ACCESS_METADATA);
+ }
+
+ private static LineageAuthorizationExecutor executor(RunEvent event, String
expression)
+ throws Exception {
+ return new LineageAuthorizationExecutor(
+ lineageMethod().getParameters(), new Object[] {event}, expression);
+ }
+
+ private static AuthorizationExecutor createWithFactory(RunEvent event)
throws Exception {
+ Map<Entity.EntityType, NameIdentifier> metadataContext = new HashMap<>();
+ return AuthorizeExecutorFactory.create(
+ CAN_ACCESS_METADATA,
+ AuthorizationRequest.RequestType.LINEAGE,
+ metadataContext,
+ Map.of(),
+ Optional.empty(),
+ lineageMethod().getParameters(),
+ new Object[] {event},
+ "",
+ ExpressionCondition.NEVER);
+ }
+
+ private static Method lineageMethod() throws NoSuchMethodException {
+ return LineageOperations.class.getMethod("postLineage", RunEvent.class);
+ }
+
+ private static MethodInterceptor lineageInterceptor() throws
NoSuchMethodException {
+ return new
GravitinoInterceptionService().getMethodInterceptors(lineageMethod()).get(0);
+ }
+
+ private static MethodInvocation invocation(RunEvent event) throws
NoSuchMethodException {
+ MethodInvocation invocation = mock(MethodInvocation.class);
+ when(invocation.getMethod()).thenReturn(lineageMethod());
+ when(invocation.getArguments()).thenReturn(new Object[] {event});
+ return invocation;
+ }
+
+ private static UserPrincipal principal() {
+ return new UserPrincipal("tester");
+ }
+
+ private static InputDataset input(String name) {
+ return new InputDataset(METALAKE, name, null, null);
+ }
+
+ private static OutputDataset output(String name) {
+ return new OutputDataset(METALAKE, name, null, null);
+ }
+
+ private static InputDataset dataset(String datasetType, String namespace,
String name) {
+ if (datasetType == null) {
+ return new InputDataset(namespace, name, null, null);
+ }
+
+ DatasetFacets facets = new DatasetFacets();
+ DatasetFacet typeFacet = new DatasetFacet(PRODUCER, SCHEMA_URL);
+ typeFacet.getAdditionalProperties().put("datasetType", datasetType);
+ facets.getAdditionalProperties().put("datasetType", typeFacet);
+ return new InputDataset(namespace, name, facets, null);
+ }
+
+ private static RunEvent event() {
+ return event(METALAKE, List.of(input(DATASET_NAME)), List.of());
+ }
+
+ private static RunEvent event(
+ String jobNamespace, List<InputDataset> inputs, List<OutputDataset>
outputs) {
+ return new RunEvent(
+ ZonedDateTime.now(ZoneOffset.UTC),
+ PRODUCER,
+ SCHEMA_URL,
+ EventType.START,
+ new Run(UUID.randomUUID(), null),
+ new Job(jobNamespace, "job", null),
+ inputs,
+ outputs);
+ }
+}
