[net] Add SendPutReq to RCurlConnection for S3 uploads

net/curl/inc/ROOT/RCurlConnection.hxx

@@ -75,6 +75,7 @@ private:
 
    void SetupErrorBuffer();
    void SetOptions();
+   void ResetHandle();
    RResult<void> SetUrl(const std::string &url);
    void Perform(RStatus &status);
 
@@ -116,6 +117,8 @@ public:
    /// a valid batching of requests into multiple multi-range requests takes place automatically.
    /// The fNBytesRecv member of the ranges is only well-defined on success.
    RStatus SendRangesReq(std::size_t N, RUserRange *ranges);
+   /// Uploads data to the URL using an HTTP PUT request.
+   RStatus SendPutReq(const unsigned char *data, std::size_t length);
 
    const std::string &GetEscapedUrl() const { return fEscapedUrl; }
 

net/curl/src/RCurlConnection.cxx

@@ -552,6 +552,45 @@ void ReverseDisplacements(std::vector<std::size_t> &displacements, ROOT::Interna
    }
 }
 
+struct RUploadState {
+   const unsigned char *fData = nullptr;
+   std::size_t fLength = 0;
+   std::size_t fOffset = 0;
+   std::string fErrorMsg; ///< Set on failure, reported back to the caller via RStatus
+};
+
+std::size_t CallbackPutRead(char *buffer, std::size_t size, std::size_t nmemb, void *userdata)
+{
+   auto *state = static_cast<RUploadState *>(userdata);
+   assert(state->fOffset <= state->fLength);
+
+   std::size_t remaining = state->fLength - state->fOffset;
+   if (remaining == 0)
+      return 0;
+
+   std::size_t requested = size * nmemb;
+   if (requested > remaining) {
+      state->fErrorMsg = "curl requested " + std::to_string(requested) +
+                         " bytes but only " + std::to_string(remaining) + " remain";
+      return CURL_READFUNC_ABORT;
+   }
+
+   memcpy(buffer, state->fData + state->fOffset, requested);
+   state->fOffset += requested;
+   return requested;
+}
+
+int CallbackPutSeek(void *userdata, curl_off_t offset, int origin)
+{
+   auto *state = static_cast<RUploadState *>(userdata);
+   if (origin != SEEK_SET)
+      return CURL_SEEKFUNC_CANTSEEK;
+   if (offset < 0 || static_cast<std::size_t>(offset) > state->fLength)
+      return CURL_SEEKFUNC_FAIL;
+   state->fOffset = static_cast<std::size_t>(offset);
+   return CURL_SEEKFUNC_OK;
+}
+
 std::string GetCurlErrorString(CURLcode code)
 {
    return std::string(curl_easy_strerror(code)) + " (" + std::to_string(code) + ")";
@@ -660,6 +699,35 @@ void ROOT::Internal::RCurlConnection::SetOptions()
    R__ASSERT(rc == CURLE_OK);
 }
 
+void ROOT::Internal::RCurlConnection::ResetHandle()
+{
+   auto rc = curl_easy_setopt(fHandle, CURLOPT_NOBODY, 0L);
+   R__ASSERT(rc == CURLE_OK);
+   rc = curl_easy_setopt(fHandle, CURLOPT_HTTPGET, 0L);
+   R__ASSERT(rc == CURLE_OK);
+   rc = curl_easy_setopt(fHandle, CURLOPT_UPLOAD, 0L);
+   R__ASSERT(rc == CURLE_OK);
+   rc = curl_easy_setopt(fHandle, CURLOPT_RANGE, NULL);
+   R__ASSERT(rc == CURLE_OK);
+   rc = curl_easy_setopt(fHandle, CURLOPT_READFUNCTION, NULL);
+   R__ASSERT(rc == CURLE_OK);
+   rc = curl_easy_setopt(fHandle, CURLOPT_READDATA, NULL);
+   R__ASSERT(rc == CURLE_OK);
+   rc = curl_easy_setopt(fHandle, CURLOPT_SEEKFUNCTION, NULL);
+   R__ASSERT(rc == CURLE_OK);
+   rc = curl_easy_setopt(fHandle, CURLOPT_SEEKDATA, NULL);
+   R__ASSERT(rc == CURLE_OK);
+   rc = curl_easy_setopt(fHandle, CURLOPT_INFILESIZE_LARGE, static_cast<curl_off_t>(-1));
+   R__ASSERT(rc == CURLE_OK);
+
+#ifndef HAS_CURL_EASY_HEADER
+   rc = curl_easy_setopt(fHandle, CURLOPT_HEADERFUNCTION, NULL);
+   R__ASSERT(rc == CURLE_OK);
+   rc = curl_easy_setopt(fHandle, CURLOPT_HEADERDATA, NULL);
+   R__ASSERT(rc == CURLE_OK);
+#endif
+}
+
 ROOT::RResult<void> ROOT::Internal::RCurlConnection::SetUrl(const std::string &url)
 {
    CURLU *cu = curl_url();
@@ -731,17 +799,9 @@ ROOT::Internal::RCurlConnection::RStatus ROOT::Internal::RCurlConnection::SendHe
 {
    remoteSize = kUnknownSize;
 
+   ResetHandle();
    auto rc = curl_easy_setopt(fHandle, CURLOPT_NOBODY, 1);
    R__ASSERT(rc == CURLE_OK);
-   rc = curl_easy_setopt(fHandle, CURLOPT_RANGE, NULL); // may have been set by a previous SendRangesReq() on the object
-   R__ASSERT(rc == CURLE_OK);
-
-#ifndef HAS_CURL_EASY_HEADER
-   rc = curl_easy_setopt(fHandle, CURLOPT_HEADERFUNCTION, NULL);
-   R__ASSERT(rc == CURLE_OK);
-   rc = curl_easy_setopt(fHandle, CURLOPT_HEADERDATA, NULL);
-   R__ASSERT(rc == CURLE_OK);
-#endif
 
    RStatus status;
    Perform(status);
@@ -778,6 +838,7 @@ ROOT::Internal::RCurlConnection::SendRangesReq(std::size_t N, RUserRange *ranges
       return RStatus(RStatus::kSuccess);
    }
 
+   ResetHandle();
    auto rc = curl_easy_setopt(fHandle, CURLOPT_HTTPGET, 1);
    R__ASSERT(rc == CURLE_OK);
 
@@ -849,6 +910,58 @@ ROOT::Internal::RCurlConnection::SendRangesReq(std::size_t N, RUserRange *ranges
    return status;
 }
 
+ROOT::Internal::RCurlConnection::RStatus
+ROOT::Internal::RCurlConnection::SendPutReq(const unsigned char *data, std::size_t length)
+{
+   static constexpr std::size_t kExpect100Threshold = 4096;
+
+   ResetHandle();
+
+   auto rc = curl_easy_setopt(fHandle, CURLOPT_UPLOAD, 1L);
+   R__ASSERT(rc == CURLE_OK);
+   rc = curl_easy_setopt(fHandle, CURLOPT_INFILESIZE_LARGE, static_cast<curl_off_t>(length));
+   R__ASSERT(rc == CURLE_OK);
+
+   // For small uploads, disable Expect: 100-continue to avoid an extra round trip.
+   // For large uploads, let curl send it (the default) so the server can reject early.
+   struct curl_slist *extraHeaders = nullptr;
+   if (length < kExpect100Threshold) {
+      extraHeaders = curl_slist_append(extraHeaders, "Expect:");
+      rc = curl_easy_setopt(fHandle, CURLOPT_HTTPHEADER, extraHeaders);
+      R__ASSERT(rc == CURLE_OK);
+   }
+
+   RUploadState uploadState{data, length, 0, {}};
+   rc = curl_easy_setopt(fHandle, CURLOPT_READFUNCTION, CallbackPutRead);
+   R__ASSERT(rc == CURLE_OK);
+   rc = curl_easy_setopt(fHandle, CURLOPT_READDATA, &uploadState);
+   R__ASSERT(rc == CURLE_OK);
+
+   // Seek callback is required when CURLOPT_FOLLOWLOCATION is enabled: if the server responds
+   // with a redirect, curl needs to rewind the upload data and resend it to the new location.
+   rc = curl_easy_setopt(fHandle, CURLOPT_SEEKFUNCTION, CallbackPutSeek);
+   R__ASSERT(rc == CURLE_OK);
+   rc = curl_easy_setopt(fHandle, CURLOPT_SEEKDATA, &uploadState);
+   R__ASSERT(rc == CURLE_OK);
+
+   RStatus status;
+   Perform(status);
+
+   if (extraHeaders)
+      curl_slist_free_all(extraHeaders);
+   rc = curl_easy_setopt(fHandle, CURLOPT_HTTPHEADER, NULL);
+   R__ASSERT(rc == CURLE_OK);
+
+   if (!uploadState.fErrorMsg.empty()) {
+      status.fStatusCode = RStatus::kIOError;
+      if (!status.fStatusMsg.empty())
+         status.fStatusMsg += "; ";
+      status.fStatusMsg += uploadState.fErrorMsg;
+   }
+
+   return status;
+}
+
 void ROOT::Internal::RCurlConnection::SetCredentials(const RS3Credentials &credentials)
 {
    ClearCredentials();

net/curl/test/curl_connection.cxx

@@ -4,11 +4,68 @@
 
 #include "TServerSocket.h"
 
+#include <algorithm>
+#include <cctype>
 #include <cstdint>
 #include <cstring>
 #include <string>
 #include <thread>
 
+/// Convert a string to lower-case in place and return a reference to it.
+static std::string &ToLower(std::string &s)
+{
+   std::transform(s.begin(), s.end(), s.begin(), [](unsigned char c) { return std::tolower(c); });
+   return s;
+}
+
+/// Accept a PUT request: read headers + body, optionally respond to Expect: 100-continue, send 200 OK.
+static void TaskRecvPut(TServerSocket *serverSocket, std::string *requestHeaders, std::string *requestBody)
+{
+   requestHeaders->clear();
+   requestBody->clear();
+   auto sock = serverSocket->Accept();
+
+   const char *eof = "\r\n\r\n";
+   const std::size_t eofLen = strlen(eof);
+   std::size_t nextInEof = 0;
+   char c;
+   while (sock->RecvRaw(&c, 1)) {
+      requestHeaders->push_back(c);
+      if (c == eof[nextInEof]) {
+         if (++nextInEof == eofLen)
+            break;
+      } else {
+         nextInEof = 0;
+      }
+   }
+
+   // If the client sent Expect: 100-continue, respond with HTTP 100 before reading the body
+   std::string headersLower = *requestHeaders;
+   ToLower(headersLower);
+   if (headersLower.find("expect: 100-continue") != std::string::npos) {
+      const char *continueResponse = "HTTP/1.1 100 Continue\r\n\r\n";
+      sock->SendRaw(continueResponse, strlen(continueResponse));
+   }
+
+   // Parse content-length (case-insensitive)
+   std::size_t contentLength = 0;
+   auto pos = headersLower.find("content-length: ");
+   if (pos != std::string::npos) {
+      auto valStart = pos + strlen("content-length: ");
+      auto valEnd = headersLower.find("\r\n", valStart);
+      contentLength = std::stoul(headersLower.substr(valStart, valEnd - valStart));
+   }
+
+   if (contentLength > 0) {
+      requestBody->resize(contentLength);
+      sock->RecvRaw(&(*requestBody)[0], contentLength);
+   }
+
+   const char *response = "HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n";
+   sock->SendRaw(response, strlen(response));
+   sock->Close();
+}
+
 static void TaskRecv(TServerSocket *serverSocket, std::string *request)
 {
    request->clear();
@@ -63,3 +120,32 @@ TEST(RCurlConnection, Cred)
    threadRecv.join();
    EXPECT_EQ(std::string::npos, request.find("\r\nAuthorization: "));
 }
+
+TEST(RCurlConnection, Put)
+{
+   TServerSocket sock(0, false, TServerSocket::kDefaultBacklog, -1, ESocketBindOption::kInaddrLoopback);
+   const std::string url =
+      std::string("http://") + sock.GetLocalInetAddress().GetHostAddress() + ":" + std::to_string(sock.GetLocalPort());
+
+   const unsigned char payload[] = "Hello, S3!";
+   const std::size_t payloadLen = sizeof(payload) - 1; // exclude null terminator
+
+   std::string headers;
+   std::string body;
+   std::thread threadRecv(TaskRecvPut, &sock, &headers, &body);
+
+   ROOT::Internal::RCurlConnection conn(url);
+   auto status = conn.SendPutReq(payload, payloadLen);
+
+   threadRecv.join();
+   EXPECT_TRUE(static_cast<bool>(status));
+   EXPECT_EQ(0u, headers.find("PUT "));
+
+   // Normalize headers to lower-case for case-insensitive matching
+   std::string headersLower = headers;
+   ToLower(headersLower);
+   auto clPos = headersLower.find("content-length: " + std::to_string(payloadLen));
+   ASSERT_NE(std::string::npos, clPos) << "content-length header not found in request";
+
+   EXPECT_EQ(std::string(reinterpret_cast<const char *>(payload), payloadLen), body);
+}

net/curl/test/curl_env.cxx

@@ -11,8 +11,10 @@
 #include "TCurlFile.h"
 #include "TSystem.h"
 
+#include <cstring>
 #include <memory>
 #include <utility>
+#include <vector>
 
 TEST(RCurlConnection, CredFromEnv)
 {
@@ -93,3 +95,58 @@ TEST(CurlFile, S3Credentials)
    gSystem->Unsetenv("S3_ACCESS_KEY");
    gSystem->Unsetenv("S3_SECRET_KEY");
 }
+
+TEST(CurlFile, S3PutAndRead)
+{
+   const auto testAccessKey = std::getenv("ROOT_TEST_S3_ACCESS_KEY");
+   const auto testSecretKey = std::getenv("ROOT_TEST_S3_SECRET_KEY");
+   if (!testAccessKey || testAccessKey[0] == '\0' || !testSecretKey || testSecretKey[0] == '\0') {
+      GTEST_SKIP() << "Missing S3 test credentials <ROOT_TEST_S3_[ACCESS|SECRET]_KEY>, skipping";
+   }
+   if (ROOT::Internal::RCurlConnection::GetCurlVersion() <= 0x078100) {
+      GTEST_SKIP() << "libcurl <= 7.81 is known to produce an AWSv4 signature incompatible with Ceph S3";
+   }
+
+   const std::string url = "https://root-project-s3test.s3.cern.ch/test-curl-put-roundtrip.bin";
+
+   ROOT::Internal::RS3Credentials creds;
+   creds.fAccessKey = testAccessKey;
+   creds.fSecretKey = testSecretKey;
+
+   // PUT a known payload
+   const unsigned char payload[] = "RCurlConnection::SendPutReq round-trip test";
+   const std::size_t payloadLen = sizeof(payload) - 1;
+
+   {
+      ROOT::Internal::RCurlConnection conn(url);
+      conn.SetCredentials(creds);
+      auto status = conn.SendPutReq(payload, payloadLen);
+      ASSERT_TRUE(static_cast<bool>(status)) << "PUT failed: " << status.fStatusMsg;
+   }
+
+   // HEAD to verify size
+   {
+      ROOT::Internal::RCurlConnection conn(url);
+      conn.SetCredentials(creds);
+      std::uint64_t remoteSize = 0;
+      auto status = conn.SendHeadReq(remoteSize);
+      ASSERT_TRUE(static_cast<bool>(status)) << "HEAD failed: " << status.fStatusMsg;
+      EXPECT_EQ(static_cast<std::uint64_t>(payloadLen), remoteSize);
+   }
+
+   // GET (range read) to verify content
+   {
+      std::vector<unsigned char> readback(payloadLen, 0);
+      ROOT::Internal::RCurlConnection::RUserRange range;
+      range.fDestination = readback.data();
+      range.fOffset = 0;
+      range.fLength = payloadLen;
+
+      ROOT::Internal::RCurlConnection conn(url);
+      conn.SetCredentials(creds);
+      auto status = conn.SendRangesReq(1, &range);
+      ASSERT_TRUE(static_cast<bool>(status)) << "GET failed: " << status.fStatusMsg;
+      EXPECT_EQ(payloadLen, range.fNBytesRecv);
+      EXPECT_EQ(0, memcmp(readback.data(), payload, payloadLen));
+   }
+}