common/cpp/obj-client/auth/gcp/gcp_s3_client.cpp

Source
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you 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
//
//   http://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.

#include "cpp/obj-client/auth/gcp/gcp_s3_client.h"

#include <aws/core/auth/AWSCredentialsProvider.h>
#include <aws/s3/model/AbortMultipartUploadRequest.h>
#include <aws/s3/model/AbortMultipartUploadResult.h>
#include <aws/s3/model/CompleteMultipartUploadRequest.h>
#include <aws/s3/model/CompleteMultipartUploadResult.h>
#include <aws/s3/model/CreateMultipartUploadRequest.h>
#include <aws/s3/model/CreateMultipartUploadResult.h>
#include <aws/s3/model/DeleteObjectRequest.h>
#include <aws/s3/model/DeleteObjectResult.h>
#include <aws/s3/model/DeleteObjectsRequest.h>
#include <aws/s3/model/DeleteObjectsResult.h>
#include <aws/s3/model/GetBucketLifecycleConfigurationRequest.h>
#include <aws/s3/model/GetBucketLifecycleConfigurationResult.h>
#include <aws/s3/model/GetBucketVersioningRequest.h>
#include <aws/s3/model/GetBucketVersioningResult.h>
#include <aws/s3/model/GetObjectRequest.h>
#include <aws/s3/model/GetObjectResult.h>
#include <aws/s3/model/HeadObjectRequest.h>
#include <aws/s3/model/HeadObjectResult.h>
#include <aws/s3/model/ListObjectsV2Request.h>
#include <aws/s3/model/ListObjectsV2Result.h>
#include <aws/s3/model/PutObjectRequest.h>
#include <aws/s3/model/PutObjectResult.h>
#include <aws/s3/model/UploadPartRequest.h>
#include <aws/s3/model/UploadPartResult.h>

#include <algorithm>
#include <utility>

namespace doris {
namespace {

Aws::Client::ClientConfiguration gcp_http_config(Aws::Client::ClientConfiguration config) {
    // Do not forward bearer headers on a redirect or allow TLS verification to be disabled.
    config.followRedirects = Aws::Client::FollowRedirectsPolicy::NEVER;
    config.verifySSL = true;
    config.scheme = Aws::Http::Scheme::HTTPS;
    return config;
}

bool trusted_endpoint(const Aws::Client::ClientConfiguration& config) {
    std::string endpoint(config.endpointOverride.c_str());
    if (endpoint.find("://") == std::string::npos) {
        endpoint = "https://" + endpoint;
    }
    return is_valid_gcp_storage_endpoint(endpoint);
}

} // namespace

GcpS3Client::GcpS3Client(const GcpCredentialConfig& credential, std::string ca_cert_path,
                         Aws::Client::ClientConfiguration aws_config,
                         Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy payload_signing_policy,
                         bool use_virtual_addressing)
        : Aws::S3::S3Client(std::make_shared<Aws::Auth::AnonymousAWSCredentialsProvider>(),
                            gcp_http_config(aws_config), payload_signing_policy,
                            use_virtual_addressing),
          _trusted_endpoint(trusted_endpoint(aws_config)),
          _token_provider(credential, ca_cert_path) {}

std::optional<Aws::S3::S3Error> GcpS3Client::authorize_request(
        Aws::AmazonWebServiceRequest& request, std::string_view bucket) const {
    // GCS bucket names cannot contain ARN, URL or authority delimiters. In particular,
    // do not let the AWS endpoint resolver route an access-point ARN to an AWS host.
    bool valid_bucket = std::all_of(bucket.begin(), bucket.end(), [](unsigned char c) {
        return (c >= 'a' && c <= 'z') || (c >= '0' && c <= '9') || c == '-' || c == '_' || c == '.';
    });
    if (!_trusted_endpoint || !valid_bucket) {
        Aws::Client::AWSError<Aws::S3::S3Errors> error(
                Aws::S3::S3Errors::ACCESS_DENIED, "GcpEndpointError",
                "Native GCP OAuth requires a trusted Google Cloud Storage HTTPS "
                "endpoint on port 443 and a GCS bucket name",
                false);
        error.SetResponseCode(Aws::Http::HttpResponseCode::FORBIDDEN);
        return Aws::S3::S3Error(std::move(error));
    }
    auto token = fetch_token();
    if (!token.has_value() || token->empty()) {
        // Return through the SDK outcome path before any object-storage HTTP request.
        // The token provider has already logged the underlying credential error.
        Aws::Client::AWSError<Aws::S3::S3Errors> error(Aws::S3::S3Errors::ACCESS_DENIED,
                                                       "GcpAuthenticationError",
                                                       "Failed to obtain GCP access token", false);
        error.SetResponseCode(Aws::Http::HttpResponseCode::UNAUTHORIZED);
        return Aws::S3::S3Error(std::move(error));
    }
    request.SetAdditionalCustomHeaderValue("Authorization",
                                           Aws::String("Bearer ") + token->c_str());
    return std::nullopt;
}

Aws::S3::Model::AbortMultipartUploadOutcome GcpS3Client::AbortMultipartUpload(
        const Aws::S3::Model::AbortMultipartUploadRequest& request) const {
    auto authorized_request = request;
    if (auto error = authorize_request(authorized_request, request.GetBucket())) {
        return std::move(*error);
    }
    return Aws::S3::S3Client::AbortMultipartUpload(authorized_request);
}

Aws::S3::Model::CompleteMultipartUploadOutcome GcpS3Client::CompleteMultipartUpload(
        const Aws::S3::Model::CompleteMultipartUploadRequest& request) const {
    auto authorized_request = request;
    if (auto error = authorize_request(authorized_request, request.GetBucket())) {
        return std::move(*error);
    }
    return Aws::S3::S3Client::CompleteMultipartUpload(authorized_request);
}

Aws::S3::Model::CreateMultipartUploadOutcome GcpS3Client::CreateMultipartUpload(
        const Aws::S3::Model::CreateMultipartUploadRequest& request) const {
    auto authorized_request = request;
    if (auto error = authorize_request(authorized_request, request.GetBucket())) {
        return std::move(*error);
    }
    return Aws::S3::S3Client::CreateMultipartUpload(authorized_request);
}

Aws::S3::Model::DeleteObjectOutcome GcpS3Client::DeleteObject(
        const Aws::S3::Model::DeleteObjectRequest& request) const {
    auto authorized_request = request;
    if (auto error = authorize_request(authorized_request, request.GetBucket())) {
        return std::move(*error);
    }
    return Aws::S3::S3Client::DeleteObject(authorized_request);
}

Aws::S3::Model::DeleteObjectsOutcome GcpS3Client::DeleteObjects(
        const Aws::S3::Model::DeleteObjectsRequest& request) const {
    auto authorized_request = request;
    if (auto error = authorize_request(authorized_request, request.GetBucket())) {
        return std::move(*error);
    }
    return Aws::S3::S3Client::DeleteObjects(authorized_request);
}

Aws::S3::Model::GetBucketLifecycleConfigurationOutcome GcpS3Client::GetBucketLifecycleConfiguration(
        const Aws::S3::Model::GetBucketLifecycleConfigurationRequest& request) const {
    auto authorized_request = request;
    if (auto error = authorize_request(authorized_request, request.GetBucket())) {
        return std::move(*error);
    }
    return Aws::S3::S3Client::GetBucketLifecycleConfiguration(authorized_request);
}

Aws::S3::Model::GetBucketVersioningOutcome GcpS3Client::GetBucketVersioning(
        const Aws::S3::Model::GetBucketVersioningRequest& request) const {
    auto authorized_request = request;
    if (auto error = authorize_request(authorized_request, request.GetBucket())) {
        return std::move(*error);
    }
    return Aws::S3::S3Client::GetBucketVersioning(authorized_request);
}

Aws::S3::Model::GetObjectOutcome GcpS3Client::GetObject(
        const Aws::S3::Model::GetObjectRequest& request) const {
    auto authorized_request = request;
    if (auto error = authorize_request(authorized_request, request.GetBucket())) {
        return std::move(*error);
    }
    return Aws::S3::S3Client::GetObject(authorized_request);
}

Aws::S3::Model::HeadObjectOutcome GcpS3Client::HeadObject(
        const Aws::S3::Model::HeadObjectRequest& request) const {
    auto authorized_request = request;
    if (auto error = authorize_request(authorized_request, request.GetBucket())) {
        return std::move(*error);
    }
    return Aws::S3::S3Client::HeadObject(authorized_request);
}

Aws::S3::Model::ListObjectsV2Outcome GcpS3Client::ListObjectsV2(
        const Aws::S3::Model::ListObjectsV2Request& request) const {
    auto authorized_request = request;
    if (auto error = authorize_request(authorized_request, request.GetBucket())) {
        return std::move(*error);
    }
    return Aws::S3::S3Client::ListObjectsV2(authorized_request);
}

Aws::S3::Model::PutObjectOutcome GcpS3Client::PutObject(
        const Aws::S3::Model::PutObjectRequest& request) const {
    auto authorized_request = request;
    if (auto error = authorize_request(authorized_request, request.GetBucket())) {
        return std::move(*error);
    }
    return Aws::S3::S3Client::PutObject(authorized_request);
}

Aws::S3::Model::UploadPartOutcome GcpS3Client::UploadPart(
        const Aws::S3::Model::UploadPartRequest& request) const {
    auto authorized_request = request;
    if (auto error = authorize_request(authorized_request, request.GetBucket())) {
        return std::move(*error);
    }
    return Aws::S3::S3Client::UploadPart(authorized_request);
}

std::optional<std::string> GcpS3Client::fetch_token() const {
    return _token_provider.get_token();
}

} // namespace doris

Coverage Report

Created: 2026-10-26 16:27

Line	Count	Source
1		// Licensed to the Apache Software Foundation (ASF) under one
2		// or more contributor license agreements. See the NOTICE file
3		// distributed with this work for additional information
4		// regarding copyright ownership. The ASF licenses this file
5		// to you under the Apache License, Version 2.0 (the
6		// "License"); you may not use this file except in compliance
7		// with the License. You may obtain a copy of the License at
8		//
9		// http://www.apache.org/licenses/LICENSE-2.0
10		//
11		// Unless required by applicable law or agreed to in writing,
12		// software distributed under the License is distributed on an
13		// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14		// KIND, either express or implied. See the License for the
15		// specific language governing permissions and limitations
16		// under the License.
17
18		#include "cpp/obj-client/auth/gcp/gcp_s3_client.h"
19
20		#include <aws/core/auth/AWSCredentialsProvider.h>
21		#include <aws/s3/model/AbortMultipartUploadRequest.h>
22		#include <aws/s3/model/AbortMultipartUploadResult.h>
23		#include <aws/s3/model/CompleteMultipartUploadRequest.h>
24		#include <aws/s3/model/CompleteMultipartUploadResult.h>
25		#include <aws/s3/model/CreateMultipartUploadRequest.h>
26		#include <aws/s3/model/CreateMultipartUploadResult.h>
27		#include <aws/s3/model/DeleteObjectRequest.h>
28		#include <aws/s3/model/DeleteObjectResult.h>
29		#include <aws/s3/model/DeleteObjectsRequest.h>
30		#include <aws/s3/model/DeleteObjectsResult.h>
31		#include <aws/s3/model/GetBucketLifecycleConfigurationRequest.h>
32		#include <aws/s3/model/GetBucketLifecycleConfigurationResult.h>
33		#include <aws/s3/model/GetBucketVersioningRequest.h>
34		#include <aws/s3/model/GetBucketVersioningResult.h>
35		#include <aws/s3/model/GetObjectRequest.h>
36		#include <aws/s3/model/GetObjectResult.h>
37		#include <aws/s3/model/HeadObjectRequest.h>
38		#include <aws/s3/model/HeadObjectResult.h>
39		#include <aws/s3/model/ListObjectsV2Request.h>
40		#include <aws/s3/model/ListObjectsV2Result.h>
41		#include <aws/s3/model/PutObjectRequest.h>
42		#include <aws/s3/model/PutObjectResult.h>
43		#include <aws/s3/model/UploadPartRequest.h>
44		#include <aws/s3/model/UploadPartResult.h>
45
46		#include <algorithm>
47		#include <utility>
48
49		namespace doris {
50		namespace {
51
52	12	Aws::Client::ClientConfiguration gcp_http_config(Aws::Client::ClientConfiguration config) {
53		// Do not forward bearer headers on a redirect or allow TLS verification to be disabled.
54	12	config.followRedirects = Aws::Client::FollowRedirectsPolicy::NEVER;
55	12	config.verifySSL = true;
56	12	config.scheme = Aws::Http::Scheme::HTTPS;
57	12	return config;
58	12	}
59
60	12	bool trusted_endpoint(const Aws::Client::ClientConfiguration& config) {
61	12	std::string endpoint(config.endpointOverride.c_str());
62	12	if (endpoint.find("://") == std::string::npos) {
63	1	endpoint = "https://" + endpoint;
64	1	}
65	12	return is_valid_gcp_storage_endpoint(endpoint);
66	12	}
67
68		} // namespace
69
70		GcpS3Client::GcpS3Client(const GcpCredentialConfig& credential, std::string ca_cert_path,
71		Aws::Client::ClientConfiguration aws_config,
72		Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy payload_signing_policy,
73		bool use_virtual_addressing)
74	12	: Aws::S3::S3Client(std::make_shared<Aws::Auth::AnonymousAWSCredentialsProvider>(),
75	12	gcp_http_config(aws_config), payload_signing_policy,
76	12	use_virtual_addressing),
77	12	_trusted_endpoint(trusted_endpoint(aws_config)),
78	12	_token_provider(credential, ca_cert_path) {}
79
80		std::optional<Aws::S3::S3Error> GcpS3Client::authorize_request(
81	35	Aws::AmazonWebServiceRequest& request, std::string_view bucket) const {
82		// GCS bucket names cannot contain ARN, URL or authority delimiters. In particular,
83		// do not let the AWS endpoint resolver route an access-point ARN to an AWS host.
84	64	bool valid_bucket = std::all_of(bucket.begin(), bucket.end(), [](unsigned char c) {
85	64	return (c >= 'a' && c <= 'z') \|\| (c >= '0' && c <= '9') \|\| c == '-' \|\| c == '_' \|\| c == '.';
86	64	});
87	35	if (!_trusted_endpoint \|\| !valid_bucket) {
88	10	Aws::Client::AWSError<Aws::S3::S3Errors> error(
89	10	Aws::S3::S3Errors::ACCESS_DENIED, "GcpEndpointError",
90	10	"Native GCP OAuth requires a trusted Google Cloud Storage HTTPS "
91	10	"endpoint on port 443 and a GCS bucket name",
92	10	false);
93	10	error.SetResponseCode(Aws::Http::HttpResponseCode::FORBIDDEN);
94	10	return Aws::S3::S3Error(std::move(error));
95	10	}
96	25	auto token = fetch_token();
97	25	if (!token.has_value() \|\| token->empty()) {
98		// Return through the SDK outcome path before any object-storage HTTP request.
99		// The token provider has already logged the underlying credential error.
100	24	Aws::Client::AWSError<Aws::S3::S3Errors> error(Aws::S3::S3Errors::ACCESS_DENIED,
101	24	"GcpAuthenticationError",
102	24	"Failed to obtain GCP access token", false);
103	24	error.SetResponseCode(Aws::Http::HttpResponseCode::UNAUTHORIZED);
104	24	return Aws::S3::S3Error(std::move(error));
105	24	}
106	1	request.SetAdditionalCustomHeaderValue("Authorization",
107	1	Aws::String("Bearer ") + token->c_str());
108	1	return std::nullopt;
109	25	}
110
111		Aws::S3::Model::AbortMultipartUploadOutcome GcpS3Client::AbortMultipartUpload(
112	2	const Aws::S3::Model::AbortMultipartUploadRequest& request) const {
113	2	auto authorized_request = request;
114	2	if (auto error = authorize_request(authorized_request, request.GetBucket())) {
115	2	return std::move(*error);
116	2	}
117	0	return Aws::S3::S3Client::AbortMultipartUpload(authorized_request);
118	2	}
119
120		Aws::S3::Model::CompleteMultipartUploadOutcome GcpS3Client::CompleteMultipartUpload(
121	2	const Aws::S3::Model::CompleteMultipartUploadRequest& request) const {
122	2	auto authorized_request = request;
123	2	if (auto error = authorize_request(authorized_request, request.GetBucket())) {
124	2	return std::move(*error);
125	2	}
126	0	return Aws::S3::S3Client::CompleteMultipartUpload(authorized_request);
127	2	}
128
129		Aws::S3::Model::CreateMultipartUploadOutcome GcpS3Client::CreateMultipartUpload(
130	2	const Aws::S3::Model::CreateMultipartUploadRequest& request) const {
131	2	auto authorized_request = request;
132	2	if (auto error = authorize_request(authorized_request, request.GetBucket())) {
133	2	return std::move(*error);
134	2	}
135	0	return Aws::S3::S3Client::CreateMultipartUpload(authorized_request);
136	2	}
137
138		Aws::S3::Model::DeleteObjectOutcome GcpS3Client::DeleteObject(
139	2	const Aws::S3::Model::DeleteObjectRequest& request) const {
140	2	auto authorized_request = request;
141	2	if (auto error = authorize_request(authorized_request, request.GetBucket())) {
142	2	return std::move(*error);
143	2	}
144	0	return Aws::S3::S3Client::DeleteObject(authorized_request);
145	2	}
146
147		Aws::S3::Model::DeleteObjectsOutcome GcpS3Client::DeleteObjects(
148	2	const Aws::S3::Model::DeleteObjectsRequest& request) const {
149	2	auto authorized_request = request;
150	2	if (auto error = authorize_request(authorized_request, request.GetBucket())) {
151	2	return std::move(*error);
152	2	}
153	0	return Aws::S3::S3Client::DeleteObjects(authorized_request);
154	2	}
155
156		Aws::S3::Model::GetBucketLifecycleConfigurationOutcome GcpS3Client::GetBucketLifecycleConfiguration(
157	2	const Aws::S3::Model::GetBucketLifecycleConfigurationRequest& request) const {
158	2	auto authorized_request = request;
159	2	if (auto error = authorize_request(authorized_request, request.GetBucket())) {
160	2	return std::move(*error);
161	2	}
162	0	return Aws::S3::S3Client::GetBucketLifecycleConfiguration(authorized_request);
163	2	}
164
165		Aws::S3::Model::GetBucketVersioningOutcome GcpS3Client::GetBucketVersioning(
166	2	const Aws::S3::Model::GetBucketVersioningRequest& request) const {
167	2	auto authorized_request = request;
168	2	if (auto error = authorize_request(authorized_request, request.GetBucket())) {
169	2	return std::move(*error);
170	2	}
171	0	return Aws::S3::S3Client::GetBucketVersioning(authorized_request);
172	2	}
173
174		Aws::S3::Model::GetObjectOutcome GcpS3Client::GetObject(
175	12	const Aws::S3::Model::GetObjectRequest& request) const {
176	12	auto authorized_request = request;
177	12	if (auto error = authorize_request(authorized_request, request.GetBucket())) {
178	12	return std::move(*error);
179	12	}
180	0	return Aws::S3::S3Client::GetObject(authorized_request);
181	12	}
182
183		Aws::S3::Model::HeadObjectOutcome GcpS3Client::HeadObject(
184	2	const Aws::S3::Model::HeadObjectRequest& request) const {
185	2	auto authorized_request = request;
186	2	if (auto error = authorize_request(authorized_request, request.GetBucket())) {
187	2	return std::move(*error);
188	2	}
189	0	return Aws::S3::S3Client::HeadObject(authorized_request);
190	2	}
191
192		Aws::S3::Model::ListObjectsV2Outcome GcpS3Client::ListObjectsV2(
193	2	const Aws::S3::Model::ListObjectsV2Request& request) const {
194	2	auto authorized_request = request;
195	2	if (auto error = authorize_request(authorized_request, request.GetBucket())) {
196	2	return std::move(*error);
197	2	}
198	0	return Aws::S3::S3Client::ListObjectsV2(authorized_request);
199	2	}
200
201		Aws::S3::Model::PutObjectOutcome GcpS3Client::PutObject(
202	2	const Aws::S3::Model::PutObjectRequest& request) const {
203	2	auto authorized_request = request;
204	2	if (auto error = authorize_request(authorized_request, request.GetBucket())) {
205	2	return std::move(*error);
206	2	}
207	0	return Aws::S3::S3Client::PutObject(authorized_request);
208	2	}
209
210		Aws::S3::Model::UploadPartOutcome GcpS3Client::UploadPart(
211	2	const Aws::S3::Model::UploadPartRequest& request) const {
212	2	auto authorized_request = request;
213	2	if (auto error = authorize_request(authorized_request, request.GetBucket())) {
214	2	return std::move(*error);
215	2	}
216	0	return Aws::S3::S3Client::UploadPart(authorized_request);
217	2	}
218
219	0	std::optional<std::string> GcpS3Client::fetch_token() const {
220	0	return _token_provider.get_token();
221	0	}
222
223		} // namespace doris