Introduce ExternalAuthenticator and isExternal flag on PolarisCredential

extensions/auth/opa/impl/src/test/java/org/apache/polaris/extension/auth/opa/OpaPolarisAuthorizerTest.java

@@ -36,6 +36,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
 import org.apache.hc.client5.http.classic.methods.HttpPost;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
 import org.apache.hc.client5.http.impl.classic.HttpClients;
@@ -108,6 +110,50 @@ void testOpaInputJsonFormat() throws Exception {
     }
   }
 
+  @Test
+  void testExternalPrincipalRolesAndNameAreSentToOpa() throws Exception {
+    final String[] capturedRequestBody = new String[1];
+
+    HttpServer server = createServerWithRequestCapture(capturedRequestBody);
+    try {
+      URI policyUri =
+          URI.create(
+              "http://localhost:" + server.getAddress().getPort() + "/v1/data/polaris/allow");
+      OpaPolarisAuthorizer authorizer =
+          new OpaPolarisAuthorizer(
+              policyUri, HttpClients.createDefault(), new ObjectMapper(), null);
+
+      PolarisPrincipal principal =
+          PolarisPrincipal.of(
+              "external-user",
+              Map.of("external", "true", "principalId", "42"),
+              Set.of("ext-role", "common-role"));
+
+      PolarisResolvedPathWrapper target = new PolarisResolvedPathWrapper(List.of());
+
+      assertThatNoException()
+          .isThrownBy(
+              () ->
+                  authorizer.authorizeOrThrow(
+                      principal,
+                      Set.of(),
+                      PolarisAuthorizableOperation.LIST_NAMESPACES,
+                      target,
+                      null));
+
+      JsonNode actorNode =
+          new ObjectMapper().readTree(capturedRequestBody[0]).path("input").path("actor");
+      assertThat(actorNode.get("principal").asText()).isEqualTo("external-user");
+      assertThat(
+              StreamSupport.stream(actorNode.get("roles").spliterator(), false)
+                  .map(JsonNode::asText)
+                  .collect(Collectors.toSet()))
+          .containsExactlyInAnyOrder("ext-role", "common-role");
+    } finally {
+      server.stop(0);
+    }
+  }
+
   @Test
   void testOpaRequestJsonWithHierarchicalResource() throws Exception {
     // Capture the request body for verification

extensions/auth/opa/tests/build.gradle.kts

@@ -38,8 +38,12 @@ dependencies {
   testImplementation(project(":polaris-runtime-test-common"))
 
   // Test dependencies
+  intTestImplementation(platform(libs.iceberg.bom))
+  intTestImplementation("org.apache.iceberg:iceberg-api")
+  intTestImplementation(project(":polaris-core"))
   intTestImplementation("io.quarkus:quarkus-junit5")
   intTestImplementation("io.rest-assured:rest-assured")
+  intTestImplementation("io.quarkus:quarkus-security")
 
   // Test container dependencies
   intTestImplementation(platform(libs.testcontainers.bom))

extensions/auth/opa/tests/src/intTest/java/org/apache/polaris/extension/auth/opa/test/OpaExternalAuthIntegrationTest.java

@@ -0,0 +1,91 @@
+/*
+ * 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.polaris.extension.auth.opa.test;
+
+import static io.restassured.RestAssured.given;
+
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.QuarkusTestProfile;
+import io.quarkus.test.junit.QuarkusTestProfile.TestResourceEntry;
+import io.quarkus.test.junit.TestProfile;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Integration test that exercises OPA authorization when Polaris uses the external authenticator.
+ *
+ * <p>Authentication is driven entirely by test headers via {@link
+ * TestExternalHeaderAuthenticationMechanism}, ensuring no internal principal lookups occur.
+ */
+@QuarkusTest
+@TestProfile(OpaExternalAuthIntegrationTest.ExternalOpaProfile.class)
+public class OpaExternalAuthIntegrationTest extends OpaIntegrationTestBase {
+
+  public static class ExternalOpaProfile implements QuarkusTestProfile {
+    @Override
+    public Map<String, String> getConfigOverrides() {
+      Map<String, String> config = new HashMap<>();
+      config.put("polaris.authorization.type", "opa");
+      config.put("polaris.authorization.opa.auth.type", "bearer");
+      config.put(
+          "polaris.authorization.opa.auth.bearer.static-token.value",
+          "test-opa-bearer-token-12345");
+
+      // Enable external authentication and skip internal token services
+      config.put("polaris.authentication.type", "external");
+      config.put("polaris.authentication.authenticator.type", "external");
+      config.put("polaris.authentication.token-broker.type", "none");
+      config.put("polaris.authentication.token-service.type", "disabled");
+
+      // OIDC not used in this flow
+      config.put("quarkus.oidc.enabled", "false");
+
+      return config;
+    }
+
+    @Override
+    public List<TestResourceEntry> testResources() {
+      return List.of(new TestResourceEntry(OpaTestResource.class));
+    }
+  }
+
+  @Test
+  void testExternalPrincipalAllowed() {
+    given()
+        .header(TestExternalHeaderAuthenticationMechanism.PRINCIPAL_HEADER, "admin")
+        .header(TestExternalHeaderAuthenticationMechanism.ROLES_HEADER, "ext-role")
+        .when()
+        .get("/api/management/v1/catalogs")
+        .then()
+        .statusCode(200);
+  }
+
+  @Test
+  void testExternalPrincipalDenied() {
+    given()
+        .header(TestExternalHeaderAuthenticationMechanism.PRINCIPAL_HEADER, "stranger")
+        .header(TestExternalHeaderAuthenticationMechanism.ROLES_HEADER, "unknown-role")
+        .when()
+        .get("/api/management/v1/catalogs")
+        .then()
+        .statusCode(403);
+  }
+}

extensions/auth/opa/tests/src/intTest/java/org/apache/polaris/extension/auth/opa/test/TestExternalHeaderAuthenticationMechanism.java

@@ -0,0 +1,105 @@
+/*
+ * 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.polaris.extension.auth.opa.test;
+
+import io.quarkus.security.identity.IdentityProviderManager;
+import io.quarkus.security.identity.SecurityIdentity;
+import io.quarkus.security.runtime.QuarkusSecurityIdentity;
+import io.quarkus.vertx.http.runtime.security.ChallengeData;
+import io.quarkus.vertx.http.runtime.security.HttpAuthenticationMechanism;
+import io.quarkus.vertx.http.runtime.security.HttpCredentialTransport;
+import io.smallrye.mutiny.Uni;
+import io.vertx.ext.web.RoutingContext;
+import jakarta.annotation.Priority;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.polaris.core.auth.PolarisPrincipal;
+import org.apache.polaris.service.auth.Authenticator;
+import org.apache.polaris.service.auth.PolarisCredential;
+import org.jboss.logging.Logger;
+
+/**
+ * Test-only authentication mechanism that turns test headers into external {@link
+ * PolarisCredential} instances. This avoids any DB lookups and exercises the external authenticator
+ * flow.
+ */
+@ApplicationScoped
+@Priority(HttpAuthenticationMechanism.DEFAULT_PRIORITY + 200)
+public class TestExternalHeaderAuthenticationMechanism implements HttpAuthenticationMechanism {
+
+  private static final Logger LOG =
+      Logger.getLogger(TestExternalHeaderAuthenticationMechanism.class);
+
+  static final String PRINCIPAL_HEADER = "X-External-Principal";
+  static final String ROLES_HEADER = "X-External-Roles";
+
+  @Inject Authenticator authenticator;
+
+  @Override
+  public Uni<SecurityIdentity> authenticate(
+      RoutingContext context, IdentityProviderManager identityProviderManager) {
+    String principal = context.request().getHeader(PRINCIPAL_HEADER);
+    if (principal == null || principal.isBlank()) {
+      return Uni.createFrom().nullItem();
+    }
+    Set<String> roles = parseRoles(context.request().getHeader(ROLES_HEADER));
+    PolarisCredential credential = PolarisCredential.of(null, principal, roles, true);
+    PolarisPrincipal polarisPrincipal = authenticator.authenticate(credential);
+    QuarkusSecurityIdentity identity =
+        QuarkusSecurityIdentity.builder()
+            .setPrincipal(polarisPrincipal)
+            .addCredential(credential)
+            .addRoles(polarisPrincipal.getRoles())
+            .setAnonymous(false)
+            .build();
+    LOG.debugf(
+        "Authenticated external principal from headers principal=%s roles=%s", principal, roles);
+    return Uni.createFrom().item(identity);
+  }
+
+  @Override
+  public Uni<ChallengeData> getChallenge(RoutingContext context) {
+    return Uni.createFrom().nullItem();
+  }
+
+  @Override
+  public Set<Class<? extends io.quarkus.security.identity.request.AuthenticationRequest>>
+      getCredentialTypes() {
+    return Collections.emptySet();
+  }
+
+  @Override
+  public Uni<HttpCredentialTransport> getCredentialTransport(RoutingContext context) {
+    return Uni.createFrom().nullItem();
+  }
+
+  private Set<String> parseRoles(String header) {
+    if (header == null || header.isBlank()) {
+      return Set.of();
+    }
+    return Arrays.stream(header.split(","))
+        .map(String::trim)
+        .filter(s -> !s.isEmpty())
+        .collect(Collectors.toSet());
+  }
+}

polaris-core/src/main/java/org/apache/polaris/core/persistence/resolver/Resolver.java

@@ -41,6 +41,7 @@
 import org.apache.polaris.core.entity.PolarisEntityType;
 import org.apache.polaris.core.entity.PolarisGrantRecord;
 import org.apache.polaris.core.entity.PolarisPrivilege;
+import org.apache.polaris.core.entity.PrincipalEntity;
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 import org.apache.polaris.core.persistence.ResolvedPolarisEntity;
 import org.apache.polaris.core.persistence.cache.EntityCache;
@@ -744,13 +745,42 @@ private ResolverStatus resolvePaths(
   private ResolverStatus resolveCallerPrincipalAndPrincipalRoles(
       List<ResolvedPolarisEntity> toValidate) {
 
+    if (isExternalPrincipal()) {
+      PrincipalEntity externalPrincipal = createExternalPrincipalEntity();
+      this.resolvedCallerPrincipal =
+          new ResolvedPolarisEntity(
+              diagnostics,
+              externalPrincipal,
+              List.of(),
+              externalPrincipal.getGrantRecordsVersion());
+      this.resolvedEntriesById.put(
+          this.resolvedCallerPrincipal.getEntity().getId(), this.resolvedCallerPrincipal);
+      this.resolvedCallerPrincipalRoles = List.of();
+      return new ResolverStatus(ResolverStatus.StatusEnum.SUCCESS);
+    }
+
     // resolve the principal, by name or id
     this.resolvedCallerPrincipal =
         this.resolveByName(toValidate, PolarisEntityType.PRINCIPAL, polarisPrincipal.getName());
 
     // if the principal was not found, we can end right there
     if (this.resolvedCallerPrincipal == null
         || this.resolvedCallerPrincipal.getEntity().isDropped()) {
+      if (isExternalPrincipal()) {
+        // For external principals we do not maintain principal entities in the metastore,
+        // so synthesize a placeholder entry and continue without resolving grants.
+        PrincipalEntity externalPrincipal = createExternalPrincipalEntity();
+        this.resolvedCallerPrincipal =
+            new ResolvedPolarisEntity(
+                diagnostics,
+                externalPrincipal,
+                List.of(),
+                externalPrincipal.getGrantRecordsVersion());
+        this.resolvedEntriesById.put(
+            this.resolvedCallerPrincipal.getEntity().getId(), this.resolvedCallerPrincipal);
+        this.resolvedCallerPrincipalRoles = List.of();
+        return new ResolverStatus(ResolverStatus.StatusEnum.SUCCESS);
+      }
       return new ResolverStatus(ResolverStatus.StatusEnum.CALLER_PRINCIPAL_DOES_NOT_EXIST);
     }
 
@@ -764,6 +794,17 @@ private ResolverStatus resolveCallerPrincipalAndPrincipalRoles(
     return new ResolverStatus(ResolverStatus.StatusEnum.SUCCESS);
   }
 
+  private boolean isExternalPrincipal() {
+    return Boolean.parseBoolean(polarisPrincipal.getProperties().getOrDefault("external", "false"));
+  }
+
+  private PrincipalEntity createExternalPrincipalEntity() {
+    return new PrincipalEntity.Builder()
+        .setName(polarisPrincipal.getName())
+        .setProperties(polarisPrincipal.getProperties())
+        .build();
+  }
+
   /**
    * Resolve all principal roles that the principal has grants for
    *