Support mysql client OpenID Connect Pluggable Authentication

fe/fe-core/src/main/java/com/starrocks/authentication/AuthenticationMgr.java

@@ -122,6 +122,13 @@ public AuthenticationMgr() {
         AuthenticationProviderFactory.installPlugin(
                 KerberosAuthenticationProvider.PLUGIN_NAME, new KerberosAuthenticationProvider());
 
+        AuthenticationProviderFactory.installPlugin(OpenIdConnectAuthenticationProvider.PLUGIN_NAME,
+                new OpenIdConnectAuthenticationProvider(
+                        Config.oidc_jwks_url,
+                        Config.oidc_principal_field,
+                        Config.oidc_required_issuer,
+                        Config.oidc_required_audience));
+
         // default user
         userToAuthenticationInfo = new UserAuthInfoTreeMap();
         UserAuthenticationInfo info = new UserAuthenticationInfo();

fe/fe-core/src/main/java/com/starrocks/authentication/JwkMgr.java

@@ -0,0 +1,37 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed 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
+//
+//     https://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 com.starrocks.authentication;
+
+import com.nimbusds.jose.jwk.JWKSet;
+import com.starrocks.StarRocksFE;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URL;
+import java.text.ParseException;
+
+public class JwkMgr {
+    public JWKSet getJwkSet(String jwksUrl) throws IOException, ParseException {
+        InputStream jwksInputStream;
+        if (jwksUrl.startsWith("http://") || jwksUrl.startsWith("https://")) {
+            jwksInputStream = new URL(jwksUrl).openStream();
+        } else {
+            String filePath = StarRocksFE.STARROCKS_HOME_DIR + "/conf/" + jwksUrl;
+            jwksInputStream = new FileInputStream(filePath);
+        }
+        return JWKSet.load(jwksInputStream);
+    }
+}

fe/fe-core/src/main/java/com/starrocks/authentication/OpenIdConnectAuthenticationProvider.java

@@ -0,0 +1,72 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed 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
+//
+//     https://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 com.starrocks.authentication;
+
+import com.nimbusds.jose.jwk.JWKSet;
+import com.starrocks.mysql.MysqlPassword;
+import com.starrocks.mysql.MysqlProto;
+import com.starrocks.mysql.privilege.AuthPlugin;
+import com.starrocks.server.GlobalStateMgr;
+import com.starrocks.sql.ast.UserAuthOption;
+import com.starrocks.sql.ast.UserIdentity;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.text.ParseException;
+
+public class OpenIdConnectAuthenticationProvider implements AuthenticationProvider {
+    public static final String PLUGIN_NAME = AuthPlugin.AUTHENTICATION_OPENID_CONNECT.name();
+
+    private final String jwksUrl;
+    private final String principalFiled;
+    private final String requireIssuer;
+    private final String requireAudience;
+
+    public OpenIdConnectAuthenticationProvider(String jwksUrl, String principalFiled,
+                                               String requireIssuer, String requireAudience) {
+        this.jwksUrl = jwksUrl;
+        this.principalFiled = principalFiled;
+        this.requireIssuer = requireIssuer;
+        this.requireAudience = requireAudience;
+    }
+
+    @Override
+    public UserAuthenticationInfo analyzeAuthOption(UserIdentity userIdentity, UserAuthOption userAuthOption)
+            throws AuthenticationException {
+        UserAuthenticationInfo info = new UserAuthenticationInfo();
+        info.setAuthPlugin(PLUGIN_NAME);
+        info.setPassword(MysqlPassword.EMPTY_PASSWORD);
+        info.setOrigUserHost(userIdentity.getUser(), userIdentity.getHost());
+        info.setTextForAuthPlugin(userAuthOption == null ? null : userAuthOption.getAuthString());
+        return info;
+    }
+
+    @Override
+    public void authenticate(String user, String host, byte[] authResponse, byte[] randomString,
+                             UserAuthenticationInfo authenticationInfo) throws AuthenticationException {
+        ByteBuffer authBuffer = ByteBuffer.wrap(authResponse);
+        //1 Byte for capability mysql client
+        MysqlProto.readInt1(authBuffer);
+        byte[] openIdConnect = MysqlProto.readLenEncodedString(authBuffer);
+
+        JWKSet jwkSet;
+        try {
+            jwkSet = GlobalStateMgr.getCurrentState().getJwkMgr().getJwkSet(jwksUrl);
+        } catch (IOException | ParseException e) {
+            throw new AuthenticationException(e.getMessage());
+        }
+        OpenIdConnectVerifier.verify(new String(openIdConnect), user, jwkSet, principalFiled, requireIssuer, requireAudience);
+    }
+}

fe/fe-core/src/main/java/com/starrocks/authentication/OpenIdConnectVerifier.java

@@ -0,0 +1,86 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed 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
+//
+//     https://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 com.starrocks.authentication;
+
+import com.google.common.base.Preconditions;
+import com.nimbusds.jose.JOSEException;
+import com.nimbusds.jose.crypto.RSASSAVerifier;
+import com.nimbusds.jose.jwk.JWK;
+import com.nimbusds.jose.jwk.JWKSet;
+import com.nimbusds.jwt.JWTClaimsSet;
+import com.nimbusds.jwt.SignedJWT;
+import org.json.JSONObject;
+
+import java.security.interfaces.RSAPublicKey;
+import java.text.ParseException;
+
+public class OpenIdConnectVerifier {
+
+    public static void verify(String oidcToken,
+                              String userName,
+                              JWKSet jwkSet,
+                              String principalFiled,
+                              String requiredIssuer,
+                              String requiredAudience) throws AuthenticationException {
+        try {
+            JSONObject openIdConnectJson = new JSONObject(oidcToken);
+            String accessToken = openIdConnectJson.getString("access_token");
+            OpenIdConnectVerifier.verifyJWT(accessToken, jwkSet);
+
+            String idToken = openIdConnectJson.getString("id_token");
+            OpenIdConnectVerifier.verifyJWT(idToken, jwkSet);
+
+            SignedJWT signedJWT = SignedJWT.parse(idToken);
+            JWTClaimsSet claims = signedJWT.getJWTClaimsSet();
+            String jwtUserName = claims.getStringClaim(principalFiled);
+
+            if (jwtUserName == null) {
+                throw new AuthenticationException("Can not get specified principal " + principalFiled);
+            }
+
+            if (!jwtUserName.equalsIgnoreCase(userName)) {
+                throw new AuthenticationException("Login name " + userName + " is not matched to user " + jwtUserName);
+            }
+
+            if (requiredIssuer != null && !requiredIssuer.isEmpty() && !requiredIssuer.equals(claims.getIssuer())) {
+                throw new AuthenticationException("Issuer (iss) field " + claims.getIssuer() + " is invalid");
+            }
+
+            if (requiredAudience != null && !requiredAudience.isEmpty() && !claims.getAudience().contains(requiredAudience)) {
+                throw new AuthenticationException("Audience (aud) field " + claims.getAudience() + " is invalid");
+            }
+        } catch (Exception e) {
+            throw new AuthenticationException(e.getMessage());
+        }
+    }
+
+    private static void verifyJWT(String jwt, JWKSet jwkSet) throws AuthenticationException, ParseException, JOSEException {
+        Preconditions.checkNotNull(jwt);
+
+        SignedJWT signedJWT = SignedJWT.parse(jwt);
+        String kid = signedJWT.getHeader().getKeyID();
+
+        JWK jwk = jwkSet.getKeyByKeyId(kid);
+        if (jwk == null) {
+            throw new AuthenticationException("Cannot find public key for kid: " + kid);
+        }
+
+        RSAPublicKey publicKey = jwk.toRSAKey().toRSAPublicKey();
+        RSASSAVerifier verifier = new RSASSAVerifier(publicKey);
+        if (!signedJWT.verify(verifier)) {
+            throw new AuthenticationException("JWT with kid " + kid + " is invalid");
+        }
+    }
+}

fe/fe-core/src/main/java/com/starrocks/common/Config.java

@@ -3474,4 +3474,32 @@ public class Config extends ConfigBase {
 
     @ConfField(mutable = false)
     public static int max_historical_automated_cluster_snapshot_jobs = 100;
+
+    /**
+     * The URL to a JWKS service or a local file in the conf dir
+     */
+    @ConfField(mutable = false)
+    public static String oidc_jwks_url = "";
+
+    /**
+     * String to identify the field in the JWT that identifies the subject of the JWT.
+     * The default value is sub.
+     * The value of this field must be the same as the user specified when logging into StarRocks.
+     */
+    @ConfField(mutable = false)
+    public static String oidc_principal_field = "sub";
+
+    /**
+     * Specifies a string that must match the value of the JWT’s issuer (iss) field in order to consider this JWT valid.
+     * The iss field in the JWT identifies the principal that issued the JWT.
+     */
+    @ConfField(mutable = false)
+    public static String oidc_required_issuer = "";
+
+    /**
+     * Specifies a string that must match the value of the JWT’s Audience (aud) field in order to consider this JWT valid.
+     * The aud field in the JWT identifies the recipients that the JWT is intended for.
+     */
+    @ConfField(mutable = false)
+    public static String oidc_required_audience = "";
 }

fe/fe-core/src/main/java/com/starrocks/mysql/MysqlHandshakePacket.java

@@ -62,10 +62,12 @@ public class MysqlHandshakePacket extends MysqlPacket {
     public static final String NATIVE_AUTH_PLUGIN_NAME = "mysql_native_password";
     public static final String CLEAR_PASSWORD_PLUGIN_NAME = "mysql_clear_password";
     public static final String AUTHENTICATION_KERBEROS_CLIENT = "authentication_kerberos_client";
+    public static final String AUTHENTICATION_OPENID_CONNECT_CLIENT = "authentication_openid_connect_client";
 
     private static final ImmutableMap<String, Boolean> SUPPORTED_PLUGINS = new ImmutableMap.Builder<String, Boolean>()
             .put(NATIVE_AUTH_PLUGIN_NAME, true)
             .put(CLEAR_PASSWORD_PLUGIN_NAME, true)
+            .put(AUTHENTICATION_OPENID_CONNECT_CLIENT, true)
             .build();
 
     // connection id used in KILL statement.

fe/fe-core/src/main/java/com/starrocks/mysql/privilege/AuthPlugin.java

@@ -19,5 +19,6 @@ public enum AuthPlugin {
     MYSQL_NATIVE_PASSWORD,
     AUTHENTICATION_LDAP_SIMPLE,
     AUTHENTICATION_KERBEROS,
-    AUTHENTICATION_LDAP_SIMPLE_FOR_EXTERNAL
+    AUTHENTICATION_LDAP_SIMPLE_FOR_EXTERNAL,
+    AUTHENTICATION_OPENID_CONNECT
 }

fe/fe-core/src/main/java/com/starrocks/server/GlobalStateMgr.java

@@ -47,6 +47,7 @@
 import com.starrocks.analysis.LiteralExpr;
 import com.starrocks.analysis.TableName;
 import com.starrocks.authentication.AuthenticationMgr;
+import com.starrocks.authentication.JwkMgr;
 import com.starrocks.authorization.AccessControlProvider;
 import com.starrocks.authorization.AuthorizationMgr;
 import com.starrocks.authorization.DefaultAuthorizationProvider;
@@ -526,6 +527,8 @@ public class GlobalStateMgr {
     private final ReportHandler reportHandler;
     private final TabletCollector tabletCollector;
 
+    private JwkMgr jwkMgr;
+
     public NodeMgr getNodeMgr() {
         return nodeMgr;
     }
@@ -834,6 +837,8 @@ public void transferToNonLeader(FrontendNodeType newType) {
 
         this.reportHandler = new ReportHandler();
         this.tabletCollector = new TabletCollector();
+
+        this.jwkMgr = new JwkMgr();
     }
 
     public static void destroyCheckpoint() {
@@ -2724,4 +2729,12 @@ public void shutdown() {
     public ReportHandler getReportHandler() {
         return reportHandler;
     }
+
+    public JwkMgr getJwkMgr() {
+        return jwkMgr;
+    }
+
+    public void setJwkMgr(JwkMgr jwkMgr) {
+        this.jwkMgr = jwkMgr;
+    }
 }