add IT for auth fail

jerry-024 · Jan 7, 2025 · f5037c4 · f5037c4
1 parent d00623a
commit f5037c4
Show file tree

Hide file tree

Showing 3 changed files with 32 additions and 10 deletions.
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/auth/AuthSession.java b/paimon-core/src/main/java/org/apache/paimon/rest/auth/AuthSession.java
@@ -41,8 +41,8 @@ public class AuthSession {
     private volatile Map<String, String> headers;
 
     public AuthSession(Map<String, String> headers, CredentialsProvider credentialsProvider) {
-        this.headers = headers;
         this.credentialsProvider = credentialsProvider;
+        this.headers = RESTUtil.merge(headers, this.credentialsProvider.authHeader());
     }
 
     public static AuthSession fromRefreshCredentialsProvider(

diff --git a/...flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkRESTCatalogTest.java b/...flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkRESTCatalogTest.java
@@ -23,8 +23,7 @@
 import org.apache.paimon.options.Options;
 import org.apache.paimon.rest.RESTCatalogFactory;
 import org.apache.paimon.rest.RESTCatalogOptions;
-
-import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.paimon.rest.exceptions.NotAuthorizedException;
 
 import org.apache.flink.table.catalog.Catalog;
 import org.apache.flink.table.catalog.CatalogDatabase;
@@ -60,12 +59,12 @@ public class FlinkRESTCatalogTest {
     @Before
     public void beforeEach() throws IOException {
         warehouse = new File(temporaryFolder.newFolder(), UUID.randomUUID().toString()).toString();
-        mockRESTCatalogServer = new MockRESTCatalogServer(warehouse);
+        String initToken = "init_token";
+        mockRESTCatalogServer = new MockRESTCatalogServer(warehouse, initToken);
         mockRESTCatalogServer.start();
         serverUrl = mockRESTCatalogServer.getUrl();
         Options options = new Options();
         options.set(RESTCatalogOptions.URI, serverUrl);
-        String initToken = "init_token";
         options.set(RESTCatalogOptions.TOKEN, initToken);
         options.set(RESTCatalogOptions.THREAD_POOL_SIZE, 1);
         options.set(LOG_SYSTEM_AUTO_REGISTER, true);
@@ -83,7 +82,24 @@ public void tearDown() throws IOException {
     }
 
     @Test
-    public void testListDatabases() throws JsonProcessingException {
+    public void testAuthFail() {
+        Options options = new Options();
+        options.set(RESTCatalogOptions.URI, serverUrl);
+        options.set(RESTCatalogOptions.TOKEN, "aaaaa");
+        options.set(RESTCatalogOptions.THREAD_POOL_SIZE, 1);
+        options.set(LOG_SYSTEM_AUTO_REGISTER, true);
+        options.set(CatalogOptions.METASTORE, RESTCatalogFactory.IDENTIFIER);
+        assertThatThrownBy(
+                        () ->
+                                FlinkCatalogFactory.createCatalog(
+                                        "test-catalog",
+                                        CatalogContext.create(options),
+                                        FlinkCatalogTest.class.getClassLoader()))
+                .isInstanceOf(NotAuthorizedException.class);
+    }
+
+    @Test
+    public void testListDatabases() {
         List<String> result = catalog.listDatabases();
         assertEquals(1, result.size());
     }

diff --git a/...link/paimon-flink-common/src/test/java/org/apache/paimon/flink/MockRESTCatalogServer.java b/...link/paimon-flink-common/src/test/java/org/apache/paimon/flink/MockRESTCatalogServer.java
@@ -53,15 +53,17 @@ public class MockRESTCatalogServer {
     private final Catalog catalog;
     private final Dispatcher dispatcher;
     private final MockWebServer server;
+    private final String authToken;
 
-    public MockRESTCatalogServer(String warehouse) {
+    public MockRESTCatalogServer(String warehouse, String initToken) {
+        authToken = initToken;
         Options conf = new Options();
         conf.setString("warehouse", warehouse);
         conf.set(LOG_SYSTEM_AUTO_REGISTER, true);
         this.catalog =
                 CatalogFactory.createCatalog(
                         CatalogContext.create(conf), this.getClass().getClassLoader());
-        this.dispatcher = initDispatcher(catalog);
+        this.dispatcher = initDispatcher(catalog, authToken);
         try {
             catalog.createDatabase("default", true);
         } catch (Exception e) {
@@ -84,14 +86,18 @@ public void shutdown() throws IOException {
         server.shutdown();
     }
 
-    public static Dispatcher initDispatcher(Catalog catalog) {
+    public static Dispatcher initDispatcher(Catalog catalog, String authToken) {
         return new Dispatcher() {
             @Override
             public MockResponse dispatch(RecordedRequest request) throws InterruptedException {
 
-                RESTResponse response;
                 System.out.println(request.getPath() + "  method " + request.getMethod());
+                String token = request.getHeaders().get("Authorization");
+                RESTResponse response;
                 try {
+                    if (!("Bearer " + authToken).equals(token)) {
+                        return new MockResponse().setResponseCode(401);
+                    }
                     if ("/v1/config".equals(request.getPath())) {
                         return new MockResponse()
                                 .setResponseCode(200)