Auth Manager API part 2: AuthManager (#11809)

* Auth Manager API part 2: AuthManager, AuthSession

* [review] close()

* [review] nits

* HTTPAuthSession close()

* mainSession -> catalogSession

* AuthManager constructor

core/src/main/java/org/apache/iceberg/rest/HTTPHeaders.java

@@ -19,6 +19,7 @@
 package org.apache.iceberg.rest;
 
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
@@ -85,6 +86,15 @@ static HTTPHeaders of(HTTPHeader... headers) {
     return ImmutableHTTPHeaders.builder().addEntries(headers).build();
   }
 
+  static HTTPHeaders of(Map<String, String> headers) {
+    return ImmutableHTTPHeaders.builder()
+        .entries(
+            headers.entrySet().stream()
+                .map(e -> HTTPHeader.of(e.getKey(), e.getValue()))
+                .collect(Collectors.toList()))
+        .build();
+  }
+
   /** Represents an HTTP header as a name-value pair. */
   @Value.Style(redactedMask = "****", depluralize = true)
   @Value.Immutable

core/src/main/java/org/apache/iceberg/rest/auth/AuthManager.java

@@ -0,0 +1,106 @@
+/*
+ * 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.iceberg.rest.auth;
+
+import java.util.Map;
+import org.apache.iceberg.catalog.SessionCatalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.RESTClient;
+
+/**
+ * Manager for authentication sessions. This interface is used to create sessions for the catalog,
+ * the tables/views, and any other context that requires authentication.
+ *
+ * <p>Managers are usually stateful and may require initialization and cleanup. The manager is
+ * created by the catalog and is closed when the catalog is closed.
+ */
+public interface AuthManager extends AutoCloseable {
+
+  /**
+   * Returns a temporary session to use for contacting the configuration endpoint only. Note that
+   * the returned session will be closed after the configuration endpoint is contacted, and should
+   * not be cached.
+   *
+   * <p>The provided REST client is a short-lived client; it should only be used to fetch initial
+   * credentials, if required, and must be discarded after that.
+   *
+   * <p>This method cannot return null. By default, it returns the catalog session.
+   */
+  default AuthSession initSession(RESTClient initClient, Map<String, String> properties) {
+    return catalogSession(initClient, properties);
+  }
+
+  /**
+   * Returns a long-lived session whose lifetime is tied to the owning catalog. This session serves
+   * as the parent session for all other sessions (contextual and table-specific). It is closed when
+   * the owning catalog is closed.
+   *
+   * <p>The provided REST client is a long-lived, shared client; if required, implementors may store
+   * it and reuse it for all subsequent requests to the authorization server, e.g. for renewing or
+   * refreshing credentials. It is not necessary to close it when {@link #close()} is called.
+   *
+   * <p>This method cannot return null.
+   *
+   * <p>It is not required to cache the returned session internally, as the catalog will keep it
+   * alive for the lifetime of the catalog.
+   */
+  AuthSession catalogSession(RESTClient sharedClient, Map<String, String> properties);
+
+  /**
+   * Returns a session for a specific context.
+   *
+   * <p>If the context requires a specific {@link AuthSession}, this method should return a new
+   * {@link AuthSession} instance, otherwise it should return the parent session.
+   *
+   * <p>This method cannot return null. By default, it returns the parent session.
+   *
+   * <p>Implementors should cache contextual sessions internally, as the catalog will not cache
+   * them. Also, the owning catalog never closes contextual sessions; implementations should manage
+   * their lifecycle themselves and close them when they are no longer needed.
+   */
+  default AuthSession contextualSession(SessionCatalog.SessionContext context, AuthSession parent) {
+    return parent;
+  }
+
+  /**
+   * Returns a new session targeting a specific table or view. The properties are the ones returned
+   * by the table/view endpoint.
+   *
+   * <p>If the table or view requires a specific {@link AuthSession}, this method should return a
+   * new {@link AuthSession} instance, otherwise it should return the parent session.
+   *
+   * <p>This method cannot return null. By default, it returns the parent session.
+   *
+   * <p>Implementors should cache table sessions internally, as the catalog will not cache them.
+   * Also, the owning catalog never closes table sessions; implementations should manage their
+   * lifecycle themselves and close them when they are no longer needed.
+   */
+  default AuthSession tableSession(
+      TableIdentifier table, Map<String, String> properties, AuthSession parent) {
+    return parent;
+  }
+
+  /**
+   * Closes the manager and releases any resources.
+   *
+   * <p>This method is called when the owning catalog is closed.
+   */
+  @Override
+  void close();
+}

core/src/main/java/org/apache/iceberg/rest/auth/AuthManagers.java

@@ -0,0 +1,75 @@
+/*
+ * 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.iceberg.rest.auth;
+
+import java.util.Locale;
+import java.util.Map;
+import org.apache.iceberg.common.DynConstructors;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class AuthManagers {
+
+  private static final Logger LOG = LoggerFactory.getLogger(AuthManagers.class);
+
+  private AuthManagers() {}
+
+  public static AuthManager loadAuthManager(String name, Map<String, String> properties) {
+    String authType =
+        properties.getOrDefault(AuthProperties.AUTH_TYPE, AuthProperties.AUTH_TYPE_NONE);
+
+    String impl;
+    switch (authType.toLowerCase(Locale.ROOT)) {
+      case AuthProperties.AUTH_TYPE_NONE:
+        impl = AuthProperties.AUTH_MANAGER_IMPL_NONE;
+        break;
+      case AuthProperties.AUTH_TYPE_BASIC:
+        impl = AuthProperties.AUTH_MANAGER_IMPL_BASIC;
+        break;
+      default:
+        impl = authType;
+    }
+
+    LOG.info("Loading AuthManager implementation: {}", impl);
+    DynConstructors.Ctor<AuthManager> ctor;
+    try {
+      ctor =
+          DynConstructors.builder(AuthManager.class)
+              .loader(AuthManagers.class.getClassLoader())
+              .impl(impl, String.class) // with name
+              .buildChecked();
+    } catch (NoSuchMethodException e) {
+      throw new IllegalArgumentException(
+          String.format(
+              "Cannot initialize AuthManager implementation %s: %s", impl, e.getMessage()),
+          e);
+    }
+
+    AuthManager authManager;
+    try {
+      authManager = ctor.newInstance(name);
+    } catch (ClassCastException e) {
+      throw new IllegalArgumentException(
+          String.format("Cannot initialize AuthManager, %s does not implement AuthManager", impl),
+          e);
+    }
+
+    return authManager;
+  }
+}

core/src/main/java/org/apache/iceberg/rest/auth/AuthProperties.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.rest.auth;
+
+public final class AuthProperties {
+
+  private AuthProperties() {}
+
+  public static final String AUTH_TYPE = "rest.auth.type";
+
+  public static final String AUTH_TYPE_NONE = "none";
+  public static final String AUTH_TYPE_BASIC = "basic";
+
+  public static final String AUTH_MANAGER_IMPL_NONE =
+      "org.apache.iceberg.rest.auth.NoopAuthManager";
+  public static final String AUTH_MANAGER_IMPL_BASIC =
+      "org.apache.iceberg.rest.auth.BasicAuthManager";
+
+  public static final String BASIC_USERNAME = "rest.auth.basic.username";
+  public static final String BASIC_PASSWORD = "rest.auth.basic.password";
+}

core/src/main/java/org/apache/iceberg/rest/auth/AuthSession.java

@@ -0,0 +1,57 @@
+/*
+ * 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.iceberg.rest.auth;
+
+import org.apache.iceberg.rest.HTTPRequest;
+
+/**
+ * An authentication session that can be used to authenticate outgoing HTTP requests.
+ *
+ * <p>Authentication sessions are usually immutable, but may hold resources that need to be released
+ * when the session is no longer needed. Implementations should override {@link #close()} to release
+ * any resources.
+ */
+public interface AuthSession extends AutoCloseable {
+
+  /** An empty session that does nothing. */
+  AuthSession EMPTY =
+      new AuthSession() {
+        @Override
+        public HTTPRequest authenticate(HTTPRequest request) {
+          return request;
+        }
+
+        @Override
+        public void close() {}
+      };
+
+  /**
+   * Authenticates the given request and returns a new request with the necessary authentication.
+   */
+  HTTPRequest authenticate(HTTPRequest request);
+
+  /**
+   * Closes the session and releases any resources. This method is called when the session is no
+   * longer needed. Note that since sessions may be cached, this method may not be called
+   * immediately after the session is no longer needed, but rather when the session is evicted from
+   * the cache, or the cache itself is closed.
+   */
+  @Override
+  void close();
+}

core/src/main/java/org/apache/iceberg/rest/auth/BasicAuthManager.java

@@ -0,0 +1,53 @@
+/*
+ * 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.iceberg.rest.auth;
+
+import java.util.Map;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.rest.HTTPHeaders;
+import org.apache.iceberg.rest.RESTClient;
+
+/** An auth manager that adds static BASIC authentication data to outgoing HTTP requests. */
+public final class BasicAuthManager implements AuthManager {
+
+  public BasicAuthManager(String ignored) {
+    // no-op
+  }
+
+  @Override
+  public AuthSession catalogSession(RESTClient sharedClient, Map<String, String> properties) {
+    Preconditions.checkArgument(
+        properties.containsKey(AuthProperties.BASIC_USERNAME),
+        "Invalid username: missing required property %s",
+        AuthProperties.BASIC_USERNAME);
+    Preconditions.checkArgument(
+        properties.containsKey(AuthProperties.BASIC_PASSWORD),
+        "Invalid password: missing required property %s",
+        AuthProperties.BASIC_PASSWORD);
+    String username = properties.get(AuthProperties.BASIC_USERNAME);
+    String password = properties.get(AuthProperties.BASIC_PASSWORD);
+    String credentials = username + ":" + password;
+    return DefaultAuthSession.of(HTTPHeaders.of(OAuth2Util.basicAuthHeaders(credentials)));
+  }
+
+  @Override
+  public void close() {
+    // no resources to close
+  }
+}