Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions native/csrc/catalog/bindings_store.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -36,12 +36,19 @@ dmi_store::S3Config s3_config(const py::dict& d) {
c.secret_key = get<std::string>(d, "s3_secret_key", "");
c.session_token = get<std::string>(d, "s3_session_token", "");
c.allow_insecure_http = get<bool>(d, "s3_allow_insecure_http", false);
c.ca_file = get<std::string>(d, "s3_ca_file", "");
c.ca_path = get<std::string>(d, "s3_ca_path", "");
c.connect_timeout_s = get<int>(d, "s3_connect_timeout_s", c.connect_timeout_s);
c.read_timeout_s = get<int>(d, "s3_read_timeout_s", c.read_timeout_s);
c.max_attempts = get<int>(d, "s3_max_attempts", c.max_attempts);
if (c.endpoint.empty() || c.bucket.empty()) {
throw py::value_error("s3_endpoint and s3_bucket are required");
}
// The client would refuse every request of an invalid config (a CA on
// http://, a missing CA file, a sub-5 MiB part); raising here makes the
// refusal land when the service or reader is built instead.
const std::string invalid = dmi_store::S3Client::ValidateConfig(c);
if (!invalid.empty()) throw py::value_error("s3: " + invalid);
return c;
}

Expand Down
8 changes: 8 additions & 0 deletions native/csrc/catalog/conformance_catalog.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -479,6 +479,8 @@ std::string respond(const std::string& line, Session* session) {
s3_config.access_key = jc::FindString(line, "access");
s3_config.secret_key = jc::FindString(line, "secret");
s3_config.allow_insecure_http = jc::FindBool(line, "insecure");
s3_config.ca_file = jc::FindString(line, "ca_file");
s3_config.ca_path = jc::FindString(line, "ca_path");
dmi_store::S3Client s3(s3_config);
dmi_catalog::NativeCaptureReader reader(
&s3, s3_config.bucket, session->client, rc);
Expand Down Expand Up @@ -511,6 +513,8 @@ std::string respond(const std::string& line, Session* session) {
s3_config.access_key = jc::FindString(line, "access");
s3_config.secret_key = jc::FindString(line, "secret");
s3_config.allow_insecure_http = jc::FindBool(line, "insecure");
s3_config.ca_file = jc::FindString(line, "ca_file");
s3_config.ca_path = jc::FindString(line, "ca_path");
dmi_store::S3Client s3(s3_config);
dmi_catalog::NativeCaptureReader reader(
&s3, s3_config.bucket, session->client, rc);
Expand Down Expand Up @@ -547,6 +551,8 @@ std::string respond(const std::string& line, Session* session) {
s3_config.access_key = jc::FindString(line, "access");
s3_config.secret_key = jc::FindString(line, "secret");
s3_config.allow_insecure_http = jc::FindBool(line, "insecure");
s3_config.ca_file = jc::FindString(line, "ca_file");
s3_config.ca_path = jc::FindString(line, "ca_path");
dmi_store::S3Client s3(s3_config);
dmi_catalog::NativeCaptureReader reader(
&s3, s3_config.bucket, session->client, rc);
Expand Down Expand Up @@ -954,6 +960,8 @@ std::string respond(const std::string& line, Session* session) {
s3_config.access_key = jc::FindString(line, "access");
s3_config.secret_key = jc::FindString(line, "secret");
s3_config.allow_insecure_http = jc::FindBool(line, "insecure");
s3_config.ca_file = jc::FindString(line, "ca_file");
s3_config.ca_path = jc::FindString(line, "ca_path");
dmi_store::S3Client s3(s3_config);
std::vector<dmi_catalog::PackRefData> refs;
for (const std::string& element : jc::SplitElements(
Expand Down
4 changes: 3 additions & 1 deletion native/csrc/store/conformance_store.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
// "access":"...","secret":"...","token":null,"insecure":true,
// "key":"...","data_b64":"...","metadata":{...},"content_type":"...",
// "multipart_threshold":N,"multipart_chunk":N,"max_attempts":N,
// "connect_timeout":N,"read_timeout":N}
// "connect_timeout":N,"read_timeout":N,"ca_file":"...","ca_path":"..."}
// -> {"ok":true,"etag":"...","attempts":N}
// {"op":"get",...,"offset":N,"length":N} -> {"ok":true,"data_b64":"...","attempts":N}
// {"op":"head",...} -> {"ok":true,"found":bool,"size":N,"metadata":{...},
Expand Down Expand Up @@ -92,6 +92,8 @@ dmi_store::S3Config ReadConfig(const std::string& line) {
}
}
config.allow_insecure_http = jc::FindBool(line, "insecure");
config.ca_file = jc::FindString(line, "ca_file");
config.ca_path = jc::FindString(line, "ca_path");
const int64_t connect_timeout = Integer(line, "connect_timeout");
config.connect_timeout_s =
static_cast<int>(connect_timeout > 0 ? connect_timeout : 5);
Expand Down
99 changes: 76 additions & 23 deletions native/csrc/store/s3_client.cpp
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
#include "s3_client.h"

#include <curl/curl.h>
#include <sys/stat.h>
#include <unistd.h>

#include <chrono>
#include <cstring>
Expand Down Expand Up @@ -109,8 +111,50 @@ std::string XmlTag(const std::string& xml, const std::string& tag) {
return xml.substr(start, end - start);
}

// Checked when the client is built, so a mistyped path fails with its name
// rather than as libcurl's "problem with the SSL CA cert" at the first
// request.
bool IsReadableFile(const std::string& path) {
struct stat st {};
return ::stat(path.c_str(), &st) == 0 && S_ISREG(st.st_mode) &&
::access(path.c_str(), R_OK) == 0;
}

bool IsDirectory(const std::string& path) {
struct stat st {};
return ::stat(path.c_str(), &st) == 0 && S_ISDIR(st.st_mode);
}

} // namespace

std::string S3Client::ValidateConfig(const S3Config& config) {
const bool https = config.endpoint.compare(0, 8, "https://") == 0;
if (https && config.allow_insecure_http) {
// Silently downgrading TLS was never the intent of the flag -- it gates
// plain http:// endpoints for local Garage.
return "https endpoint with allow_insecure_http is refused";
}
if (!https && (!config.ca_file.empty() || !config.ca_path.empty())) {
// A CA on a plain-http endpoint would read as "this is TLS" while every
// byte, credentials included, goes in the clear.
return "ca_file and ca_path apply only to an https endpoint";
}
if (!config.ca_file.empty() && !IsReadableFile(config.ca_file)) {
return "ca_file is not a readable file: " + config.ca_file;
}
if (!config.ca_path.empty() && !IsDirectory(config.ca_path)) {
return "ca_path is not a directory: " + config.ca_path;
}
if (config.multipart_chunk_bytes < kMinMultipartPartBytes) {
// The Python store refuses the same (s3.py _MIN_MULTIPART_BYTES). Left
// to the server, the upload fails only at CompleteMultipartUpload, after
// every part was sent.
return "multipart_chunk_bytes must be at least " +
std::to_string(kMinMultipartPartBytes) + " (S3's minimum part size)";
}
return "";
}

S3Client::S3Client(S3Config config) : config_(std::move(config)) {
// Explicit, rather than leaning on the implicit init inside
// curl_easy_init: that implicit path carries libcurl's thread-safety
Expand All @@ -132,11 +176,7 @@ S3Client::S3Client(S3Config config) : config_(std::move(config)) {
}
while (!rest.empty() && rest.back() == '/') rest.pop_back();
host_ = rest;
if (is_https_ && config_.allow_insecure_http) {
// Refused at construction: silently downgrading TLS was never the intent
// of the flag — it gates plain http:// endpoints for local Garage.
host_.clear();
}
config_error_ = ValidateConfig(config_);
}

S3Client::~S3Client() = default;
Expand All @@ -147,30 +187,24 @@ S3Response S3Client::Exchange(
const std::map<std::string, std::string>& extra_headers,
const uint8_t* body, size_t body_len, const std::string& body_hash_hex) {
S3Response response;
if (host_.empty()) {
response.error = "https endpoint with allow_insecure_http is refused";
if (!config_error_.empty()) {
last_attempts_ = 0;
response.error = config_error_;
return response;
}
const std::string amz_date = AmzDate(std::time(nullptr));
const std::string datestamp = Datestamp(amz_date);

const std::string encoded_resource = "/" + config_.bucket + "/" + key;
const std::string encoded_path = UriEncode(encoded_resource, true);

// Every signed header except x-amz-date, which each attempt stamps.
std::map<std::string, std::string> headers;
headers["host"] = host_;
headers["x-amz-date"] = amz_date;
headers["x-amz-content-sha256"] = body_hash_hex;
if (!config_.session_token.empty()) {
headers["x-amz-security-token"] = config_.session_token;
}
for (const auto& [name, value] : extra_headers) {
headers[LowerHeader(name)] = value;
}
const std::string authz = AuthorizationHeader(
config_.access_key, config_.secret_key, datestamp, amz_date,
config_.region, "s3", method, encoded_path, query, headers,
body_hash_hex);

// Canonical query string for the URL (same encoding the signer used).
std::string query_text;
Expand All @@ -196,6 +230,16 @@ S3Response S3Client::Exchange(
last_attempts_ = 0;
for (int attempt = 0; attempt < config_.max_attempts; ++attempt) {
++last_attempts_;
// Signed per attempt, not once before the loop: SigV4 binds the
// signature to x-amz-date, and S3 refuses a date more than 15 minutes
// off. Replaying the first attempt's date let a slow first attempt (up
// to read_timeout_s each, plus backoff) age every retry after it.
const std::string amz_date = AmzDate(std::time(nullptr));
headers["x-amz-date"] = amz_date;
const std::string authz = AuthorizationHeader(
config_.access_key, config_.secret_key, Datestamp(amz_date), amz_date,
config_.region, "s3", method, encoded_path, query, headers,
body_hash_hex);
CURL* curl = curl_easy_init();
if (!curl) {
response.error = "curl_easy_init failed";
Expand All @@ -218,15 +262,24 @@ S3Response S3Client::Exchange(
curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT, config_.connect_timeout_s);
curl_easy_setopt(curl, CURLOPT_TIMEOUT, config_.read_timeout_s);
curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1L);
if (!is_https_) {
// Plain http only by explicit opt-in (local Garage); https always
// verifies (no CURLOPT_SSL_VERIFYPEER toggle exists anywhere here).
if (!config_.allow_insecure_http) {
response.error = "plain http endpoint requires allow_insecure_http";
curl_slist_free_all(chunk);
curl_easy_cleanup(curl);
return response;
if (is_https_) {
// https always verifies: peer and host name, stated explicitly rather
// than left to libcurl's defaults, and never switched off. A private
// CA adds trust; it does not relax the check.
curl_easy_setopt(curl, CURLOPT_SSL_VERIFYPEER, 1L);
curl_easy_setopt(curl, CURLOPT_SSL_VERIFYHOST, 2L);
if (!config_.ca_file.empty()) {
curl_easy_setopt(curl, CURLOPT_CAINFO, config_.ca_file.c_str());
}
if (!config_.ca_path.empty()) {
curl_easy_setopt(curl, CURLOPT_CAPATH, config_.ca_path.c_str());
}
} else if (!config_.allow_insecure_http) {
// Plain http only by explicit opt-in (local Garage).
response.error = "plain http endpoint requires allow_insecure_http";
curl_slist_free_all(chunk);
curl_easy_cleanup(curl);
return response;
}
std::string response_body;
std::map<std::string, std::string> response_headers;
Expand Down
18 changes: 18 additions & 0 deletions native/csrc/store/s3_client.h
Original file line number Diff line number Diff line change
Expand Up @@ -32,10 +32,18 @@ struct S3Config {
std::string secret_key;
std::string session_token; // empty when unused
bool allow_insecure_http = false;
// https only: trust a private CA. ca_file is a PEM bundle
// (CURLOPT_CAINFO), ca_path an OpenSSL-hashed certificate directory
// (CURLOPT_CAPATH). Both empty uses libcurl's default trust store. Either
// way https always verifies the peer and the host name.
std::string ca_file;
std::string ca_path;
int connect_timeout_s = 5;
int read_timeout_s = 120;
int max_attempts = 4;
uint64_t multipart_threshold_bytes = 64ull * 1024 * 1024;
// The size of every part but the last. S3 refuses a smaller non-final
// part (EntityTooSmall), so the client refuses one under kMinMultipartPartBytes.
uint64_t multipart_chunk_bytes = 16ull * 1024 * 1024;
std::string user_agent = "dmi-native-store/1";
};
Expand Down Expand Up @@ -70,9 +78,18 @@ struct ListResult {
std::vector<ListedObject> objects;
};

// S3's minimum size for every part of a multipart upload but the last.
inline constexpr uint64_t kMinMultipartPartBytes = 5ull * 1024 * 1024;

class S3Client {
public:
// An invalid config (see ValidateConfig) does not throw: the client
// refuses every request with the reason, before anything goes out.
explicit S3Client(S3Config config);

// Empty when `config` is usable; otherwise why not. Callers with an error
// channel of their own (the Python bindings) check it at construction.
static std::string ValidateConfig(const S3Config& config);
~S3Client();

S3Client(const S3Client&) = delete;
Expand Down Expand Up @@ -116,6 +133,7 @@ class S3Client {

private:
S3Config config_;
std::string config_error_; // non-empty: every request is refused
std::string host_; // endpoint host (with :port when non-default)
std::string scheme_;
bool is_https_ = false;
Expand Down
17 changes: 17 additions & 0 deletions src/dmi/storage/native_capture.py
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,11 @@ class NativeCaptureStorageConfig:
s3_session_token: str = field(default="", repr=False)
# Plain-HTTP endpoints (a local Garage or MinIO) must be opted into.
s3_allow_insecure_http: bool = False
# https only: trust a private CA, as a PEM bundle (s3_ca_file) or an
# OpenSSL-hashed certificate directory (s3_ca_path). Empty uses the
# system trust store. https always verifies the peer either way.
s3_ca_file: str = ""
s3_ca_path: str = ""
# The name packs are indexed under; readers resolve it to this store.
store_id: str = "s3"

Expand Down Expand Up @@ -131,6 +136,16 @@ def __post_init__(self) -> None:
"never downgrades TLS; leave it False for https://")
else:
raise ValueError("s3_endpoint must start with http:// or https://")
for name in ("s3_ca_file", "s3_ca_path"):
value = getattr(self, name)
if type(value) is not str:
raise TypeError(f"{name} must be a str (empty for the system "
"trust store)")
# The native client refuses the same pairing: a CA on http://
# reads as "this is TLS" while credentials go in the clear.
if value and not self.s3_endpoint.startswith("https://"):
raise ValueError(f"{name} applies only to an https:// "
"s3_endpoint")
if type(self.clickhouse_port) is not int or not 0 < self.clickhouse_port < 65536:
raise ValueError("clickhouse_port must be in 1..65535")
_positive("poll_interval_s", self.poll_interval_s, float)
Expand Down Expand Up @@ -158,6 +173,8 @@ def _native_dict(self) -> dict[str, Any]:
"s3_secret_key": self.s3_secret_key,
"s3_session_token": self.s3_session_token,
"s3_allow_insecure_http": self.s3_allow_insecure_http,
"s3_ca_file": self.s3_ca_file,
"s3_ca_path": self.s3_ca_path,
"store_id": self.store_id,
"clickhouse_host": self.clickhouse_host,
"clickhouse_port": self.clickhouse_port,
Expand Down
Loading
Loading