(incubator-streampark) branch dev-2.1.5 updated: fix(streampark-console): Proxy and Session Issues with YARN HTTP Kerberos Authentication (#3979)

[be...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

benjobs pushed a commit to branch dev-2.1.5
in repository https://gitbox.apache.org/repos/asf/incubator-streampark.git


The following commit(s) were added to refs/heads/dev-2.1.5 by this push:
     new 5aae2e589 fix(streampark-console): Proxy and Session Issues with YARN HTTP Kerberos Authentication (#3979)
5aae2e589 is described below

commit 5aae2e589c8c209f5bbccbbfeff701c2d3e709dc
Author: TiDra <34...@qq.com>
AuthorDate: Tue Aug 20 19:59:45 2024 +0800

    fix(streampark-console): Proxy and Session Issues with YARN HTTP Kerberos Authentication (#3979)
---
 .../console/core/entity/FlinkCluster.java          |  4 +-
 .../core/service/impl/FlinkClusterServiceImpl.java |  2 +-
 .../core/service/impl/ProxyServiceImpl.java        | 95 ++++++++++++++++++++--
 3 files changed, 91 insertions(+), 10 deletions(-)

diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/FlinkCluster.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/FlinkCluster.java
index f43aac68d..3fef77703 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/FlinkCluster.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/FlinkCluster.java
@@ -174,9 +174,7 @@ public class FlinkCluster implements Serializable {
     } else if (ExecutionMode.YARN_SESSION.equals(this.getExecutionModeEnum())) {
       try {
         String restUrl = YarnUtils.getRMWebAppURL(true) + "/proxy/" + this.clusterId + "/overview";
-        String result =
-            HttpClientUtils.httpGetRequest(
-                restUrl, RequestConfig.custom().setConnectTimeout(2000).build());
+        String result = YarnUtils.restRequest(restUrl, 2000);
         JacksonUtils.read(result, Overview.class);
         return true;
       } catch (Exception e) {
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkClusterServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkClusterServiceImpl.java
index 1cc9a82f4..9c1c54b45 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkClusterServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkClusterServiceImpl.java
@@ -185,7 +185,7 @@ public class FlinkClusterServiceImpl extends ServiceImpl<FlinkClusterMapper, Fli
         // 2) setAddress
         if (ExecutionMode.YARN_SESSION.equals(executionModeEnum)) {
           String address =
-              YarnUtils.getRMWebAppURL(true) + "/proxy/" + deployResponse.clusterId() + "/";
+              YarnUtils.getRMWebAppProxyURL() + "/proxy/" + deployResponse.clusterId() + "/";
           flinkCluster.setAddress(address);
         } else {
           flinkCluster.setAddress(deployResponse.address());
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ProxyServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ProxyServiceImpl.java
index bc6d712e7..9bf47b5e3 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ProxyServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ProxyServiceImpl.java
@@ -17,6 +17,8 @@
 
 package org.apache.streampark.console.core.service.impl;
 
+import org.apache.streampark.common.util.HadoopUtils;
+import org.apache.streampark.common.util.SystemPropertyUtils;
 import org.apache.streampark.common.util.YarnUtils;
 import org.apache.streampark.console.base.util.EncryptUtils;
 import org.apache.streampark.console.core.entity.Application;
@@ -34,6 +36,12 @@ import org.apache.streampark.console.system.service.MemberService;
 import org.apache.streampark.flink.kubernetes.FlinkK8sWatcher;
 
 import org.apache.commons.io.IOUtils;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.web.client.RestTemplateBuilder;
@@ -43,6 +51,7 @@ import org.springframework.http.HttpMethod;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
 import org.springframework.http.client.ClientHttpResponse;
+import org.springframework.http.client.HttpComponentsClientHttpRequestFactory;
 import org.springframework.stereotype.Service;
 import org.springframework.web.client.DefaultResponseErrorHandler;
 import org.springframework.web.client.RestTemplate;
@@ -52,6 +61,8 @@ import javax.servlet.http.HttpServletRequest;
 
 import java.io.ByteArrayOutputStream;
 import java.io.InputStream;
+import java.net.URI;
+import java.security.PrivilegedExceptionAction;
 import java.util.Enumeration;
 
 @Service
@@ -73,6 +84,10 @@ public class ProxyServiceImpl implements ProxyService {
 
   private final RestTemplate proxyRestTemplate;
 
+  private final boolean hasYarnHttpKerberosAuth;
+
+  private String lastUsername = "";
+
   public ProxyServiceImpl(RestTemplateBuilder restTemplateBuilder) {
     this.proxyRestTemplate =
         restTemplateBuilder
@@ -84,6 +99,9 @@ public class ProxyServiceImpl implements ProxyService {
                   }
                 })
             .build();
+
+    String yarnHttpAuth = SystemPropertyUtils.get("streampark.yarn.http-auth");
+    this.hasYarnHttpKerberosAuth = "kerberos".equalsIgnoreCase(yarnHttpAuth);
   }
 
   @Override
@@ -117,7 +135,8 @@ public class ProxyServiceImpl implements ProxyService {
       case YARN_SESSION:
         String yarnURL = YarnUtils.getRMWebAppProxyURL();
         url = yarnURL + "/proxy/" + app.getClusterId();
-        break;
+        url += getRequestURL(request).replace("/proxy/flink-ui/" + appId, "");
+        return proxyYarnRequest(request, url);
       case KUBERNETES_NATIVE_APPLICATION:
       case KUBERNETES_NATIVE_SESSION:
         String jobManagerUrl = app.getJobManagerUrl();
@@ -143,7 +162,7 @@ public class ProxyServiceImpl implements ProxyService {
     String yarnURL = YarnUtils.getRMWebAppProxyURL();
     String url = yarnURL + "/proxy/" + appId + "/";
     url += getRequestURL(request).replace("/proxy/yarn/" + appId, "");
-    return proxyRequest(request, url);
+    return proxyYarnRequest(request, url);
   }
 
   @Override
@@ -155,7 +174,8 @@ public class ProxyServiceImpl implements ProxyService {
     return proxyRequest(request, url);
   }
 
-  private ResponseEntity<?> proxyRequest(HttpServletRequest request, String url) throws Exception {
+  private HttpEntity<?> getRequestEntity(HttpServletRequest request, String url, boolean setAuth)
+      throws Exception {
     HttpHeaders headers = new HttpHeaders();
     Enumeration<String> headerNames = request.getHeaderNames();
     while (headerNames.hasMoreElements()) {
@@ -163,9 +183,15 @@ public class ProxyServiceImpl implements ProxyService {
       headers.set(headerName, request.getHeader(headerName));
     }
 
-    String token = serviceHelper.getAuthorization();
-    if (token != null) {
-      headers.set("Authorization", EncryptUtils.encrypt(token));
+    // Ensure the Host header is set correctly.
+    URI uri = new URI(url);
+    headers.set("Host", uri.getHost());
+
+    if (setAuth) {
+      String token = serviceHelper.getAuthorization();
+      if (token != null) {
+        headers.set("Authorization", EncryptUtils.encrypt(token));
+      }
     }
 
     byte[] body = null;
@@ -177,12 +203,69 @@ public class ProxyServiceImpl implements ProxyService {
     }
 
     HttpEntity<?> requestEntity = new HttpEntity<>(body, headers);
+    return requestEntity;
+  }
+
+  private ResponseEntity<?> proxyRequest(HttpServletRequest request, String url) throws Exception {
+    HttpEntity<?> requestEntity = getRequestEntity(request, url, true);
     return proxyRestTemplate.exchange(
         url, HttpMethod.valueOf(request.getMethod()), requestEntity, byte[].class);
   }
 
+  private ResponseEntity<?> proxyYarnRequest(HttpServletRequest request, String url)
+      throws Exception {
+    if (hasYarnHttpKerberosAuth) {
+      UserGroupInformation ugi = HadoopUtils.getUgi();
+
+      HttpEntity<?> requestEntity = getRequestEntity(request, url, false);
+      setRestTemplateCredentials(ugi.getShortUserName());
+
+      return ugi.doAs(
+          new PrivilegedExceptionAction<ResponseEntity<?>>() {
+            @Override
+            public ResponseEntity<?> run() throws Exception {
+              return proxyRestTemplate.exchange(
+                  url, HttpMethod.valueOf(request.getMethod()), requestEntity, byte[].class);
+            }
+          });
+    } else {
+      return proxyRequest(request, url);
+    }
+  }
+
   private String getRequestURL(HttpServletRequest request) {
     return request.getRequestURI()
         + (request.getQueryString() != null ? "?" + request.getQueryString() : "");
   }
+
+  private void setRestTemplateCredentials(String username) {
+    setRestTemplateCredentials(username, null);
+  }
+
+  /**
+   * Configures the RestTemplate's HttpClient connector. This method is primarily used to configure
+   * the HttpClient authentication information and SSL certificate validation policies.
+   *
+   * @param username The username for HTTP basic authentication.
+   * @param password The password for HTTP basic authentication.
+   */
+  private void setRestTemplateCredentials(String username, String password) {
+    // Check if the username is not null and has changed since the last configuration
+    if (username != null && !this.lastUsername.equals(username)) {
+      // Create a new credentials provider
+      BasicCredentialsProvider credentialsProvider = new BasicCredentialsProvider();
+      // Add the username and password for HTTP basic authentication
+      credentialsProvider.setCredentials(
+          AuthScope.ANY, new UsernamePasswordCredentials(username, password));
+
+      // Customize the HttpClient with the credentials provider
+      CloseableHttpClient httpClient =
+          HttpClients.custom().setDefaultCredentialsProvider(credentialsProvider).build();
+      // Set the HttpClient request factory for the RestTemplate
+      this.proxyRestTemplate.setRequestFactory(
+          new HttpComponentsClientHttpRequestFactory(httpClient));
+      // Update the last known username
+      this.lastUsername = username;
+    }
+  }
 }