[curl] Add SendPutReq to RCurlConnection with tests

Author: JasMehta08

net/curl/inc/ROOT/RCurlConnection.hxx

@@ -117,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,52 @@ void ReverseDisplacements(std::vector<std::size_t> &displacements, ROOT::Interna
    }
 }
 
+/// State for the PUT upload read callback: tracks progress through the upload buffer.
+struct RUploadState {
+   const unsigned char *fData = nullptr;
+   std::size_t fLength = 0;
+   std::size_t fOffset = 0;
+};
+
+/// CURLOPT_READFUNCTION callback for PUT uploads.  Copies up to `size * nmemb` bytes from the
+/// upload buffer into `buffer` and advances the offset.  Returns the number of bytes copied,
+/// i.e. min(requested, remaining), or 0 at end-of-data to signal that the upload is complete.
+std::size_t CallbackPutRead(char *buffer, std::size_t size, std::size_t nmemb, void *userdata)
+{
+   auto *state = static_cast<RUploadState *>(userdata);
+   R__ASSERT(state->fOffset <= state->fLength);
+
+   std::size_t remaining = state->fLength - state->fOffset;
+   if (remaining == 0)
+      return 0;
+
+   std::size_t requested = size * nmemb;
+   // CURL_READFUNC_ABORT (0x10000000) collides with a valid byte count at 256 MiB;
+   // assert that curl never asks for that much in a single callback invocation.
+   R__ASSERT(requested < CURL_READFUNC_ABORT);
+   std::size_t nbytes = std::min(requested, remaining);
+   memcpy(buffer, state->fData + state->fOffset, nbytes);
+   state->fOffset += nbytes;
+   return nbytes;
+}
+
+/// CURLOPT_SEEKFUNCTION callback for PUT uploads.  Required because CURLOPT_FOLLOWLOCATION
+/// is enabled: on a redirect curl needs to rewind the upload data before resending.
+/// Returns CURL_SEEKFUNC_OK (0) on success, CURL_SEEKFUNC_FAIL (1) for invalid offsets
+/// which aborts the transfer, or CURL_SEEKFUNC_CANTSEEK (2) for unsupported seek origins
+/// which lets curl try to work around it.
+int CallbackPutSeek(void *userdata, curl_off_t offset, int origin)
+{
+   auto *state = static_cast<RUploadState *>(userdata);
+   // curl documents that it will only use SEEK_SET; guard against anything else defensively.
+   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;
+}
+
 /// Wrapper around curl_easy_setopt that asserts on failure.  Most option-setting calls in this
 /// file use valid options and values by construction, so failure indicates a programming error.
 template <typename T>
@@ -655,7 +701,6 @@ void ROOT::Internal::RCurlConnection::SetOptions()
    static const std::string kUserAgent = GetUserAgentString();
    SetCurlOption(fHandle, CURLOPT_USERAGENT, kUserAgent.c_str());
    SetCurlOption(fHandle, CURLOPT_FOLLOWLOCATION, 1);
-   SetCurlOption(fHandle, CURLOPT_WRITEFUNCTION, CallbackData);
 }
 
 /// Reset method-specific sticky curl options so that the easy handle is in a clean state
@@ -666,6 +711,8 @@ void ROOT::Internal::RCurlConnection::ResetHandle()
    SetCurlOption(fHandle, CURLOPT_HTTPGET, 0L);
    SetCurlOption(fHandle, CURLOPT_UPLOAD, 0L);
    SetCurlOption(fHandle, CURLOPT_RANGE, static_cast<const char *>(nullptr));
+   SetCurlOption(fHandle, CURLOPT_WRITEFUNCTION, static_cast<curl_write_callback>(nullptr));
+   SetCurlOption(fHandle, CURLOPT_WRITEDATA, static_cast<void *>(nullptr));
    SetCurlOption(fHandle, CURLOPT_READFUNCTION, static_cast<curl_read_callback>(nullptr));
    SetCurlOption(fHandle, CURLOPT_READDATA, static_cast<void *>(nullptr));
    SetCurlOption(fHandle, CURLOPT_SEEKFUNCTION, static_cast<curl_seek_callback>(nullptr));
@@ -791,6 +838,7 @@ ROOT::Internal::RCurlConnection::SendRangesReq(std::size_t N, RUserRange *ranges
    SetCurlOption(fHandle, CURLOPT_HTTPGET, 1);
 
    RTransferState transfer(ranges, order, fHandle);
+   SetCurlOption(fHandle, CURLOPT_WRITEFUNCTION, CallbackData);
    SetCurlOption(fHandle, CURLOPT_WRITEDATA, &transfer);
 
 #ifndef HAS_CURL_EASY_HEADER
@@ -854,6 +902,26 @@ 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)
+{
+   ResetHandle();
+
+   SetCurlOption(fHandle, CURLOPT_UPLOAD, 1L);
+   SetCurlOption(fHandle, CURLOPT_INFILESIZE_LARGE, static_cast<curl_off_t>(length));
+
+   RUploadState uploadState{data, length, 0};
+   SetCurlOption(fHandle, CURLOPT_READFUNCTION, CallbackPutRead);
+   SetCurlOption(fHandle, CURLOPT_READDATA, &uploadState);
+   SetCurlOption(fHandle, CURLOPT_SEEKFUNCTION, CallbackPutSeek);
+   SetCurlOption(fHandle, CURLOPT_SEEKDATA, &uploadState);
+
+   RStatus status;
+   Perform(status);
+
+   return status;
+}
+
 void ROOT::Internal::RCurlConnection::SetCredentials(const RS3Credentials &credentials)
 {
    ClearCredentials();

net/curl/test/curl_connection.cxx

@@ -4,10 +4,67 @@
 
 #include "TServerSocket.h"
 
+#include <algorithm>
+#include <cctype>
 #include <cstdint>
 #include <cstring>
 #include <string>
 #include <thread>
+#include <vector>
+
+/// Return a lower-cased copy of the input string.
+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 = ToLower(*requestHeaders);
+   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)
 {
@@ -63,3 +120,131 @@ 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 = ToLower(headers);
+   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);
+}
+
+/// GET (range read) after PUT on the same handle — verifies that WRITEFUNCTION is set correctly
+/// in SendRangesReq after a PUT cleared it.
+TEST(RCurlConnection, GetAfterPut)
+{
+   TServerSocket sock(0, false, TServerSocket::kDefaultBacklog, -1, ESocketBindOption::kInaddrLoopback);
+   const std::string url =
+      std::string("http://") + sock.GetLocalInetAddress().GetHostAddress() + ":" + std::to_string(sock.GetLocalPort());
+
+   // First: do a PUT
+   const unsigned char putPayload[] = "put-data";
+   const std::size_t putPayloadLen = sizeof(putPayload) - 1;
+
+   std::string putHeaders;
+   std::string putBody;
+   std::thread threadRecvPut(TaskRecvPut, &sock, &putHeaders, &putBody);
+
+   ROOT::Internal::RCurlConnection conn(url);
+   auto putStatus = conn.SendPutReq(putPayload, putPayloadLen);
+
+   threadRecvPut.join();
+   EXPECT_TRUE(static_cast<bool>(putStatus));
+   EXPECT_EQ(0u, putHeaders.find("PUT "));
+
+   // Second: do a GET (SendRangesReq) on the same handle.
+   // The server sends a plain 200 response with the body "response-from-get".
+   const std::string expectedBody = "response-from-get";
+   std::string getHeaders;
+   auto taskRecvGet = [&](TServerSocket *serverSocket) {
+      getHeaders.clear();
+      auto s = serverSocket->Accept();
+
+      const char *eof = "\r\n\r\n";
+      const std::size_t eofLen = strlen(eof);
+      std::size_t nextInEof = 0;
+      char c;
+      while (s->RecvRaw(&c, 1)) {
+         getHeaders.push_back(c);
+         if (c == eof[nextInEof]) {
+            if (++nextInEof == eofLen)
+               break;
+         } else {
+            nextInEof = 0;
+         }
+      }
+
+      std::string response = "HTTP/1.1 200 OK\r\nContent-Length: " + std::to_string(expectedBody.size()) +
+                             "\r\n\r\n" + expectedBody;
+      s->SendRaw(response.data(), response.size());
+      s->Close();
+   };
+   std::thread threadRecvGet(taskRecvGet, &sock);
+
+   std::vector<unsigned char> readBuf(expectedBody.size(), 0);
+   ROOT::Internal::RCurlConnection::RUserRange range;
+   range.fDestination = readBuf.data();
+   range.fOffset = 0;
+   range.fLength = expectedBody.size();
+   auto getStatus = conn.SendRangesReq(1, &range);
+
+   threadRecvGet.join();
+   EXPECT_TRUE(static_cast<bool>(getStatus));
+   EXPECT_EQ(0u, getHeaders.find("GET "));
+   EXPECT_EQ(expectedBody.size(), range.fNBytesRecv);
+   std::string received(reinterpret_cast<char *>(readBuf.data()), range.fNBytesRecv);
+   EXPECT_EQ(expectedBody, received);
+}
+
+/// PUT with a payload larger than libcurl's internal Expect: 100-continue threshold (1 MB since curl 7.69).
+/// Verifies that the server-side 100 Continue handshake works and all bytes arrive correctly.
+TEST(RCurlConnection, PutLargeExpect100)
+{
+   TServerSocket sock(0, false, TServerSocket::kDefaultBacklog, -1, ESocketBindOption::kInaddrLoopback);
+   const std::string url =
+      std::string("http://") + sock.GetLocalInetAddress().GetHostAddress() + ":" + std::to_string(sock.GetLocalPort());
+
+   // 2 MB payload with a known repeating pattern
+   const std::size_t payloadLen = 2 * 1024 * 1024;
+   std::vector<unsigned char> payload(payloadLen);
+   for (std::size_t i = 0; i < payloadLen; ++i)
+      payload[i] = static_cast<unsigned char>(i & 0xFF);
+
+   std::string headers;
+   std::string body;
+   std::thread threadRecv(TaskRecvPut, &sock, &headers, &body);
+
+   ROOT::Internal::RCurlConnection conn(url);
+   auto status = conn.SendPutReq(payload.data(), payloadLen);
+
+   threadRecv.join();
+   EXPECT_TRUE(static_cast<bool>(status));
+   EXPECT_EQ(0u, headers.find("PUT "));
+
+   std::string headersLower = ToLower(headers);
+   EXPECT_NE(std::string::npos, headersLower.find("expect: 100-continue"))
+      << "large upload should include Expect: 100-continue header";
+   EXPECT_NE(std::string::npos, headersLower.find("content-length: " + std::to_string(payloadLen)));
+   ASSERT_EQ(payloadLen, body.size());
+   EXPECT_EQ(0, memcmp(body.data(), payload.data(), payloadLen));
+}

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));
+   }
+}