9 #include "XrdVersion.hh"
16 #include <curl/curl.h>
36 uint64_t TPCHandler::m_monid{0};
37 int TPCHandler::m_marker_period = 5;
38 size_t TPCHandler::m_block_size = 16*1024*1024;
39 size_t TPCHandler::m_small_block_size = 1*1024*1024;
41 bool TPCHandler::allowMissingCRL =
false;
49 TPCHandler::TPCLogRecord::~TPCLogRecord()
56 monInfo.
clID = clID.c_str();
58 gettimeofday(&monInfo.
endT, 0);
61 {monInfo.
dstURL = local.c_str();
62 monInfo.
srcURL = remote.c_str();
64 monInfo.
dstURL = remote.c_str();
65 monInfo.
srcURL = local.c_str();
69 if (!status) monInfo.
endRC = 0;
70 else if (tpc_status > 0) monInfo.
endRC = tpc_status;
71 else monInfo.
endRC = 1;
72 monInfo.
strm =
static_cast<unsigned char>(streams);
73 monInfo.
fSize = (bytes_transferred < 0 ? 0 : bytes_transferred);
76 tpcMonitor->Report(monInfo);
86 if (curl) curl_easy_cleanup(curl);
101 int TPCHandler::sockopt_callback(
void *clientp, curl_socket_t curlfd, curlsocktype purpose) {
102 TPCLogRecord * rec = (TPCLogRecord *)clientp;
103 if (purpose == CURLSOCKTYPE_IPCXN && rec && rec->pmarkManager.isEnabled()) {
106 return CURL_SOCKOPT_ALREADY_CONNECTED;
108 return CURL_SOCKOPT_OK;
120 int TPCHandler::opensocket_callback(
void *clientp,
121 curlsocktype purpose,
122 struct curl_sockaddr *aInfo)
126 if (purpose != CURLSOCKTYPE_IPCXN)
127 return CURL_SOCKET_BAD;
130 return CURL_SOCKET_BAD;
133 int fd = XrdSysFD_Socket(aInfo->family, aInfo->socktype, aInfo->protocol);
136 return CURL_SOCKET_BAD;
143 TPCLogRecord *rec =
static_cast<TPCLogRecord*
>(clientp);
146 if ((!rec->allow_private && thePeer.isPrivate()) || (!rec->allow_local && thePeer.isLocal())) {
147 rec->tpc_status = 403;
148 rec->m_log->Emsg(rec->log_prefix.c_str(),
149 "Connection to local/private address is forbidden");
151 return CURL_SOCKET_BAD;
156 std::stringstream connectErrMsg;
157 if(!rec->pmarkManager.connect(fd, &(aInfo->addr), aInfo->addrlen, CONNECT_TIMEOUT, connectErrMsg)) {
158 rec->m_log->Emsg(rec->log_prefix.c_str(),
"Unable to connect socket: ", connectErrMsg.str().c_str());
160 return CURL_SOCKET_BAD;
166 int TPCHandler::closesocket_callback(
void *clientp, curl_socket_t fd) {
167 TPCLogRecord * rec = (TPCLogRecord *)clientp;
172 rec->pmarkManager.endPmark(fd);
187 int TPCHandler::ssl_ctx_callback(
CURL *curl,
void *ssl_ctx,
void *clientp) {
189 SSL_CTX* ctx =
static_cast<SSL_CTX*
>(ssl_ctx);
190 SSL_CTX_set_verify(ctx, SSL_VERIFY_PEER, verify_callback);
191 return CURL_SOCKOPT_OK;
194 int TPCHandler::verify_callback(
int preverify_ok, X509_STORE_CTX* ctx) {
195 if (preverify_ok == 1)
return 1;
197 int err = X509_STORE_CTX_get_error(ctx);
199 if (err == X509_V_ERR_UNABLE_TO_GET_CRL) {
200 X509_STORE_CTX_set_error(ctx, X509_V_OK);
217 bool TPCHandler::mismatchReprDigest(
const std::map<std::string, std::string> & passiveSrvReprDigest,
XrdHttpExtReq &req,
219 if(passiveSrvReprDigest.size()) {
220 for (
const auto & [digestName, digestValue]: passiveSrvReprDigest) {
221 auto clientDigestMatch = req.
mReprDigest.find(digestName);
224 if (clientDigestMatch->second != digestValue) {
226 std::stringstream errMsg;
227 errMsg <<
"Mismatch between client-provided and remote server checksums:"
228 <<
" client = (" << clientDigestMatch->first <<
"=" << clientDigestMatch->second <<
")"
229 <<
" server = (" << digestName <<
"=" << digestValue <<
")";
230 logTransferEvent(
LogMask::Error, rec,
"REPRDIGEST_VERIFY_FAIL", errMsg.str());
232 req.
SendSimpleResp(rec.status, NULL, NULL, generateClientErr(errMsg, rec, CURLcode::CURLE_OK).c_str(), 0);
252 std::stringstream parser(opaque);
253 std::string sequence;
254 std::stringstream output;
256 while (
getline(parser, sequence,
'&')) {
257 if (sequence.empty()) {
continue;}
258 size_t equal_pos = sequence.find(
'=');
260 if (equal_pos != std::string::npos)
261 val = curl_easy_escape(curl, sequence.c_str() + equal_pos + 1, sequence.size() - equal_pos - 1);
263 if (!val && equal_pos != std::string::npos) {
continue;}
265 if (!first) output <<
"&";
267 output << sequence.substr(0, equal_pos);
269 output <<
"=" << val;
281 TPCHandler::ConfigureCurlCA(
CURL *curl)
283 auto ca_filename = m_ca_file ? m_ca_file->CAFilename() :
"";
284 auto crl_filename = m_ca_file ? m_ca_file->CRLFilename() :
"";
285 if (!ca_filename.empty() && !crl_filename.empty()) {
286 curl_easy_setopt(curl, CURLOPT_CAINFO, ca_filename.c_str());
290 std::ifstream in(crl_filename, std::ifstream::ate | std::ifstream::binary);
291 if(in.tellg() > 0 && m_ca_file->atLeastOneValidCRLFound()){
292 curl_easy_setopt(curl, CURLOPT_CRLFILE, crl_filename.c_str());
293 if (allowMissingCRL) {
295 curl_easy_setopt(curl, CURLOPT_SSL_CTX_FUNCTION, ssl_ctx_callback);
298 std::ostringstream oss;
299 oss <<
"No valid CRL file has been found in the file " << crl_filename <<
". Disabling CRL checking.";
300 m_log.
Log(
Warning,
"TpcHandler",oss.str().c_str());
303 else if (!m_cadir.empty()) {
304 curl_easy_setopt(curl, CURLOPT_CAPATH, m_cadir.c_str());
306 if (!m_cafile.empty()) {
307 curl_easy_setopt(curl, CURLOPT_CAINFO, m_cafile.c_str());
313 return !strcmp(verb,
"COPY") || !strcmp(verb,
"OPTIONS");
322 const std::string replace_schemes[] = {
"davs://",
"s3://",
"s3s://" };
324 for (
const auto& s : replace_schemes)
325 if (url.compare(0, s.size(), s) == 0)
326 return "https://" + url.substr(s.size());
333 const std::string allowed_schemes[] = {
"https://",
"http://" };
335 for (
const auto& s : allowed_schemes)
336 if (url.compare(0, s.size(), s) == 0)
347 if (req.
verb ==
"OPTIONS") {
348 return ProcessOptionsReq(req);
351 if (header != req.
headers.end()) {
352 if (header->second !=
"none") {
353 m_log.
Emsg(
"ProcessReq",
"COPY requested an unsupported credential type: ", header->second.c_str());
354 return req.
SendSimpleResp(400, NULL, NULL,
"COPY requestd an unsupported Credential type", 0);
358 if (header != req.
headers.end()) {
361 const char *error_src =
"COPY rejected: disallowed scheme in source URL";
362 m_log.
Emsg(
"ProcessReq", error_src, src.c_str());
365 return ProcessPullReq(src, req);
368 if (header != req.
headers.end()) {
369 const std::string& dst = header->second;
371 const char *error_dst =
"COPY rejected: disallowed scheme in destination URL";
372 m_log.
Emsg(
"ProcessReq", error_dst, dst.c_str());
375 return ProcessPushReq(header->second, req);
377 m_log.
Emsg(
"ProcessReq",
"COPY verb requested but no source or destination specified.");
378 return req.
SendSimpleResp(400, NULL, NULL,
"No Source or Destination specified", 0);
394 m_allow_local(false),
395 m_allow_private(true),
397 m_fixed_route(false),
399 m_first_timeout(120),
400 m_log(log->logger(),
"TPC_"),
403 if (!Configure(config, myEnv)) {
404 throw std::runtime_error(
"Failed to configure the HTTP third-party-copy handler.");
422 return req.
SendSimpleResp(200, NULL, (
char *)
"DAV: 1\r\nDAV: <http://apache.org/dav/propset/fs/1>\r\nAllow: HEAD,GET,PUT,PROPFIND,DELETE,OPTIONS,COPY", NULL, 0);
432 if (authz_header != req.
headers.end()) {
433 std::stringstream ss;
434 ss <<
"authz=" <<
encode_str(authz_header->second);
444 int TPCHandler::RedirectTransfer(
CURL *curl,
const std::string &redirect_resource,
449 if ((ptr == NULL) || (*ptr ==
'\0') || (port == 0)) {
451 std::stringstream ss;
452 ss <<
"Internal error: redirect without hostname";
453 logTransferEvent(
LogMask::Error, rec,
"REDIRECT_INTERNAL_ERROR", ss.str());
454 return req.
SendSimpleResp(rec.status, NULL, NULL, generateClientErr(ss, rec).c_str(), 0);
458 std::string rdr_info = ptr;
459 std::string host, opaque;
460 size_t pos = rdr_info.find(
'?');
461 host = rdr_info.substr(0, pos);
463 if (pos != std::string::npos) {
464 opaque = rdr_info.substr(pos + 1);
467 std::stringstream ss;
468 ss <<
"Location: http" << (m_desthttps ?
"s" :
"") <<
"://" << host <<
":" << port <<
"/" << redirect_resource;
470 if (!opaque.empty()) {
476 return req.
SendSimpleResp(rec.status, NULL,
const_cast<char *
>(ss.str().c_str()),
484 int TPCHandler::OpenWaitStall(
XrdSfsFile &fh,
const std::string &resource,
486 const std::string &authz)
493 size_t pos = resource.find(
'?');
495 std::string path = resource.substr(0, pos);
497 if (pos != std::string::npos) {
498 opaque = resource.substr(pos + 1);
503 opaque += (opaque.empty() ?
"" :
"&");
506 open_result = fh.
open(path.c_str(), mode, openMode, &sec, opaque.c_str());
510 if (open_result ==
SFS_STARTED) {secs_to_stall = secs_to_stall/2 + 5;}
511 std::this_thread::sleep_for (std::chrono::seconds(secs_to_stall));
528 bool &success, TPCLogRecord &rec,
bool shouldReturnErrorToClient) {
530 curl_easy_setopt(curl, CURLOPT_NOBODY, 1);
532 curl_easy_setopt(curl, CURLOPT_TIMEOUT, CONNECT_TIMEOUT);
534 res = curl_easy_perform(curl);
537 curl_easy_setopt(curl, CURLOPT_NOBODY, 0);
539 curl_easy_setopt(curl, CURLOPT_TIMEOUT, 0L);
540 curl_easy_setopt(curl, CURLOPT_FAILONERROR,
true);
542 std::stringstream ss;
545 res = CURLE_HTTP_RETURNED_ERROR;
547 if (res != CURLE_OK) {
548 ss << curl_easy_strerror(res);
550 case CURLE_HTTP_RETURNED_ERROR:
552 ss <<
": remote host returned '" << rec.tpc_status <<
" "
555 case CURLE_COULDNT_CONNECT:
556 switch (rec.tpc_status) {
558 ss <<
": connection to local/private addresses is forbidden";
561 ss <<
": internal server failure";
562 rec.tpc_status = 500;
566 rec.tpc_status = 500;
571 if (rec.tpc_status >= 400) {
573 return shouldReturnErrorToClient ? req.
SendSimpleResp(rec.tpc_status, NULL, NULL, generateClientErr(ss, rec, res).c_str(), 0) : -1;
577 ss <<
"Successfully determined remote file information for pull request: "
580 unsigned int cksumIndex = 1;
581 for(
const auto & [cksumType,cksumValue]: state.
GetReprDigest()) {
582 ss <<
" chksum" << cksumIndex <<
"=(" << cksumType <<
"," << cksumValue <<
")";
590 int TPCHandler::GetRemoteFileInfoTPCPull(
CURL *curl,
XrdHttpExtReq &req, uint64_t &contentLength, std::map<std::string,std::string> & reprDigest,
bool & success, TPCLogRecord &rec) {
597 if ((result = PerformHEADRequest(curl, req, state, success, rec)) || !success) {
610 std::stringstream ss;
611 const std::string crlf =
"\n";
612 ss <<
"Perf Marker" << crlf;
613 ss <<
"Timestamp: " << time(NULL) << crlf;
614 ss <<
"Stripe Index: 0" << crlf;
616 ss <<
"Total Stripe Count: 1" << crlf;
621 ss <<
"RemoteConnections: " << desc << crlf;
626 return req.
ChunkResp(ss.str().c_str(), 0);
633 int TPCHandler::SendPerfMarker(
XrdHttpExtReq &req, TPCLogRecord &rec, std::vector<State*> &state,
634 off_t bytes_transferred)
648 std::stringstream ss;
649 const std::string crlf =
"\n";
650 ss <<
"Perf Marker" << crlf;
651 ss <<
"Timestamp: " << time(NULL) << crlf;
652 ss <<
"Stripe Index: 0" << crlf;
653 ss <<
"Stripe Bytes Transferred: " << bytes_transferred << crlf;
654 ss <<
"Total Stripe Count: 1" << crlf;
658 std::stringstream ss2;
659 for (std::vector<State*>::const_iterator iter = state.begin();
660 iter != state.end(); iter++)
662 std::string desc = (*iter)->GetConnectionDescription();
664 ss2 << (first ?
"" :
",") << desc;
669 ss <<
"RemoteConnections: " << ss2.str() << crlf;
671 rec.bytes_transferred = bytes_transferred;
674 return req.
ChunkResp(ss.str().c_str(), 0);
685 CURLM *multi_handle = curl_multi_init();
689 "Failed to initialize a libcurl multi-handle");
690 std::stringstream ss;
691 ss <<
"Failed to initialize internal server memory";
692 return req.
SendSimpleResp(rec.status, NULL, NULL, generateClientErr(ss, rec).c_str(), 0);
698 mres = curl_multi_add_handle(multi_handle, curl);
701 std::stringstream ss;
702 ss <<
"Failed to add transfer to libcurl multi-handle: HTTP library failure=" << curl_multi_strerror(mres);
703 logTransferEvent(
LogMask::Error, rec,
"CURL_INIT_FAIL", ss.str());
704 curl_multi_cleanup(multi_handle);
705 return req.
SendSimpleResp(rec.status, NULL, NULL, generateClientErr(ss, rec).c_str(), 0);
711 curl_multi_cleanup(multi_handle);
713 "Failed to send the initial response to the TPC client");
717 "Initial transfer response sent to the TPC client");
722 int running_handles = 1;
723 time_t last_marker = 0;
725 off_t last_advance_bytes = 0;
726 time_t last_advance_time = time(NULL);
727 time_t transfer_start = last_advance_time;
728 CURLcode res =
static_cast<CURLcode
>(-1);
730 time_t now = time(NULL);
731 time_t next_marker = last_marker + m_marker_period;
732 if (now >= next_marker) {
734 if (bytes_xfer > last_advance_bytes) {
735 last_advance_bytes = bytes_xfer;
736 last_advance_time = now;
738 if (SendPerfMarker(req, rec, state)) {
739 curl_multi_remove_handle(multi_handle, curl);
740 curl_multi_cleanup(multi_handle);
742 "Failed to send a perf marker to the TPC client");
745 int timeout = (transfer_start == last_advance_time) ? m_first_timeout : m_timeout;
746 if (now > last_advance_time + timeout) {
747 const char *log_prefix = rec.log_prefix.c_str();
748 bool tpc_pull = strncmp(
"Pull", log_prefix, 4) == 0;
751 std::stringstream ss;
752 ss <<
"Transfer failed because no bytes have been "
753 << (tpc_pull ?
"received from the source (pull mode) in "
754 :
"transmitted to the destination (push mode) in ") << timeout <<
" seconds.";
756 curl_multi_remove_handle(multi_handle, curl);
757 curl_multi_cleanup(multi_handle);
763 rec.pmarkManager.startTransfer();
764 mres = curl_multi_perform(multi_handle, &running_handles);
765 if (mres == CURLM_CALL_MULTI_PERFORM) {
769 }
else if (mres != CURLM_OK) {
771 }
else if (running_handles == 0) {
775 rec.pmarkManager.beginPMarks();
782 msg = curl_multi_info_read(multi_handle, &msgq);
783 if (msg && (msg->msg == CURLMSG_DONE)) {
784 CURL *easy_handle = msg->easy_handle;
785 res = msg->data.result;
786 curl_multi_remove_handle(multi_handle, easy_handle);
790 int64_t max_sleep_time = next_marker - time(NULL);
791 if (max_sleep_time <= 0) {
795 mres = curl_multi_wait(multi_handle, NULL, 0, max_sleep_time*1000, &fd_count);
796 if (mres != CURLM_OK) {
799 }
while (running_handles);
801 if (mres != CURLM_OK) {
802 std::stringstream ss;
803 ss <<
"Internal libcurl multi-handle error: HTTP library failure=" << curl_multi_strerror(mres);
804 logTransferEvent(
LogMask::Error, rec,
"TRANSFER_CURL_ERROR", ss.str());
806 curl_multi_remove_handle(multi_handle, curl);
807 curl_multi_cleanup(multi_handle);
809 if ((retval = req.
ChunkResp(generateClientErr(ss, rec).c_str(), 0))) {
811 "Failed to send error message to the TPC client");
821 msg = curl_multi_info_read(multi_handle, &msgq);
822 if (msg && (msg->msg == CURLMSG_DONE)) {
823 CURL *easy_handle = msg->easy_handle;
824 res = msg->data.result;
825 curl_multi_remove_handle(multi_handle, easy_handle);
829 if (!state.
GetErrorCode() && res ==
static_cast<CURLcode
>(-1)) {
830 curl_multi_remove_handle(multi_handle, curl);
831 curl_multi_cleanup(multi_handle);
832 std::stringstream ss;
833 ss <<
"Internal state error in libcurl";
834 logTransferEvent(
LogMask::Error, rec,
"TRANSFER_CURL_ERROR", ss.str());
836 if ((retval = req.
ChunkResp(generateClientErr(ss, rec).c_str(), 0))) {
838 "Failed to send error message to the TPC client");
843 curl_multi_cleanup(multi_handle);
857 std::stringstream ss;
858 bool success =
false;
861 std::stringstream ss2;
862 ss2 <<
"Remote side failed with status code " << state.
GetStatusCode();
864 std::replace(err.begin(), err.end(),
'\n',
' ');
865 ss2 <<
"; error message: \"" << err <<
"\"";
867 logTransferEvent(
LogMask::Error, rec,
"TRANSFER_FAIL", ss2.str());
868 ss << generateClientErr(ss2, rec);
871 if (err.empty()) {err =
"(no error message provided)";}
872 else {std::replace(err.begin(), err.end(),
'\n',
' ');}
873 std::stringstream ss2;
874 ss2 <<
"Error when interacting with local filesystem: " << err;
875 logTransferEvent(
LogMask::Error, rec,
"TRANSFER_FAIL", ss2.str());
876 ss << generateClientErr(ss2, rec);
877 }
else if (res != CURLE_OK) {
878 std::stringstream ss2;
879 ss2 <<
"Internal transfer failure";
880 std::stringstream ss3;
881 ss3 << ss2.str() <<
": " << curl_easy_strerror(res);
882 logTransferEvent(
LogMask::Error, rec,
"TRANSFER_FAIL", ss3.str());
883 ss << generateClientErr(ss2, rec, res);
885 ss <<
"success: Created";
889 if ((retval = req.
ChunkResp(ss.str().c_str(), 0))) {
891 "Failed to send last update to remote client");
893 }
else if (success) {
904 int TPCHandler::ProcessPushReq(
const std::string & resource,
XrdHttpExtReq &req) {
906 rec.allow_local = m_allow_local;
907 rec.allow_private = m_allow_private;
908 rec.log_prefix =
"PushRequest";
910 rec.remote = resource;
914 if (name) rec.name = name;
915 logTransferEvent(
LogMask::Info, rec,
"PUSH_START",
"Starting a push request");
918 auto curl = curlPtr.get();
920 std::stringstream ss;
921 ss <<
"Failed to initialize internal transfer resources";
924 return req.
SendSimpleResp(rec.status, NULL, NULL, generateClientErr(ss, rec).c_str(), 0);
926 curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1);
927 curl_easy_setopt(curl, CURLOPT_HTTP_VERSION, (
long) CURL_HTTP_VERSION_1_1);
928 #if CURL_AT_LEAST_VERSION(7, 85, 0)
929 curl_easy_setopt(curl, CURLOPT_PROTOCOLS_STR,
"https,http");
930 curl_easy_setopt(curl, CURLOPT_REDIR_PROTOCOLS_STR,
"https,http");
932 long protocols = CURLPROTO_HTTP | CURLPROTO_HTTPS;
933 curl_easy_setopt(curl, CURLOPT_PROTOCOLS, protocols);
934 curl_easy_setopt(curl, CURLOPT_REDIR_PROTOCOLS, protocols);
936 curl_easy_setopt(curl, CURLOPT_OPENSOCKETFUNCTION, opensocket_callback);
937 curl_easy_setopt(curl, CURLOPT_OPENSOCKETDATA, &rec);
938 curl_easy_setopt(curl, CURLOPT_CLOSESOCKETFUNCTION, closesocket_callback);
939 curl_easy_setopt(curl, CURLOPT_SOCKOPTFUNCTION, sockopt_callback);
940 curl_easy_setopt(curl, CURLOPT_CLOSESOCKETDATA, &rec);
941 curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT, CONNECT_TIMEOUT);
944 std::string redirect_resource = req.
resource;
945 if (query_header != req.
headers.end()) {
946 redirect_resource = query_header->second;
950 uint64_t file_monid =
AtomicInc(m_monid);
952 std::unique_ptr<XrdSfsFile> fh(m_sfs->
newFile(name, file_monid));
955 std::stringstream ss;
956 ss <<
"Failed to initialize internal transfer file handle";
959 return req.
SendSimpleResp(rec.status, NULL, NULL, generateClientErr(ss, rec).c_str(), 0);
961 std::string full_url = prepareURL(req);
963 std::string authz = GetAuthz(req);
965 int open_results = OpenWaitStall(*fh, full_url,
SFS_O_RDONLY, 0644,
968 int result = RedirectTransfer(curl, redirect_resource, req, fh->
error, rec);
970 }
else if (
SFS_OK != open_results) {
972 std::stringstream ss;
974 if (msg == NULL) ss <<
"Failed to open local resource";
978 int resp_result = req.
SendSimpleResp(rec.status, NULL, NULL, generateClientErr(ss, rec).c_str(), 0);
982 ConfigureCurlCA(curl);
983 curl_easy_setopt(curl, CURLOPT_URL, resource.c_str());
985 Stream stream(std::move(fh), 0, 0, m_log);
989 return RunCurlWithUpdates(curl, req, state, rec);
996 int TPCHandler::ProcessPullReq(
const std::string &resource,
XrdHttpExtReq &req) {
998 rec.allow_local = m_allow_local;
999 rec.allow_private = m_allow_private;
1000 rec.log_prefix =
"PullRequest";
1002 rec.remote = resource;
1006 if (name) rec.name = name;
1007 logTransferEvent(
LogMask::Info, rec,
"PULL_START",
"Starting a pull request");
1010 auto curl = curlPtr.get();
1012 std::stringstream ss;
1013 ss <<
"Failed to initialize internal transfer resources";
1016 return req.
SendSimpleResp(rec.status, NULL, NULL, generateClientErr(ss, rec).c_str(), 0);
1034 std::string host_used;
1035 if (host_header != req.
headers.end()) {
1036 host_used = host_header->second;
1042 ip = (
char *)malloc(ip_size-1);
1045 memcpy(ip, buff+1, ip_size-2);
1049 curl_easy_setopt(curl, CURLOPT_INTERFACE, ip);
1051 curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1);
1052 curl_easy_setopt(curl, CURLOPT_HTTP_VERSION, (
long) CURL_HTTP_VERSION_1_1);
1053 #if CURL_AT_LEAST_VERSION(7, 85, 0)
1054 curl_easy_setopt(curl, CURLOPT_PROTOCOLS_STR,
"https,http");
1055 curl_easy_setopt(curl, CURLOPT_REDIR_PROTOCOLS_STR,
"https,http");
1057 long protocols = CURLPROTO_HTTP | CURLPROTO_HTTPS;
1058 curl_easy_setopt(curl, CURLOPT_PROTOCOLS, protocols);
1059 curl_easy_setopt(curl, CURLOPT_REDIR_PROTOCOLS, protocols);
1061 curl_easy_setopt(curl, CURLOPT_OPENSOCKETFUNCTION, opensocket_callback);
1062 curl_easy_setopt(curl, CURLOPT_OPENSOCKETDATA, &rec);
1063 curl_easy_setopt(curl, CURLOPT_SOCKOPTFUNCTION, sockopt_callback);
1064 curl_easy_setopt(curl, CURLOPT_SOCKOPTDATA , &rec);
1065 curl_easy_setopt(curl, CURLOPT_CLOSESOCKETFUNCTION, closesocket_callback);
1066 curl_easy_setopt(curl, CURLOPT_CLOSESOCKETDATA, &rec);
1067 curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT, CONNECT_TIMEOUT);
1068 std::unique_ptr<XrdSfsFile> fh(m_sfs->
newFile(name, m_monid++));
1070 std::stringstream ss;
1071 ss <<
"Failed to initialize internal transfer file handle";
1074 return req.
SendSimpleResp(rec.status, NULL, NULL, generateClientErr(ss, rec).c_str(), 0);
1077 std::string redirect_resource = req.
resource;
1078 if (query_header != req.
headers.end()) {
1079 redirect_resource = query_header->second;
1083 if ((overwrite_header == req.
headers.end()) || (overwrite_header->second ==
"T")) {
1089 if (streams_header != req.
headers.end()) {
1090 int stream_req = -1;
1092 stream_req = std::stol(streams_header->second);
1095 if (stream_req < 0 || stream_req > 100) {
1096 std::stringstream ss;
1097 ss <<
"Invalid request for number of streams";
1099 logTransferEvent(
LogMask::Info, rec,
"INVALID_REQUEST", ss.str());
1100 return req.
SendSimpleResp(rec.status, NULL, NULL, generateClientErr(ss, rec).c_str(), 0);
1102 streams = stream_req == 0 ? 1 : stream_req;
1105 rec.streams = streams;
1106 std::string full_url = prepareURL(req);
1107 std::string authz = GetAuthz(req);
1108 curl_easy_setopt(curl, CURLOPT_URL, resource.c_str());
1109 ConfigureCurlCA(curl);
1110 uint64_t sourceFileContentLength = 0;
1114 bool success =
false;
1115 bool mismatchDigests =
false;
1116 std::map<std::string,std::string> sourceFileReprDigest;
1117 GetRemoteFileInfoTPCPull(curl, req, sourceFileContentLength, sourceFileReprDigest, success, rec);
1121 full_url +=
"&oss.asize=" + std::to_string(sourceFileContentLength);
1122 mismatchDigests = mismatchReprDigest(sourceFileReprDigest,req,rec);
1124 if(!success || mismatchDigests) {
1131 int open_result = OpenWaitStall(*fh, full_url, mode|
SFS_O_WRONLY,
1135 int result = RedirectTransfer(curl, redirect_resource, req, fh->
error, rec);
1137 }
else if (
SFS_OK != open_result) {
1139 std::stringstream ss;
1141 if ((msg == NULL) || (*msg ==
'\0')) ss <<
"Failed to open local resource";
1146 generateClientErr(ss, rec).c_str(), 0);
1150 Stream stream(std::move(fh), streams * m_pipelining_multiplier, streams > 1 ? m_block_size : m_small_block_size, m_log);
1156 return RunCurlWithStreams(req, state, streams, rec);
1158 return RunCurlWithUpdates(curl, req, state, rec);
1166 void TPCHandler::logTransferEvent(
LogMask mask,
const TPCLogRecord &rec,
1167 const std::string &event,
const std::string &message)
1171 std::stringstream ss;
1172 ss <<
"event=" <<
event <<
", local=" << rec.local <<
", remote=" << rec.remote;
1173 if (rec.name.empty())
1174 ss <<
", user=(anonymous)";
1176 ss <<
", user=" << rec.name;
1177 if (rec.streams != 1)
1178 ss <<
", streams=" << rec.streams;
1179 if (rec.bytes_transferred >= 0)
1180 ss <<
", bytes_transferred=" << rec.bytes_transferred;
1181 if (rec.status >= 0)
1182 ss <<
", status=" << rec.status;
1183 if (rec.tpc_status >= 0)
1184 ss <<
", tpc_status=" << rec.tpc_status;
1185 if (!message.empty())
1186 ss <<
"; " << message;
1187 m_log.
Log(mask, rec.log_prefix.c_str(), ss.str().c_str());
1190 std::string TPCHandler::generateClientErr(std::stringstream &err_ss,
const TPCLogRecord &rec, CURLcode cCode) {
1191 std::stringstream ssret;
1192 ssret <<
"failure: " << err_ss.str() <<
", local=" << rec.local <<
", remote=" << rec.remote;
1193 if(cCode != CURLcode::CURLE_OK) {
1194 ssret <<
", HTTP library failure=" << curl_easy_strerror(cCode);
1205 if (curl_global_init(CURL_GLOBAL_DEFAULT)) {
1206 log->
Emsg(
"TPCInitialize",
"libcurl failed to initialize");
1212 log->
Emsg(
"TPCInitialize",
"TPC handler requires a config filename in order to load");
1216 log->
Emsg(
"TPCInitialize",
"Will load configuration for the TPC handler from", config);
1218 }
catch (std::runtime_error &re) {
1219 log->
Emsg(
"TPCInitialize",
"Encountered a runtime failure when loading ", re.what());
XrdVERSIONINFO(XrdHttpGetExtHandler, HttpTPC)
XrdHttpExtHandler * XrdHttpGetExtHandler(XrdSysError *log, const char *config, const char *, XrdOucEnv *myEnv)
static std::string PrepareURL(const std::string &url)
std::string encode_xrootd_opaque_to_uri(CURL *curl, const std::string &opaque)
static bool IsAllowedScheme(const std::string &url)
int mapErrNoToHttp(int errNo)
std::string httpStatusToString(int status)
Utility functions for XrdHTTP.
std::string encode_str(const std::string &str)
void getline(uchar *buff, int blen)
int GetStatusCode() const
off_t BytesTransferred() const
void SetErrorMessage(const std::string &error_msg)
std::string GetErrorMessage() const
std::string GetConnectionDescription()
void SetupHeaders(XrdHttpExtReq &req)
void SetContentLength(const off_t content_length)
off_t GetContentLength() const
void SetErrorCode(int error_code)
const std::map< std::string, std::string > & GetReprDigest() const
void SetupHeadersForHEAD(XrdHttpExtReq &req)
TPCHandler(XrdSysError *log, const char *config, XrdOucEnv *myEnv)
virtual int ProcessReq(XrdHttpExtReq &req)
virtual bool MatchesPath(const char *verb, const char *path)
Tells if the incoming path is recognized as one of the paths that have to be processed.
int ChunkResp(const char *body, long long bodylen)
Send a (potentially partial) body in a chunked response; invoking with NULL body.
void GetClientID(std::string &clid)
std::map< std::string, std::string > & headers
std::map< std::string, std::string > mReprDigest
Repr-Digest map where the key is the digest name and the value is the base64 encoded digest value.
int StartChunkedResp(int code, const char *desc, const char *header_to_add)
Starts a chunked response; body of request is sent over multiple parts using the SendChunkResp.
const XrdSecEntity & GetSecEntity() const
int SendSimpleResp(int code, const char *desc, const char *header_to_add, const char *body, long long bodylen)
Sends a basic response. If the length is < 0 then it is calculated internally.
static std::string prepareOpenURL(PrepareOpenURLParams ¶ms)
static const int noPort
Do not add port number.
int Format(char *bAddr, int bLen, fmtUse fmtType=fmtAuto, int fmtOpts=0)
@ fmtAddr
Address using suitable ipv4 or ipv6 format.
static const char * GetAddrs(const char *hSpec, XrdNetAddr *aListP[], int &aListN, AddrOpts opts=allIPMap, int pNum=PortInSpec)
void * GetPtr(const char *varname)
const char * getErrText()
void setUCap(int ucval)
Set user capabilties.
static std::map< std::string, T >::const_iterator caseInsensitiveFind(const std::map< std::string, T > &m, const std::string &lowerCaseSearchKey)
char * name
Entity's name.
virtual XrdSfsFile * newFile(char *user=0, int MonID=0)=0
virtual int open(const char *fileName, XrdSfsFileOpenMode openMode, mode_t createMode, const XrdSecEntity *client=0, const char *opaque=0)=0
int Emsg(const char *esfx, int ecode, const char *text1, const char *text2=0)
XrdSysLogger * logger(XrdSysLogger *lp=0)
void Log(int mask, const char *esfx, const char *text1, const char *text2=0, const char *text3=0)
std::unique_ptr< CURL, CurlDeleter > ManagedCurlHandle
void operator()(CURL *curl)
static const int uIPv64
ucap: Supports only IPv4 info