Implement OpaPolarisAuthorizer

bom/build.gradle.kts

@@ -52,6 +52,8 @@ dependencies {
     api(project(":polaris-eclipselink"))
     api(project(":polaris-relational-jdbc"))
 
+    api(project(":polaris-extensions-auth-opa"))
+
     api(project(":polaris-admin"))
     api(project(":polaris-runtime-common"))
     api(project(":polaris-runtime-test-common"))

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

@@ -0,0 +1,60 @@
+/*
+ * 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.
+ */
+
+plugins {
+  id("polaris-server")
+  id("org.kordamp.gradle.jandex")
+}
+
+dependencies {
+  implementation(project(":polaris-core"))
+  implementation(libs.apache.httpclient5)
+  implementation(platform(libs.jackson.bom))
+  implementation("com.fasterxml.jackson.core:jackson-core")
+  implementation("com.fasterxml.jackson.core:jackson-databind")
+  implementation(libs.guava)
+  implementation(libs.slf4j.api)
+  implementation(libs.auth0.jwt)
+  implementation(project(":polaris-async-api"))
+
+  // Iceberg dependency for ForbiddenException
+  implementation(platform(libs.iceberg.bom))
+  implementation("org.apache.iceberg:iceberg-api")
+
+  compileOnly(project(":polaris-immutables"))
+  annotationProcessor(project(":polaris-immutables", configuration = "processor"))
+
+  compileOnly(libs.jakarta.annotation.api)
+  compileOnly(libs.jakarta.enterprise.cdi.api)
+  compileOnly(libs.jakarta.inject.api)
+  compileOnly(libs.smallrye.config.core)
+
+  testCompileOnly(project(":polaris-immutables"))
+  testAnnotationProcessor(project(":polaris-immutables", configuration = "processor"))
+
+  testImplementation(testFixtures(project(":polaris-core")))
+  testImplementation(platform(libs.junit.bom))
+  testImplementation("org.junit.jupiter:junit-jupiter")
+  testImplementation(libs.assertj.core)
+  testImplementation(libs.mockito.core)
+  testImplementation(libs.threeten.extra)
+  testImplementation(testFixtures(project(":polaris-async-api")))
+  testImplementation(project(":polaris-async-java"))
+  testImplementation(project(":polaris-idgen-mocks"))
+}

extensions/auth/opa/impl/src/main/java/org/apache/polaris/extension/auth/opa/OpaAuthorizationConfig.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.polaris.extension.auth.opa;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+import com.google.common.base.Strings;
+import io.smallrye.config.ConfigMapping;
+import io.smallrye.config.WithDefault;
+import java.net.URI;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.Optional;
+import org.apache.polaris.immutables.PolarisImmutable;
+
+/**
+ * Configuration for OPA (Open Policy Agent) authorization.
+ *
+ * <p><strong>Beta Feature:</strong> OPA authorization is currently in Beta and is not a stable
+ * release. It may undergo breaking changes in future versions. Use with caution in production
+ * environments.
+ */
+@PolarisImmutable
+@ConfigMapping(prefix = "polaris.authorization.opa")
+public interface OpaAuthorizationConfig {
+
+  /** Authentication types supported by OPA authorization */
+  enum AuthenticationType {
+    NONE("none"),
+    BEARER("bearer");
+
+    private final String value;
+
+    AuthenticationType(String value) {
+      this.value = value;
+    }
+
+    public String getValue() {
+      return value;
+    }
+  }
+
+  Optional<URI> policyUri();
+
+  AuthenticationConfig auth();
+
+  HttpConfig http();
+
+  /** Validates the complete OPA configuration */
+  default void validate() {
+    checkArgument(
+        policyUri().isPresent(), "polaris.authorization.opa.policy-uri must be configured");
+
+    URI uri = policyUri().get();
+    String scheme = uri.getScheme();
+    checkArgument(
+        "http".equalsIgnoreCase(scheme) || "https".equalsIgnoreCase(scheme),
+        "polaris.authorization.opa.policy-uri must use http or https scheme, but got: " + scheme);
+
+    auth().validate();
+  }
+
+  /** HTTP client configuration for OPA communication. */
+  @PolarisImmutable
+  interface HttpConfig {
+    @WithDefault("PT2S")
+    Duration timeout();
+
+    @WithDefault("true")
+    boolean verifySsl();
+
+    Optional<Path> trustStorePath();
+
+    Optional<String> trustStorePassword();
+  }
+
+  /** Authentication configuration for OPA communication. */
+  @PolarisImmutable
+  interface AuthenticationConfig {
+    /** Type of authentication */
+    @WithDefault("none")
+    AuthenticationType type();
+
+    /** Bearer token authentication configuration */
+    Optional<BearerTokenConfig> bearer();
+
+    default void validate() {
+      switch (type()) {
+        case BEARER:
+          checkArgument(
+              bearer().isPresent(), "Bearer configuration is required when type is 'bearer'");
+          bearer().get().validate();
+          break;
+        case NONE:
+          // No authentication - nothing to validate
+          break;
+        default:
+          throw new IllegalArgumentException(
+              "Invalid authentication type: " + type() + ". Supported types: 'bearer', 'none'");
+      }
+    }
+  }
+
+  @PolarisImmutable
+  interface BearerTokenConfig {
+    /** Static bearer token configuration */
+    Optional<StaticTokenConfig> staticToken();
+
+    /** File-based bearer token configuration */
+    Optional<FileBasedConfig> fileBased();
+
+    default void validate() {
+      // Ensure exactly one bearer token configuration is present (mutually exclusive)
+      checkArgument(
+          staticToken().isPresent() ^ fileBased().isPresent(),
+          "Exactly one of 'static-token' or 'file-based' bearer token configuration must be specified");
+
+      // Validate the present configuration
+      if (staticToken().isPresent()) {
+        staticToken().get().validate();
+      } else {
+        fileBased().get().validate();
+      }
+    }
+
+    /** Configuration for static bearer tokens */
+    @PolarisImmutable
+    interface StaticTokenConfig {
+      /** Static bearer token value */
+      String value();
+
+      default void validate() {
+        checkArgument(
+            !Strings.isNullOrEmpty(value()), "Static bearer token value cannot be null or empty");
+      }
+    }
+
+    /** Configuration for file-based bearer tokens */
+    @PolarisImmutable
+    interface FileBasedConfig {
+      /** Path to file containing bearer token */
+      Path path();
+
+      /** How often to refresh file-based bearer tokens (defaults to 5 minutes if not specified) */
+      Optional<Duration> refreshInterval();
+
+      /**
+       * Whether to automatically detect JWT tokens and use their 'exp' field for refresh timing. If
+       * true and the token is a valid JWT with an 'exp' claim, the token will be refreshed based on
+       * the expiration time minus the buffer, rather than the fixed refresh interval. Defaults to
+       * true if not specified.
+       */
+      Optional<Boolean> jwtExpirationRefresh();
+
+      /**
+       * Buffer time before JWT expiration to refresh the token. Only used when jwtExpirationRefresh
+       * is true and the token is a valid JWT. Defaults to 1 minute if not specified.
+       */
+      Optional<Duration> jwtExpirationBuffer();
+
+      default void validate() {
+        checkArgument(
+            refreshInterval().isEmpty() || refreshInterval().get().isPositive(),
+            "refreshInterval must be positive");
+        checkArgument(
+            jwtExpirationBuffer().isEmpty() || jwtExpirationBuffer().get().isPositive(),
+            "jwtExpirationBuffer must be positive");
+      }
+    }
+  }
+}

extensions/auth/opa/impl/src/main/java/org/apache/polaris/extension/auth/opa/OpaHttpClientFactory.java

@@ -0,0 +1,130 @@
+/*
+ * 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;
+
+import java.io.FileInputStream;
+import java.nio.file.Path;
+import java.security.KeyStore;
+import java.security.cert.X509Certificate;
+import javax.net.ssl.SSLContext;
+import org.apache.hc.client5.http.config.RequestConfig;
+import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
+import org.apache.hc.client5.http.impl.classic.HttpClients;
+import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManagerBuilder;
+import org.apache.hc.client5.http.ssl.DefaultClientTlsStrategy;
+import org.apache.hc.client5.http.ssl.NoopHostnameVerifier;
+import org.apache.hc.core5.ssl.SSLContexts;
+import org.apache.hc.core5.util.Timeout;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Factory for creating HTTP clients configured for OPA communication with SSL support.
+ *
+ * <p>This factory handles the creation of Apache HttpClient instances with proper SSL
+ * configuration, timeout settings, and connection pooling for communicating with Open Policy Agent
+ * (OPA) servers.
+ */
+class OpaHttpClientFactory {
+  private static final Logger logger = LoggerFactory.getLogger(OpaHttpClientFactory.class);
+
+  /**
+   * Creates a configured HTTP client for OPA communication.
+   *
+   * @param config HTTP configuration for timeouts and SSL settings
+   * @return configured CloseableHttpClient
+   */
+  public static CloseableHttpClient createHttpClient(OpaAuthorizationConfig.HttpConfig config) {
+    RequestConfig requestConfig =
+        RequestConfig.custom()
+            .setResponseTimeout(Timeout.ofMilliseconds(config.timeout().toMillis()))
+            .build();
+
+    try {
+      // Create TLS strategy based on configuration
+      DefaultClientTlsStrategy tlsStrategy = createTlsStrategy(config);
+
+      // Create connection manager with the TLS strategy
+      var connectionManager =
+          PoolingHttpClientConnectionManagerBuilder.create()
+              .setTlsSocketStrategy(tlsStrategy)
+              .build();
+
+      return HttpClients.custom()
+          .setConnectionManager(connectionManager)
+          .setDefaultRequestConfig(requestConfig)
+          .build();
+    } catch (Exception e) {
+      throw new RuntimeException("Failed to create HTTP client for OPA communication", e);
+    }
+  }
+
+  /**
+   * Creates a TLS strategy based on the configuration.
+   *
+   * @param config HTTP configuration containing SSL settings
+   * @return DefaultClientTlsStrategy for HTTPS connections
+   */
+  private static DefaultClientTlsStrategy createTlsStrategy(
+      OpaAuthorizationConfig.HttpConfig config) throws Exception {
+    SSLContext sslContext = createSslContext(config);
+
+    if (!config.verifySsl()) {
+      // Disable hostname verification when SSL verification is disabled
+      return new DefaultClientTlsStrategy(sslContext, NoopHostnameVerifier.INSTANCE);
+    } else {
+      // Use default hostname verification when SSL verification is enabled
+      return new DefaultClientTlsStrategy(sslContext);
+    }
+  }
+
+  /**
+   * Creates an SSL context based on the configuration.
+   *
+   * @param config HTTP configuration containing SSL settings
+   * @return SSLContext for HTTPS connections
+   */
+  private static SSLContext createSslContext(OpaAuthorizationConfig.HttpConfig config)
+      throws Exception {
+    if (!config.verifySsl()) {
+      // Disable SSL verification (for development/testing)
+      logger.warn(
+          "SSL verification is disabled for OPA server. This should only be used in development/testing environments.");
+      return SSLContexts.custom()
+          .loadTrustMaterial(
+              null, (X509Certificate[] chain, String authType) -> true) // trust all certificates
+          .build();
+    } else if (config.trustStorePath().isPresent()) {
+      // Load custom trust store for SSL verification
+      Path trustStorePath = config.trustStorePath().get();
+      logger.info("Loading custom trust store for OPA SSL verification: {}", trustStorePath);
+      KeyStore trustStore = KeyStore.getInstance(KeyStore.getDefaultType());
+      try (FileInputStream trustStoreStream = new FileInputStream(trustStorePath.toFile())) {
+        String trustStorePassword = config.trustStorePassword().orElse(null);
+        trustStore.load(
+            trustStoreStream, trustStorePassword != null ? trustStorePassword.toCharArray() : null);
+      }
+      return SSLContexts.custom().loadTrustMaterial(trustStore, null).build();
+    } else {
+      // Use default system trust store for SSL verification
+      logger.debug("Using default system trust store for OPA SSL verification");
+      return SSLContexts.createDefault();
+    }
+  }
+}