IMPALA-12232: Validate JWT aud/iss claims Validate configured audienceClaims and issuerClaims after token signature verification for both JWT and OAuth auth flows, including HS2 HTTP and webserver paths. Extend oauth_servers parsing to accept audienceClaims/issuerClaims and align claim validation behavior with JWT RFC requirements for aud/iss semantics. Address follow-up review feedback by adding targeted null checks, preserving HTTP error response headers in test transport code, and strengthening JwtHttp negative tests to assert claim-specific auth failures from WWW-Authenticate headers. Remove obsolete --jwt_validate_signature usage in JwtHttpTest startup flags so custom-cluster tests use supported oauth_servers options. Testing: - Linux: fe mvn test -Dtest='JwtHttpTest#testJwtAuthWithWrongIssuerClaim+ testJwtAuthWithWrongAudienceClaim+testJwtAuthWithMissingAudienceClaim' - Linux: be/build/debug/util/oauth-server-config-test Change-Id: I0a00b126359f2bc7e2f73d894cebc2b9014c7375 Assisted-by: GPT-5.3 (Cursor) Reviewed-on: http://gerrit.cloudera.org:8080/24472 Tested-by: Impala Public Jenkins <impala-public-jenkins@cloudera.com> Reviewed-by: Jason Fehr <jfehr@cloudera.com>
diff --git a/be/src/util/jwt-util-test.cc b/be/src/util/jwt-util-test.cc index cbf0980..480eb24 100644 --- a/be/src/util/jwt-util-test.cc +++ b/be/src/util/jwt-util-test.cc
@@ -1303,4 +1303,110 @@ EXPECT_OK(status); } +TEST(JwtUtilTest, ValidateJwtTokenFormat) { + EXPECT_OK(JWTHelper::ValidateTokenFormat("abc.def.ghi")); + + Status status = JWTHelper::ValidateTokenFormat("abc.def"); + ASSERT_FALSE(status.ok()); + ASSERT_TRUE(status.GetDetail().find("expected exactly two periods") != string::npos) + << " Actual error: " << status.GetDetail(); + + status = JWTHelper::ValidateTokenFormat("abc.def.ghi!"); + ASSERT_FALSE(status.ok()); + ASSERT_TRUE(status.GetDetail().find("outside the JWT compact serialization alphabet") + != string::npos) + << " Actual error: " << status.GetDetail(); + + status = JWTHelper::ValidateTokenFormat("abc=.def.ghi"); + ASSERT_FALSE(status.ok()); + ASSERT_TRUE(status.GetDetail().find("outside the JWT compact serialization alphabet") + != string::npos) + << " Actual error: " << status.GetDetail(); +} + +TEST(JwtUtilTest, VerifyJwtAudienceAndIssuerClaims) { + picojson::array audience_values; + audience_values.push_back(picojson::value("impala-service")); + audience_values.push_back(picojson::value("other-service")); + + auto token = + jwt::create() + .set_issuer("auth0") + .set_type("JWS") + .set_algorithm("RS256") + .set_key_id(kid_1) + .set_payload_claim("aud", picojson::value(audience_values)) + .set_payload_claim("username", picojson::value("impala")) + .sign(jwt::algorithm::rs256(rsa_pub_key_pem, rsa_priv_key_pem, "", "")); + + JWTHelper::UniqueJWTDecodedToken decoded_token; + Status status = JWTHelper::Decode(token, decoded_token); + EXPECT_OK(status); + + EXPECT_OK(JWTHelper::VerifyJwtClaims( + decoded_token.get(), {"impala-service"}, {"auth0"})); + EXPECT_OK(JWTHelper::VerifyJwtClaims( + decoded_token.get(), {"missing-service", "other-service"}, {"auth0"})); + + status = JWTHelper::VerifyJwtClaims( + decoded_token.get(), {"missing-service"}, {"auth0"}); + ASSERT_FALSE(status.ok()); + ASSERT_TRUE(status.GetDetail().find("Claim 'aud' value") != string::npos) + << " Actual error: " << status.GetDetail(); + + status = JWTHelper::VerifyJwtClaims( + decoded_token.get(), {"impala-service"}, {"issuer2"}); + ASSERT_FALSE(status.ok()); + ASSERT_TRUE(status.GetDetail().find("Claim 'iss' value") != string::npos) + << " Actual error: " << status.GetDetail(); +} + +TEST(JwtUtilTest, VerifyJwtClaimsRequireConfiguredClaims) { + auto token = + jwt::create() + .set_type("JWS") + .set_algorithm("RS256") + .set_key_id(kid_1) + .set_payload_claim("username", picojson::value("impala")) + .sign(jwt::algorithm::rs256(rsa_pub_key_pem, rsa_priv_key_pem, "", "")); + + JWTHelper::UniqueJWTDecodedToken decoded_token; + Status status = JWTHelper::Decode(token, decoded_token); + EXPECT_OK(status); + + status = JWTHelper::VerifyJwtClaims(decoded_token.get(), {"impala-service"}, {}); + ASSERT_FALSE(status.ok()); + ASSERT_TRUE(status.GetDetail().find("Claim 'aud' was not present") != string::npos) + << " Actual error: " << status.GetDetail(); + + status = JWTHelper::VerifyJwtClaims(decoded_token.get(), {}, {"auth0"}); + ASSERT_FALSE(status.ok()); + ASSERT_TRUE(status.GetDetail().find("Claim 'iss' was not present") != string::npos) + << " Actual error: " << status.GetDetail(); +} + +TEST(JwtUtilTest, VerifyJwtClaimsRejectsArrayIssuerClaim) { + picojson::array issuer_values; + issuer_values.push_back(picojson::value("auth0")); + issuer_values.push_back(picojson::value("issuer2")); + + auto token = + jwt::create() + .set_type("JWS") + .set_algorithm("RS256") + .set_key_id(kid_1) + .set_payload_claim("iss", picojson::value(issuer_values)) + .set_payload_claim("username", picojson::value("impala")) + .sign(jwt::algorithm::rs256(rsa_pub_key_pem, rsa_priv_key_pem, "", "")); + + JWTHelper::UniqueJWTDecodedToken decoded_token; + Status status = JWTHelper::Decode(token, decoded_token); + EXPECT_OK(status); + + status = JWTHelper::VerifyJwtClaims(decoded_token.get(), {}, {"auth0"}); + ASSERT_FALSE(status.ok()); + ASSERT_TRUE(status.GetDetail().find("Claim 'iss' value") != string::npos) + << " Actual error: " << status.GetDetail(); +} + } // namespace impala
diff --git a/be/src/util/jwt-util.cc b/be/src/util/jwt-util.cc index ab0c6af..cd6846a 100644 --- a/be/src/util/jwt-util.cc +++ b/be/src/util/jwt-util.cc
@@ -16,6 +16,7 @@ // under the License. #include <string.h> +#include <algorithm> #include <cerrno> #include <ostream> #include <unordered_map> @@ -48,9 +49,6 @@ #include "util/kudu-status-util.h" #include "util/test-info.h" -DECLARE_int32(jwks_update_frequency_s); -DECLARE_int32(jwks_pulling_timeout_s); - // Support only a single x5c certificate. // Update MAX_X5C_CERTIFICATES when we can // support more than one. @@ -66,6 +64,55 @@ using rapidjson::Document; using rapidjson::Value; +namespace { + +bool IsAllowedJwtChar(char c) { + return std::isalnum(static_cast<unsigned char>(c)) || c == '-' || c == '_' || c == '.'; +} + +const char* const AUD_CLAIM_NAME = "aud"; +const char* const ISS_CLAIM_NAME = "iss"; + +bool ClaimValueMatches(const picojson::value& claim_value, + const vector<string>& allowed_values, bool allow_array_values) { + if (claim_value.is<string>()) { + const string& value = claim_value.get<string>(); + return std::find(allowed_values.begin(), allowed_values.end(), value) + != allowed_values.end(); + } + if (allow_array_values && claim_value.is<picojson::array>()) { + const picojson::array& values = claim_value.get<picojson::array>(); + for (const picojson::value& value : values) { + if (value.is<string>()) { + const string& str_value = value.get<string>(); + if (std::find(allowed_values.begin(), allowed_values.end(), str_value) + != allowed_values.end()) { + return true; + } + } + } + } + return false; +} + +Status VerifyClaim(const DecodedJWT& decoded_jwt, const string& claim_name, + const vector<string>& allowed_values, bool allow_array_values) { + if (allowed_values.empty()) return Status::OK(); + if (!decoded_jwt.has_payload_claim(claim_name)) { + return Status(TErrorCode::JWT_VERIFY_FAILED, + Substitute("Claim '$0' was not present", claim_name)); + } + picojson::value claim_value = decoded_jwt.get_payload_claim(claim_name).to_json(); + if (!ClaimValueMatches(claim_value, allowed_values, allow_array_values)) { + return Status(TErrorCode::JWT_VERIFY_FAILED, + Substitute("Claim '$0' value '$1' is not allowed", claim_name, + claim_value.serialize())); + } + return Status::OK(); +} + +} // anonymous namespace + // JWK Set (JSON Web Key Set) is JSON data structure that represents a set of JWKs. // This class parses JWKS file. class JWKSetParser { @@ -879,6 +926,18 @@ return jwks_mgr_->GetJWKSSnapshot(); } +Status JWTHelper::ValidateTokenFormat(const string& token) { + if (std::count(token.begin(), token.end(), '.') != 2) { + return Status(TErrorCode::JWT_VERIFY_FAILED, + "Token is not in compact JWT format: expected exactly two periods"); + } + if (!std::all_of(token.begin(), token.end(), IsAllowedJwtChar)) { + return Status(TErrorCode::JWT_VERIFY_FAILED, + "Token contains characters outside the JWT compact serialization alphabet"); + } + return Status::OK(); +} + // Decode the given JWT token. Status JWTHelper::Decode(const string& token, UniqueJWTDecodedToken& decoded_token_out) { Status status; @@ -991,6 +1050,16 @@ return status; } +Status JWTHelper::VerifyJwtClaims(const JWTDecodedToken* decoded_token, + const vector<string>& audience_claims, const vector<string>& issuer_claims) { + DCHECK(decoded_token != nullptr); + RETURN_IF_ERROR( + VerifyClaim(decoded_token->decoded_jwt_, AUD_CLAIM_NAME, audience_claims, true)); + RETURN_IF_ERROR( + VerifyClaim(decoded_token->decoded_jwt_, ISS_CLAIM_NAME, issuer_claims, false)); + return Status::OK(); +} + Status JWTHelper::CanVerify( const JWTDecodedToken* decoded_token, bool* can_verify_out) const { DCHECK(initialized_); @@ -1045,6 +1114,34 @@ return Status::OK(); } +string JWTHelper::GetPayloadClaim( + const JWTDecodedToken* decoded_token, const string& claim_name) { + DCHECK(decoded_token != nullptr); + try { + if (decoded_token->decoded_jwt_.has_payload_claim(claim_name)) { + return decoded_token->decoded_jwt_.get_payload_claim(claim_name) + .to_json() + .serialize(); + } + } catch (const std::exception& e) { + VLOG(2) << "Error retrieving JWT claim '" << claim_name << "' for logging: " + << e.what(); + } + return ""; +} + +string JWTHelper::GetKeyId(const JWTDecodedToken* decoded_token) { + DCHECK(decoded_token != nullptr); + try { + if (decoded_token->decoded_jwt_.has_key_id()) { + return decoded_token->decoded_jwt_.get_key_id(); + } + } catch (const std::exception& e) { + VLOG(2) << "Error retrieving JWT key ID for logging: " << e.what(); + } + return ""; +} + Status JWTHelper::GetCustomClaimUsername(const JWTDecodedToken* decoded_token, const string& jwt_custom_claim_username, string& username) { DCHECK(decoded_token != nullptr);
diff --git a/be/src/util/jwt-util.h b/be/src/util/jwt-util.h index 8d06fb6..53b0c6c 100644 --- a/be/src/util/jwt-util.h +++ b/be/src/util/jwt-util.h
@@ -74,11 +74,20 @@ static Status Decode( const std::string& token, UniqueJWTDecodedToken& decoded_token_out); + /// Verify the token is in compact JWT form before decoding it. + static Status ValidateTokenFormat(const std::string& token); + /// Verify the token's signature with the JWKS. The token should be already decoded by /// calling Decode(). /// Return Status::OK if the verification is successful. Status Verify(const JWTDecodedToken* decoded_token) const; + /// Verify the aud and iss claims against configured allowed values. Empty vectors + /// disable validation for that claim. + static Status VerifyJwtClaims(const JWTDecodedToken* decoded_token, + const std::vector<std::string>& audience_claims, + const std::vector<std::string>& issuer_claims); + /// Returns true if this helper has a matching key candidate for the token's key id / /// algorithm family, and false otherwise. Status CanVerify(const JWTDecodedToken* decoded_token, bool* can_verify_out) const; @@ -88,6 +97,14 @@ static Status GetCustomClaimUsername(const JWTDecodedToken* decoded_token, const std::string& custom_claim_username, std::string& username); + /// Returns the JSON value of a payload claim, or an empty string if the claim is + /// absent. + static std::string GetPayloadClaim( + const JWTDecodedToken* decoded_token, const std::string& claim_name); + + /// Returns the key ID from the decoded JWT header, or an empty string if absent. + static std::string GetKeyId(const JWTDecodedToken* decoded_token); + /// Return snapshot of JWKS. std::shared_ptr<const JWKSSnapshot> GetJWKS() const;
diff --git a/be/src/util/oauth-server-config-test.cc b/be/src/util/oauth-server-config-test.cc index ff056de..110ea10 100644 --- a/be/src/util/oauth-server-config-test.cc +++ b/be/src/util/oauth-server-config-test.cc
@@ -46,7 +46,9 @@ "jwksUrl": "https://example.com/jwks.json", "jwksPullTimeoutSecs": 15, "jwksUpdateFrequencySecs": 14400, - "usernameClaim": "preferred_username" + "usernameClaim": "preferred_username", + "audienceClaims": ["impala-service"], + "issuerClaims": ["auth0"] }, { "jwksFilePath": "/opt/auth-servers/jwks.json" @@ -62,9 +64,15 @@ EXPECT_EQ(15, configs[0].jwks_pull_timeout_secs); EXPECT_EQ(14400, configs[0].jwks_update_frequency_secs); EXPECT_EQ("preferred_username", configs[0].username_claim); + ASSERT_EQ(1, configs[0].audience_claims.size()); + EXPECT_EQ("impala-service", configs[0].audience_claims[0]); + ASSERT_EQ(1, configs[0].issuer_claims.size()); + EXPECT_EQ("auth0", configs[0].issuer_claims[0]); EXPECT_EQ("/opt/auth-servers/jwks.json", configs[1].jwks_uri); EXPECT_TRUE(configs[1].is_local_jwks); EXPECT_EQ("username", configs[1].username_claim); + EXPECT_TRUE(configs[1].audience_claims.empty()); + EXPECT_TRUE(configs[1].issuer_claims.empty()); } TEST(OAuthServerConfigTest, GetJwksUriPrefersFilePath) { @@ -109,6 +117,51 @@ status.GetDetail().find("cannot specify both jwksFilePath and jwksUrl")); } +TEST(OAuthServerConfigTest, ParseOAuthServersJsonRejectsNonArrayAudienceClaims) { + google::FlagSaver flag_saver; + vector<OAuthServerConfig> configs; + FLAGS_oauth_servers = R"([{ + "jwksFilePath":"/tmp/jwks.json", + "audienceClaims":"impala-service" + }])"; + Status status = BuildOAuthServerConfigs(&configs); + EXPECT_FALSE(status.ok()); + EXPECT_NE(string::npos, + status.GetDetail().find( + "oauth_servers entry field 'audienceClaims' " + "must be an array")); +} + +TEST(OAuthServerConfigTest, ParseOAuthServersJsonRejectsNonArrayIssuerClaims) { + google::FlagSaver flag_saver; + vector<OAuthServerConfig> configs; + FLAGS_oauth_servers = R"([{ + "jwksFilePath":"/tmp/jwks.json", + "issuerClaims":"auth0" + }])"; + Status status = BuildOAuthServerConfigs(&configs); + EXPECT_FALSE(status.ok()); + EXPECT_NE(string::npos, + status.GetDetail().find( + "oauth_servers entry field 'issuerClaims' " + "must be an array")); +} + +TEST(OAuthServerConfigTest, ParseOAuthServersJsonAcceptsEmptyClaimValuesAsUnset) { + google::FlagSaver flag_saver; + vector<OAuthServerConfig> configs; + FLAGS_oauth_servers = R"([{ + "jwksFilePath":"/tmp/jwks.json", + "usernameClaim":"sub", + "audienceClaims":[], + "issuerClaims":[""] + }])"; + ASSERT_OK(BuildOAuthServerConfigs(&configs)); + ASSERT_EQ(1, configs.size()); + EXPECT_TRUE(configs[0].audience_claims.empty()); + EXPECT_TRUE(configs[0].issuer_claims.empty()); +} + TEST(OAuthServerConfigTest, BuildOAuthServerConfigsUsesOauthServersJson) { google::FlagSaver flag_saver; const string oauth_servers_json =
diff --git a/be/src/util/oauth-server-config.cc b/be/src/util/oauth-server-config.cc index b27b2df..3bbf06b 100644 --- a/be/src/util/oauth-server-config.cc +++ b/be/src/util/oauth-server-config.cc
@@ -49,7 +49,8 @@ DEFINE_string(oauth_servers, "", "JSON array of OAuth server configurations for JWT/OAuth token verification. Each " "element may specify caCertFilePath, verifyServerCert, jwksFilePath, jwksUrl, " - "jwksPullTimeoutSecs, jwksUpdateFrequencySecs, and usernameClaim."); + "jwksPullTimeoutSecs, jwksUpdateFrequencySecs, usernameClaim, " + "audienceClaims, and issuerClaims."); namespace impala { @@ -130,6 +131,26 @@ return Status::OK(); } +Status ReadOptionalStringArrayField(const rapidjson::Value& obj, const char* field_name, + vector<string>* values_out) { + DCHECK(values_out != nullptr); + if (!obj.HasMember(field_name)) return Status::OK(); + const rapidjson::Value& field = obj[field_name]; + if (!field.IsArray()) { + return Status(Substitute("oauth_servers entry field '$0' must be an array", + field_name)); + } + for (rapidjson::SizeType i = 0; i < field.Size(); ++i) { + if (!field[i].IsString()) { + return Status(Substitute("oauth_servers entry field '$0' must be an array of " + "strings", field_name)); + } + // Preserve backward compatibility: empty entries are treated as unset values. + if (field[i].GetStringLength() > 0) values_out->push_back(field[i].GetString()); + } + return Status::OK(); +} + Status ParseOAuthServerObject( const rapidjson::Value& obj, OAuthServerConfig* config_out) { if (!obj.IsObject()) { @@ -150,6 +171,10 @@ RETURN_IF_ERROR(ReadOptionalIntField( obj, "jwksUpdateFrequencySecs", &config.jwks_update_frequency_secs)); RETURN_IF_ERROR(ReadOptionalStringField(obj, "usernameClaim", &config.username_claim)); + RETURN_IF_ERROR(ReadOptionalStringArrayField( + obj, "audienceClaims", &config.audience_claims)); + RETURN_IF_ERROR(ReadOptionalStringArrayField( + obj, "issuerClaims", &config.issuer_claims)); RETURN_IF_ERROR(ValidateJwksSettings(config)); if (config.username_claim.empty()) { return Status("oauth_servers entry field 'usernameClaim' must not be empty");
diff --git a/be/src/util/oauth-server-config.h b/be/src/util/oauth-server-config.h index 1c6dab6..1cc580b 100644 --- a/be/src/util/oauth-server-config.h +++ b/be/src/util/oauth-server-config.h
@@ -38,6 +38,8 @@ int32_t jwks_pull_timeout_secs = DEFAULT_JWKS_PULL_TIMEOUT_SECS; int32_t jwks_update_frequency_secs = DEFAULT_JWKS_UPDATE_FREQUENCY_SECS; std::string username_claim = "username"; + std::vector<std::string> audience_claims; + std::vector<std::string> issuer_claims; }; /// Builds the full list of OAuth server configs from --oauth_servers plus any legacy
diff --git a/be/src/util/oauth-servers-manager-test.cc b/be/src/util/oauth-servers-manager-test.cc index 5d18a84..3f282e4 100644 --- a/be/src/util/oauth-servers-manager-test.cc +++ b/be/src/util/oauth-servers-manager-test.cc
@@ -124,6 +124,53 @@ EXPECT_EQ("test-user", username); } +TEST(OAuthServersManagerTest, VerifyTokenFromSecondServerWhenFirstAudienceDoesNotMatch) { + google::FlagSaver flag_saver; + const string impala_home = GetImpalaHome(); + const string jwks_path = + Substitute("$0/testdata/jwt/jwks_signing.json", impala_home); + const string jwt_path = Substitute("$0/testdata/jwt/jwt_signed", impala_home); + const string token = ReadTrimmedFile(jwt_path); + ASSERT_FALSE(token.empty()); + + FLAGS_oauth_servers = Substitute( + R"([{"jwksFilePath":"$0","usernameClaim":"sub",)" + R"("audienceClaims":["wrong-audience"]},)" + R"({"jwksFilePath":"$0","usernameClaim":"sub",)" + R"("audienceClaims":["impala-tests"]}])", + jwks_path); + OAuthServersManager manager; + ASSERT_OK(manager.Init()); + ASSERT_EQ(2, manager.size()); + + string username; + ASSERT_OK(manager.AuthenticateBearerToken(token, &username)); + EXPECT_EQ("test-user", username); +} + +TEST(OAuthServersManagerTest, VerifyFailsWhenAudienceClaimsDoNotMatch) { + google::FlagSaver flag_saver; + const string impala_home = GetImpalaHome(); + const string jwks_path = + Substitute("$0/testdata/jwt/jwks_signing.json", impala_home); + const string jwt_path = Substitute("$0/testdata/jwt/jwt_signed", impala_home); + const string token = ReadTrimmedFile(jwt_path); + ASSERT_FALSE(token.empty()); + + FLAGS_oauth_servers = Substitute( + R"([{"jwksFilePath":"$0","usernameClaim":"sub",)" + R"("audienceClaims":["wrong-audience"]}])", + jwks_path); + OAuthServersManager manager; + ASSERT_OK(manager.Init()); + ASSERT_EQ(1, manager.size()); + + string username; + Status status = manager.AuthenticateBearerToken(token, &username); + EXPECT_FALSE(status.ok()); + EXPECT_NE(string::npos, status.GetDetail().find("Claim 'aud' value")); +} + TEST(OAuthServersManagerTest, VerifyFailsWhenOnlyWrongJwksConfigured) { google::FlagSaver flag_saver; const string impala_home = GetImpalaHome();
diff --git a/be/src/util/oauth-servers-manager.cc b/be/src/util/oauth-servers-manager.cc index 176f373..6f572c1 100644 --- a/be/src/util/oauth-servers-manager.cc +++ b/be/src/util/oauth-servers-manager.cc
@@ -55,6 +55,8 @@ config.jwks_update_frequency_secs); if (!status.ok()) return status; verifier.username_claim = std::move(config.username_claim); + verifier.audience_claims = std::move(config.audience_claims); + verifier.issuer_claims = std::move(config.issuer_claims); jwt_helpers_->emplace_back(std::move(verifier)); } return Status::OK(); @@ -65,6 +67,7 @@ DCHECK(jwt_helpers_); DCHECK(username_out != nullptr); username_out->clear(); + RETURN_IF_ERROR(JWTHelper::ValidateTokenFormat(token)); JWTHelper::UniqueJWTDecodedToken decoded_token; RETURN_IF_ERROR(JWTHelper::Decode(token, decoded_token)); return Verify(decoded_token.get(), username_out); @@ -83,7 +86,12 @@ RETURN_IF_ERROR(FindMatchingServer(decoded_token, next_idx, &matched_server_idx)); if (matched_server_idx == size()) break; attempted_signature_verification = true; - Status status = jwt_helpers_->at(matched_server_idx).jwt_helper.Verify(decoded_token); + const OAuthServerVerifier& verifier = jwt_helpers_->at(matched_server_idx); + Status status = verifier.jwt_helper.Verify(decoded_token); + if (status.ok()) { + status = JWTHelper::VerifyJwtClaims( + decoded_token, verifier.audience_claims, verifier.issuer_claims); + } if (status.ok()) { if (username_out != nullptr) { RETURN_IF_ERROR(GetUsername(decoded_token, matched_server_idx, username_out));
diff --git a/be/src/util/oauth-servers-manager.h b/be/src/util/oauth-servers-manager.h index 4ac2ad0..0863bfe 100644 --- a/be/src/util/oauth-servers-manager.h +++ b/be/src/util/oauth-servers-manager.h
@@ -65,6 +65,8 @@ struct OAuthServerVerifier { JWTHelper jwt_helper; std::string username_claim; + std::vector<std::string> audience_claims; + std::vector<std::string> issuer_claims; }; /// Extracts the username using the username claim from the server at 'server_idx'.
diff --git a/fe/src/test/java/org/apache/impala/customcluster/JwtHttpTest.java b/fe/src/test/java/org/apache/impala/customcluster/JwtHttpTest.java index 468b24e..13fb2de 100644 --- a/fe/src/test/java/org/apache/impala/customcluster/JwtHttpTest.java +++ b/fe/src/test/java/org/apache/impala/customcluster/JwtHttpTest.java
@@ -168,6 +168,35 @@ } } + private String readTrimmedFile(Path path) { + try { + return Files.readString(path).trim(); + } catch (IOException e) { + fail("Failed to read file: " + e.getMessage()); + return ""; + } + } + + private String getResponseHeaderIgnoreCase( + Map<String, List<String>> responseHeaders, String headerName) { + if (responseHeaders == null) return ""; + for (Map.Entry<String, List<String>> entry : responseHeaders.entrySet()) { + if (entry.getKey() == null || entry.getValue() == null) continue; + if (entry.getKey().equalsIgnoreCase(headerName) && !entry.getValue().isEmpty()) { + return entry.getValue().get(0); + } + } + return ""; + } + + private void verifyBearerErrorContainsClaim( + THttpClientWithHeaders transport, String claimName) { + String authHeader = getResponseHeaderIgnoreCase( + transport.getResponseHeaders(), "WWW-Authenticate"); + assertThat(authHeader, containsString("invalid_token")); + assertThat(authHeader, containsString(String.format("Claim '%s'", claimName))); + } + static void verifySuccess(TStatus status) throws Exception { if (status.getStatusCode() == TStatusCode.SUCCESS_STATUS || status.getStatusCode() == TStatusCode.SUCCESS_WITH_INFO_STATUS) { @@ -224,7 +253,8 @@ setUp(String.format( "--jwt_token_auth=true --jwks_file_path=%s --jwt_allow_without_tls=true", jwksFilename)); - THttpClient transport = new THttpClient("http://localhost:28000"); + THttpClientWithHeaders transport = + new THttpClientWithHeaders("http://localhost:28000"); Map<String, String> headers = new HashMap<String, String>(); // Case 1: Authenticate with valid JWT Token in HTTP header. @@ -289,6 +319,152 @@ } /** + * Tests JWT authentication when issuer claim validation is configured. + */ + @Test + public void testJwtAuthWithIssuerClaimValidation() throws Exception { + createJWKSForWebServer_ = false; + String jwksFilename = + new File(System.getenv("IMPALA_HOME"), + String.format("testdata/jwt/%s", JWKS_FILE_NAME)).getPath(); + String oauthServers = String.format( + "[{\"jwksFilePath\":\"%s\",\"issuerClaims\":[\"auth0\"]}]", + jwksFilename); + setUp(String.format( + "--jwt_token_auth=true --oauth_servers='%s' --jwt_allow_without_tls=true", + oauthServers)); + THttpClientWithHeaders transport = + new THttpClientWithHeaders("http://localhost:28000"); + Map<String, String> headers = new HashMap<String, String>(); + + headers.put("Authorization", "Bearer " + jwtToken_); + headers.put("X-Forwarded-For", "127.0.0.1"); + transport.setCustomHeaders(headers); + transport.open(); + TCLIService.Iface client = new TCLIService.Client(new TBinaryProtocol(transport)); + + TOpenSessionReq openReq = new TOpenSessionReq(); + TOpenSessionResp openResp = client.OpenSession(openReq); + verifyJwtAuthMetrics(1, 0); + execAndFetch( + client, openResp.getSessionHandle(), "select logged_in_user()", "impala"); + } + + /** + * Tests JWT authentication fails when issuer validation is configured with a + * non-matching value. + */ + @Test + public void testJwtAuthWithWrongIssuerClaim() throws Exception { + createJWKSForWebServer_ = false; + String jwksFilename = + new File(System.getenv("IMPALA_HOME"), + String.format("testdata/jwt/%s", JWKS_FILE_NAME)).getPath(); + String oauthServers = String.format( + "[{\"jwksFilePath\":\"%s\",\"issuerClaims\":[\"issuer2\"]}]", + jwksFilename); + setUp(String.format( + "--jwt_token_auth=true --oauth_servers='%s' --jwt_allow_without_tls=true", + oauthServers)); + THttpClientWithHeaders transport = + new THttpClientWithHeaders("http://localhost:28000"); + Map<String, String> headers = new HashMap<String, String>(); + + headers.put("Authorization", "Bearer " + jwtToken_); + headers.put("X-Forwarded-For", "127.0.0.1"); + transport.setCustomHeaders(headers); + transport.open(); + TCLIService.Iface client = new TCLIService.Client(new TBinaryProtocol(transport)); + + TOpenSessionReq openReq = new TOpenSessionReq(); + try { + client.OpenSession(openReq); + fail("Exception expected."); + } catch (Exception e) { + verifyJwtAuthMetrics(0, 1); + assertEquals(e.getMessage(), "HTTP Response code: 401"); + verifyBearerErrorContainsClaim(transport, "iss"); + } + } + + /** + * Tests JWT authentication fails when audience validation is configured and the token + * has no audience claim. + */ + @Test + public void testJwtAuthWithMissingAudienceClaim() throws Exception { + createJWKSForWebServer_ = false; + String jwksFilename = + new File(System.getenv("IMPALA_HOME"), + String.format("testdata/jwt/%s", JWKS_FILE_NAME)).getPath(); + String oauthServers = String.format( + "[{\"jwksFilePath\":\"%s\",\"audienceClaims\":[\"impala\"]}]", + jwksFilename); + setUp(String.format( + "--jwt_token_auth=true --oauth_servers='%s' --jwt_allow_without_tls=true", + oauthServers)); + THttpClientWithHeaders transport = + new THttpClientWithHeaders("http://localhost:28000"); + Map<String, String> headers = new HashMap<String, String>(); + + headers.put("Authorization", "Bearer " + jwtToken_); + headers.put("X-Forwarded-For", "127.0.0.1"); + transport.setCustomHeaders(headers); + transport.open(); + TCLIService.Iface client = new TCLIService.Client(new TBinaryProtocol(transport)); + + TOpenSessionReq openReq = new TOpenSessionReq(); + try { + client.OpenSession(openReq); + fail("Exception expected."); + } catch (Exception e) { + verifyJwtAuthMetrics(0, 1); + assertEquals(e.getMessage(), "HTTP Response code: 401"); + verifyBearerErrorContainsClaim(transport, "aud"); + } + } + + /** + * Tests JWT authentication fails when audience validation is configured and + * the token audience does not match. + */ + @Test + public void testJwtAuthWithWrongAudienceClaim() throws Exception { + createJWKSForWebServer_ = false; + String jwksFilename = + new File(System.getenv("IMPALA_HOME"), + "testdata/jwt/jwks_signing.json").getPath(); + String tokenWithAudience = readTrimmedFile(Paths.get( + System.getenv("IMPALA_HOME"), "testdata", "jwt", "jwt_signed")); + String oauthServers = String.format( + "[{\"jwksFilePath\":\"%s\",\"usernameClaim\":\"sub\"," + + "\"audienceClaims\":[\"wrong-audience\"]}]", + jwksFilename); + setUp(String.format( + "--jwt_token_auth=true --oauth_servers='%s' --jwt_allow_without_tls=true", + oauthServers)); + THttpClientWithHeaders transport = + new THttpClientWithHeaders("http://localhost:28000"); + Map<String, String> headers = new HashMap<String, String>(); + + headers.put("Authorization", "Bearer " + tokenWithAudience); + headers.put("X-Forwarded-For", "127.0.0.1"); + transport.setCustomHeaders(headers); + transport.open(); + TCLIService.Iface client = new TCLIService.Client(new TBinaryProtocol(transport)); + + TOpenSessionReq openReq = new TOpenSessionReq(); + try { + client.OpenSession(openReq); + fail("Exception expected."); + } catch (Exception e) { + verifyJwtAuthMetrics(0, 1); + assertEquals(e.getMessage(), "HTTP Response code: 401"); + verifyBearerErrorContainsClaim(transport, "aud"); + } + } + + /** * Tests if sessions are authenticated by verifying the JWT token for connections * to the HTTP hiveserver2 endpoint. */
diff --git a/fe/src/test/java/org/apache/impala/customcluster/THttpClientWithHeaders.java b/fe/src/test/java/org/apache/impala/customcluster/THttpClientWithHeaders.java index 7f429bb..e702915 100644 --- a/fe/src/test/java/org/apache/impala/customcluster/THttpClientWithHeaders.java +++ b/fe/src/test/java/org/apache/impala/customcluster/THttpClientWithHeaders.java
@@ -32,10 +32,12 @@ import java.net.URL; import java.net.HttpURLConnection; +import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; +import org.apache.http.Header; import org.apache.http.HttpEntity; import org.apache.http.HttpHost; import org.apache.http.HttpResponse; @@ -264,6 +266,15 @@ } } + private static Map<String, List<String>> captureResponseHeaders(HttpResponse response) { + Map<String, List<String>> headers = new HashMap<>(); + for (Header header : response.getAllHeaders()) { + headers.computeIfAbsent(header.getName(), key -> new ArrayList<>()) + .add(header.getValue()); + } + return headers; + } + private void flushUsingHttpClient() throws TTransportException { if (null == this.client) { @@ -301,6 +312,7 @@ HttpResponse response = this.client.execute(this.host, post); int responseCode = response.getStatusLine().getStatusCode(); + responseHeaders_ = captureResponseHeaders(response); // // Retrieve the inputstream BEFORE checking the status code so @@ -400,15 +412,13 @@ connection.getOutputStream().write(data); int responseCode = connection.getResponseCode(); + responseHeaders_ = connection.getHeaderFields(); if (responseCode != HttpURLConnection.HTTP_OK) { throw new TTransportException("HTTP Response code: " + responseCode); } // Read the responses inputStream_ = connection.getInputStream(); - // Capture the response headers. - // This is not in the THttpClient.java class in the apache/thrift repository. - responseHeaders_ = connection.getHeaderFields(); } catch (IOException iox) { throw new TTransportException(iox);