484 lines
17 KiB
C++
484 lines
17 KiB
C++
#include "minio_uploader.h"
|
|
|
|
#include <algorithm>
|
|
#include <chrono>
|
|
#include <cstdlib>
|
|
#include <cstring>
|
|
#include <ctime>
|
|
#include <iomanip>
|
|
#include <iostream>
|
|
#include <mutex>
|
|
#include <sstream>
|
|
#include <thread>
|
|
#include <vector>
|
|
|
|
#include "utils/simple_json_writer.h"
|
|
|
|
#if __has_include(<curl/curl.h>)
|
|
#include <curl/curl.h>
|
|
#define HAS_CURL 1
|
|
#else
|
|
#define HAS_CURL 0
|
|
#endif
|
|
|
|
#if defined(RK3588_ENABLE_OPENSSL) && __has_include(<openssl/evp.h>) && __has_include(<openssl/hmac.h>) && __has_include(<openssl/sha.h>)
|
|
#include <openssl/evp.h>
|
|
#include <openssl/hmac.h>
|
|
#include <openssl/sha.h>
|
|
#define HAS_OPENSSL 1
|
|
#else
|
|
#define HAS_OPENSSL 0
|
|
#endif
|
|
|
|
namespace rk3588 {
|
|
|
|
namespace {
|
|
|
|
std::string ResolveEnv(const std::string& env_name, const std::string& val) {
|
|
// Support explicit "${ENV_VAR}" syntax.
|
|
if (val.size() > 3 && val.substr(0, 2) == "${" && val.back() == '}') {
|
|
std::string var_name = val.substr(2, val.size() - 3);
|
|
const char* v = std::getenv(var_name.c_str());
|
|
return v ? std::string(v) : "";
|
|
}
|
|
// Default behavior: if empty, fall back to env_name.
|
|
if (val.empty()) {
|
|
const char* v = std::getenv(env_name.c_str());
|
|
return v ? std::string(v) : "";
|
|
}
|
|
return val;
|
|
}
|
|
|
|
#if HAS_CURL
|
|
size_t CurlWriteCb(char* ptr, size_t size, size_t nmemb, void* userdata) {
|
|
if (!userdata || !ptr) return 0;
|
|
auto* out = static_cast<std::string*>(userdata);
|
|
out->append(ptr, size * nmemb);
|
|
return size * nmemb;
|
|
}
|
|
|
|
struct PresignResult {
|
|
std::string upload_url;
|
|
std::string public_url;
|
|
std::vector<std::string> headers; // "Key: Value"
|
|
};
|
|
|
|
bool GetPresignedPutUrl(const std::string& endpoint,
|
|
const std::string& bucket,
|
|
const std::string& key,
|
|
const std::string& content_type,
|
|
size_t size,
|
|
int timeout_ms,
|
|
PresignResult& out,
|
|
std::string& err) {
|
|
err.clear();
|
|
out = PresignResult{};
|
|
|
|
SimpleJson::Object req;
|
|
req["bucket"] = SimpleJson(bucket);
|
|
req["key"] = SimpleJson(key);
|
|
req["content_type"] = SimpleJson(content_type);
|
|
req["size"] = SimpleJson(static_cast<double>(size));
|
|
req["method"] = SimpleJson(std::string("PUT"));
|
|
const std::string body = StringifySimpleJson(SimpleJson(std::move(req)));
|
|
|
|
CURL* curl = curl_easy_init();
|
|
if (!curl) {
|
|
err = "curl_easy_init failed";
|
|
return false;
|
|
}
|
|
|
|
std::string resp;
|
|
struct curl_slist* headers = nullptr;
|
|
headers = curl_slist_append(headers, "Content-Type: application/json");
|
|
|
|
curl_easy_setopt(curl, CURLOPT_URL, endpoint.c_str());
|
|
curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers);
|
|
curl_easy_setopt(curl, CURLOPT_POSTFIELDS, body.c_str());
|
|
curl_easy_setopt(curl, CURLOPT_POSTFIELDSIZE, static_cast<long>(body.size()));
|
|
curl_easy_setopt(curl, CURLOPT_TIMEOUT_MS, timeout_ms);
|
|
curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT_MS, std::max(500, timeout_ms / 2));
|
|
curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, &CurlWriteCb);
|
|
curl_easy_setopt(curl, CURLOPT_WRITEDATA, &resp);
|
|
|
|
CURLcode res = curl_easy_perform(curl);
|
|
long http_code = 0;
|
|
curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &http_code);
|
|
curl_slist_free_all(headers);
|
|
curl_easy_cleanup(curl);
|
|
|
|
if (res != CURLE_OK) {
|
|
err = curl_easy_strerror(res);
|
|
return false;
|
|
}
|
|
if (http_code < 200 || http_code >= 300) {
|
|
err = "presign http " + std::to_string(http_code);
|
|
return false;
|
|
}
|
|
|
|
SimpleJson j;
|
|
std::string jerr;
|
|
if (!ParseSimpleJson(resp, j, jerr) || !j.IsObject()) {
|
|
err = "presign response json invalid: " + jerr;
|
|
return false;
|
|
}
|
|
|
|
const SimpleJson* url = j.Find("upload_url");
|
|
if (!url) url = j.Find("url");
|
|
if (!url || !url->IsString()) {
|
|
err = "presign response missing upload_url";
|
|
return false;
|
|
}
|
|
out.upload_url = url->AsString("");
|
|
if (const SimpleJson* pu = j.Find("public_url"); pu && pu->IsString()) {
|
|
out.public_url = pu->AsString("");
|
|
}
|
|
if (const SimpleJson* hs = j.Find("headers"); hs && hs->IsObject()) {
|
|
for (const auto& kv : hs->AsObject()) {
|
|
if (!kv.second.IsString()) continue;
|
|
out.headers.push_back(kv.first + ": " + kv.second.AsString(""));
|
|
}
|
|
}
|
|
return !out.upload_url.empty();
|
|
}
|
|
#endif
|
|
|
|
#if HAS_CURL && HAS_OPENSSL
|
|
inline bool IsUnreserved(unsigned char c) {
|
|
return (c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') || (c >= '0' && c <= '9') ||
|
|
c == '-' || c == '_' || c == '.' || c == '~';
|
|
}
|
|
|
|
std::string UriEncodePath(const std::string& s) {
|
|
std::ostringstream oss;
|
|
oss << std::uppercase << std::hex;
|
|
for (unsigned char c : s) {
|
|
if (IsUnreserved(c) || c == '/') {
|
|
oss << static_cast<char>(c);
|
|
} else {
|
|
oss << '%' << std::setw(2) << std::setfill('0') << static_cast<int>(c);
|
|
}
|
|
}
|
|
return oss.str();
|
|
}
|
|
|
|
std::string ToHexLower(const unsigned char* data, size_t len) {
|
|
std::ostringstream oss;
|
|
oss << std::hex << std::setfill('0');
|
|
for (size_t i = 0; i < len; ++i) {
|
|
oss << std::setw(2) << static_cast<int>(data[i]);
|
|
}
|
|
return oss.str();
|
|
}
|
|
|
|
std::string Sha256Hex(const void* data, size_t len) {
|
|
unsigned char hash[SHA256_DIGEST_LENGTH];
|
|
SHA256(static_cast<const unsigned char*>(data), len, hash);
|
|
return ToHexLower(hash, SHA256_DIGEST_LENGTH);
|
|
}
|
|
|
|
std::vector<unsigned char> HmacSha256(const unsigned char* key, size_t key_len, const std::string& msg) {
|
|
unsigned int out_len = 0;
|
|
unsigned char out[EVP_MAX_MD_SIZE];
|
|
if (!HMAC(EVP_sha256(), key, static_cast<int>(key_len),
|
|
reinterpret_cast<const unsigned char*>(msg.data()), msg.size(),
|
|
out, &out_len)) {
|
|
return {};
|
|
}
|
|
return std::vector<unsigned char>(out, out + out_len);
|
|
}
|
|
|
|
std::vector<unsigned char> HmacSha256(const std::vector<unsigned char>& key, const std::string& msg) {
|
|
if (key.empty()) return {};
|
|
return HmacSha256(key.data(), key.size(), msg);
|
|
}
|
|
|
|
bool ParseHostFromEndpoint(const std::string& endpoint, std::string& host_out) {
|
|
size_t scheme = endpoint.find("://");
|
|
if (scheme == std::string::npos) return false;
|
|
size_t start = scheme + 3;
|
|
size_t end = endpoint.find('/', start);
|
|
host_out = end == std::string::npos ? endpoint.substr(start) : endpoint.substr(start, end - start);
|
|
return !host_out.empty();
|
|
}
|
|
|
|
bool FormatAmzDate(std::string& amz_date, std::string& date) {
|
|
std::time_t now = std::time(nullptr);
|
|
std::tm gm{};
|
|
#if defined(_WIN32)
|
|
if (gmtime_s(&gm, &now) != 0) return false;
|
|
#else
|
|
if (!gmtime_r(&now, &gm)) return false;
|
|
#endif
|
|
char d1[17] = {0};
|
|
char d2[9] = {0};
|
|
if (std::strftime(d1, sizeof(d1), "%Y%m%dT%H%M%SZ", &gm) == 0) return false;
|
|
if (std::strftime(d2, sizeof(d2), "%Y%m%d", &gm) == 0) return false;
|
|
amz_date.assign(d1);
|
|
date.assign(d2);
|
|
return true;
|
|
}
|
|
|
|
bool BuildS3SigV4Headers(const std::string& endpoint,
|
|
const std::string& region,
|
|
const std::string& access_key,
|
|
const std::string& secret_key,
|
|
const std::string& bucket,
|
|
const std::string& key,
|
|
const std::string& payload_hash,
|
|
std::string& url_out,
|
|
std::vector<std::string>& headers_out,
|
|
std::string& err) {
|
|
err.clear();
|
|
headers_out.clear();
|
|
url_out.clear();
|
|
|
|
std::string host;
|
|
if (!ParseHostFromEndpoint(endpoint, host)) {
|
|
err = "invalid endpoint";
|
|
return false;
|
|
}
|
|
|
|
std::string amz_date;
|
|
std::string date;
|
|
if (!FormatAmzDate(amz_date, date)) {
|
|
err = "failed to format time";
|
|
return false;
|
|
}
|
|
|
|
const std::string canonical_uri = "/" + bucket + "/" + UriEncodePath(key);
|
|
const std::string signed_headers = "host;x-amz-content-sha256;x-amz-date";
|
|
|
|
const std::string canonical_headers =
|
|
"host:" + host + "\n" +
|
|
"x-amz-content-sha256:" + payload_hash + "\n" +
|
|
"x-amz-date:" + amz_date + "\n";
|
|
|
|
const std::string canonical_request =
|
|
"PUT\n" + canonical_uri + "\n" +
|
|
"\n" +
|
|
canonical_headers + "\n" +
|
|
signed_headers + "\n" +
|
|
payload_hash;
|
|
|
|
const std::string scope = date + "/" + region + "/s3/aws4_request";
|
|
const std::string string_to_sign =
|
|
"AWS4-HMAC-SHA256\n" + amz_date + "\n" + scope + "\n" + Sha256Hex(canonical_request.data(), canonical_request.size());
|
|
|
|
const std::string ksecret = "AWS4" + secret_key;
|
|
std::vector<unsigned char> k0(ksecret.begin(), ksecret.end());
|
|
auto kDate = HmacSha256(k0, date);
|
|
auto kRegion = HmacSha256(kDate, region);
|
|
auto kService = HmacSha256(kRegion, "s3");
|
|
auto kSigning = HmacSha256(kService, "aws4_request");
|
|
auto sig = HmacSha256(kSigning, string_to_sign);
|
|
if (sig.empty()) {
|
|
err = "HMAC failed";
|
|
return false;
|
|
}
|
|
|
|
const std::string signature = ToHexLower(sig.data(), sig.size());
|
|
const std::string auth =
|
|
"AWS4-HMAC-SHA256 Credential=" + access_key + "/" + scope +
|
|
", SignedHeaders=" + signed_headers + ", Signature=" + signature;
|
|
|
|
url_out = endpoint;
|
|
if (!url_out.empty() && url_out.back() == '/') url_out.pop_back();
|
|
url_out += canonical_uri;
|
|
|
|
headers_out.push_back("x-amz-date: " + amz_date);
|
|
headers_out.push_back("x-amz-content-sha256: " + payload_hash);
|
|
headers_out.push_back("Authorization: " + auth);
|
|
return true;
|
|
}
|
|
#endif
|
|
|
|
bool IsRetryableHttp(long http_code) {
|
|
if (http_code == 408 || http_code == 429) return true;
|
|
return http_code >= 500;
|
|
}
|
|
|
|
} // namespace
|
|
|
|
bool MinioUploader::Init(const SimpleJson& config) {
|
|
endpoint_ = config.ValueOr<std::string>("endpoint", "http://localhost:9000");
|
|
bucket_ = config.ValueOr<std::string>("bucket", "alarms");
|
|
region_ = config.ValueOr<std::string>("region", "us-east-1");
|
|
|
|
presign_endpoint_ = config.ValueOr<std::string>("presign_endpoint", "");
|
|
presign_timeout_ms_ = config.ValueOr<int>("presign_timeout_ms", presign_timeout_ms_);
|
|
|
|
max_retries_ = std::max(0, config.ValueOr<int>("max_retries", max_retries_));
|
|
retry_backoff_ms_ = std::max(0, config.ValueOr<int>("retry_backoff_ms", retry_backoff_ms_));
|
|
retry_backoff_max_ms_ = std::max(retry_backoff_ms_, config.ValueOr<int>("retry_backoff_max_ms", retry_backoff_max_ms_));
|
|
retry_jitter_ms_ = std::max(0, config.ValueOr<int>("retry_jitter_ms", retry_jitter_ms_));
|
|
upload_timeout_sec_ = std::max(1, config.ValueOr<int>("upload_timeout_sec", upload_timeout_sec_));
|
|
upload_connect_timeout_ms_ = std::max(0, config.ValueOr<int>("upload_connect_timeout_ms", upload_connect_timeout_ms_));
|
|
|
|
std::string ak = config.ValueOr<std::string>("access_key", "");
|
|
std::string sk = config.ValueOr<std::string>("secret_key", "");
|
|
|
|
access_key_ = ResolveEnv("MINIO_ACCESS_KEY", ak);
|
|
secret_key_ = ResolveEnv("MINIO_SECRET_KEY", sk);
|
|
|
|
use_ssl_ = endpoint_.find("https://") == 0;
|
|
|
|
#if HAS_CURL
|
|
static std::once_flag curl_once;
|
|
std::call_once(curl_once, [] { curl_global_init(CURL_GLOBAL_DEFAULT); });
|
|
#endif
|
|
|
|
if (presign_endpoint_.empty() && (access_key_.empty() || secret_key_.empty())) {
|
|
std::cerr << "[MinioUploader] warning: access_key/secret_key not set; recommended to set presign_endpoint\n";
|
|
}
|
|
|
|
std::cout << "[MinioUploader] initialized, endpoint=" << endpoint_
|
|
<< " bucket=" << bucket_ << "\n";
|
|
return true;
|
|
}
|
|
|
|
UploadResult MinioUploader::Upload(const std::string& key,
|
|
const uint8_t* data,
|
|
size_t size,
|
|
const std::string& content_type) {
|
|
UploadResult result;
|
|
|
|
#if HAS_CURL
|
|
const std::string ct = content_type.empty() ? "application/octet-stream" : content_type;
|
|
|
|
const int max_attempts = std::max(1, max_retries_ + 1);
|
|
for (int attempt = 0; attempt < max_attempts; ++attempt) {
|
|
if (attempt > 0) {
|
|
int64_t backoff = retry_backoff_ms_;
|
|
if (attempt > 1) {
|
|
const int shift = attempt - 1;
|
|
const int64_t exp = (shift >= 30) ? static_cast<int64_t>(retry_backoff_max_ms_) : (static_cast<int64_t>(retry_backoff_ms_) << shift);
|
|
backoff = std::min<int64_t>(retry_backoff_max_ms_, exp);
|
|
}
|
|
if (retry_jitter_ms_ > 0) {
|
|
backoff += (std::rand() % (retry_jitter_ms_ + 1));
|
|
}
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(static_cast<int>(backoff)));
|
|
}
|
|
|
|
PresignResult presigned;
|
|
std::string upload_url;
|
|
std::vector<std::string> extra_headers;
|
|
if (!presign_endpoint_.empty()) {
|
|
std::string perr;
|
|
if (!GetPresignedPutUrl(presign_endpoint_, bucket_, key, ct, size, presign_timeout_ms_, presigned, perr)) {
|
|
result.error = perr;
|
|
continue;
|
|
}
|
|
upload_url = presigned.upload_url;
|
|
extra_headers = presigned.headers;
|
|
} else {
|
|
#if HAS_OPENSSL
|
|
if (access_key_.empty() || secret_key_.empty()) {
|
|
result.error = "missing access_key/secret_key (or configure presign_endpoint)";
|
|
return result;
|
|
}
|
|
const std::string payload_hash = Sha256Hex(data, size);
|
|
std::string berr;
|
|
if (!BuildS3SigV4Headers(endpoint_, region_, access_key_, secret_key_, bucket_, key, payload_hash,
|
|
upload_url, extra_headers, berr)) {
|
|
result.error = berr;
|
|
continue;
|
|
}
|
|
#else
|
|
result.error = "presign_endpoint is required (or rebuild with OpenSSL to enable SigV4 upload using access_key/secret_key)";
|
|
return result;
|
|
#endif
|
|
}
|
|
|
|
if (upload_url.empty()) {
|
|
result.error = "upload_url is empty";
|
|
continue;
|
|
}
|
|
|
|
CURL* curl = curl_easy_init();
|
|
if (!curl) {
|
|
result.error = "curl_easy_init failed";
|
|
continue;
|
|
}
|
|
|
|
struct curl_slist* headers = nullptr;
|
|
headers = curl_slist_append(headers, ("Content-Type: " + ct).c_str());
|
|
headers = curl_slist_append(headers, "Expect:");
|
|
for (const auto& h : extra_headers) {
|
|
headers = curl_slist_append(headers, h.c_str());
|
|
}
|
|
|
|
curl_easy_setopt(curl, CURLOPT_URL, upload_url.c_str());
|
|
curl_easy_setopt(curl, CURLOPT_UPLOAD, 1L);
|
|
curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers);
|
|
curl_easy_setopt(curl, CURLOPT_INFILESIZE_LARGE, static_cast<curl_off_t>(size));
|
|
curl_easy_setopt(curl, CURLOPT_TIMEOUT, static_cast<long>(upload_timeout_sec_));
|
|
curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT_MS, static_cast<long>(upload_connect_timeout_ms_));
|
|
|
|
struct ReadData {
|
|
const uint8_t* data;
|
|
size_t remaining;
|
|
} read_data{data, size};
|
|
|
|
curl_easy_setopt(curl, CURLOPT_READFUNCTION,
|
|
+[](char* buffer, size_t s, size_t n, void* userp) -> size_t {
|
|
auto* rd = static_cast<ReadData*>(userp);
|
|
size_t to_copy = std::min(s * n, rd->remaining);
|
|
if (to_copy > 0) {
|
|
std::memcpy(buffer, rd->data, to_copy);
|
|
rd->data += to_copy;
|
|
rd->remaining -= to_copy;
|
|
}
|
|
return to_copy;
|
|
});
|
|
curl_easy_setopt(curl, CURLOPT_READDATA, &read_data);
|
|
|
|
CURLcode res = curl_easy_perform(curl);
|
|
long http_code = 0;
|
|
curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &http_code);
|
|
|
|
curl_slist_free_all(headers);
|
|
curl_easy_cleanup(curl);
|
|
|
|
if (res == CURLE_OK && http_code >= 200 && http_code < 300) {
|
|
result.success = true;
|
|
if (!presigned.public_url.empty()) {
|
|
result.url = presigned.public_url;
|
|
} else if (!presigned.upload_url.empty()) {
|
|
result.url = presigned.upload_url;
|
|
} else {
|
|
result.url = endpoint_ + "/" + bucket_ + "/" + key;
|
|
}
|
|
result.error.clear();
|
|
return result;
|
|
}
|
|
|
|
if (res != CURLE_OK) {
|
|
result.error = curl_easy_strerror(res);
|
|
} else {
|
|
result.error = "HTTP " + std::to_string(http_code);
|
|
}
|
|
|
|
if (attempt == max_attempts - 1) {
|
|
return result;
|
|
}
|
|
if (res == CURLE_OK && !IsRetryableHttp(http_code)) {
|
|
return result;
|
|
}
|
|
}
|
|
|
|
return result;
|
|
#else
|
|
// Stub for Windows
|
|
std::cout << "[MinioUploader] would upload " << size << " bytes as key=" << key
|
|
<< " (configure presign_endpoint to enable real upload)\n";
|
|
result.success = true;
|
|
result.url = endpoint_ + "/" + bucket_ + "/" + key;
|
|
#endif
|
|
|
|
return result;
|
|
}
|
|
|
|
} // namespace rk3588
|