[oauth2 cpp] add support to read credentials from file (#7606)

### Motivation

Add support to read credentials from file, make it align with java client. 

### Modifications

- Add support to read credentials from file, make it align with java client. 
- add a test for it.
Co-authored-by: Nxiaolong.ran <rxl@apache.org>

pulsar-broker/src/test/resources/authentication/token/cpp_credentials_file.json

+{
+  "client_id":"Xd23RHsUnvUlP7wchjNYOaIfazgeHd9x",
+  "client_secret":"rT7ps7WY8uhdVuBTKWZkttwLdQotmdEliaM5rLfmgNibvqziZ-g07ZH52N_poGAb"
+}

pulsar-client-cpp/lib/auth/AuthOauth2.cc

@@ -119,6 +119,12 @@ Oauth2Flow::Oauth2Flow() {}
 Oauth2Flow::~Oauth2Flow() {}
 
 // ClientCredentialFlow
+static std::string readFromFile(const std::string& credentialsFilePath) {
+    std::ifstream input(credentialsFilePath);
+    std::stringstream buffer;
+    buffer << input.rdbuf();
+    return buffer.str();
+}
 
 ClientCredentialFlow::ClientCredentialFlow(const std::string& issuerUrl, const std::string& clientId,
                                            const std::string& clientSecret, const std::string& audience) {
@@ -128,6 +134,33 @@ ClientCredentialFlow::ClientCredentialFlow(const std::string& issuerUrl, const s
     audience_ = audience;
 }
 
+// read clientId/clientSecret from passed in `credentialsFilePath`
+ClientCredentialFlow::ClientCredentialFlow(const std::string& issuerUrl,
+                                           const std::string& credentialsFilePath,
+                                           const std::string& audience) {
+    issuerUrl_ = issuerUrl;
+    audience_ = audience;
+
+    boost::property_tree::ptree loadPtreeRoot;
+    try {
+        boost::property_tree::read_json(credentialsFilePath, loadPtreeRoot);
+    } catch (boost::property_tree::json_parser_error& e) {
+        LOG_ERROR("Failed to parse json input file for credentialsFilePath: " << credentialsFilePath
+                                                                              << "with error:" << e.what());
+        return;
+    }
+
+    const std::string defaultNotFoundString = "Client Id / Secret Not Found";
+
+    clientId_ = loadPtreeRoot.get<std::string>("client_id", defaultNotFoundString);
+    clientSecret_ = loadPtreeRoot.get<std::string>("client_secret", defaultNotFoundString);
+
+    if (clientId_ == defaultNotFoundString || clientSecret_ == defaultNotFoundString) {
+        LOG_ERROR("Not get valid clientId / clientSecret: " << clientId_ << "/" << clientSecret_);
+        return;
+    }
+}
+
 void ClientCredentialFlow::initialize() {}
 void ClientCredentialFlow::close() {}
 
@@ -225,8 +258,16 @@ Oauth2TokenResultPtr ClientCredentialFlow::authenticate() {
 // AuthOauth2
 
 AuthOauth2::AuthOauth2(ParamMap& params) {
-    flowPtr_ = FlowPtr(new ClientCredentialFlow(params["issuer_url"], params["client_id"],
-                                                params["client_secret"], params["audience"]));
+    std::map<std::string, std::string>::iterator it;
+    it = params.find("private_key");
+
+    if (it != params.end()) {
+        flowPtr_ = FlowPtr(
+            new ClientCredentialFlow(params["issuer_url"], params["private_key"], params["audience"]));
+    } else {
+        flowPtr_ = FlowPtr(new ClientCredentialFlow(params["issuer_url"], params["client_id"],
+                                                    params["client_secret"], params["audience"]));
+    }
 }
 
 AuthOauth2::~AuthOauth2() {}

pulsar-client-cpp/lib/auth/AuthOauth2.h

@@ -33,6 +33,8 @@ class ClientCredentialFlow : public Oauth2Flow {
    public:
     ClientCredentialFlow(const std::string& issuerUrl, const std::string& clientId,
                          const std::string& clientSecret, const std::string& audience);
+    ClientCredentialFlow(const std::string& issuerUrl, const std::string& credentialsFilePath,
+                         const std::string& audience);
     void initialize();
     Oauth2TokenResultPtr authenticate();
     void close();

pulsar-client-cpp/tests/AuthPluginTest.cc

@@ -381,3 +381,22 @@ TEST(AuthPluginTest, testOauth2WrongSecret) {
         // expected
     }
 }
+
+TEST(AuthPluginTest, testOauth2CredentialFile) {
+    // test success get token from oauth2 server.
+    pulsar::AuthenticationDataPtr data;
+    std::string params = R"({
+        "type": "client_credentials",
+        "issuer_url": "https://dev-kt-aa9ne.us.auth0.com/oauth/token",
+        "private_key": "../../pulsar-broker/src/test/resources/authentication/token/cpp_credentials_file.json",
+        "audience": "https://dev-kt-aa9ne.us.auth0.com/api/v2/"})";
+
+    int expectedTokenLength = 3379;
+    LOG_INFO("PARAMS: " << params);
+    pulsar::AuthenticationPtr auth = pulsar::AuthOauth2::create(params);
+    ASSERT_EQ(auth->getAuthMethodName(), "token");
+    ASSERT_EQ(auth->getAuthData(data), pulsar::ResultOk);
+    ASSERT_EQ(data->hasDataForHttp(), true);
+    ASSERT_EQ(data->hasDataFromCommand(), true);
+    ASSERT_EQ(data->getCommandData().length(), expectedTokenLength);
+}