CurlHandlePool.cc

// -*- mode: c++; c-basic-offset:4 -*-
 
// This file is part of the BES
 
// Copyright (c) 2018 OPeNDAP, Inc.
// Author: James Gallagher<jgallagher@opendap.org>
//
// This library is free software; you can redistribute it and/or
// modify it under the terms of the GNU Lesser General Public
// License as published by the Free Software Foundation; either
// version 2.1 of the License, or (at your option) any later version.
//
// This library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
// Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public
// License along with this library; if not, write to the Free Software
// Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
//
// You can contact OPeNDAP, Inc. at PO Box 112, Saunderstown, RI. 02874-0112.
 
#include "config.h"
 
#include <string>
#include <locale>
#include <sstream>
 
#include <cstring>
#include <unistd.h>
 
#include <curl/curl.h>
 
#include "CurlUtils.h"
#include "HttpNames.h"
 
#include <time.h>
 
#include <libdap/util.h>   // long_to_string()
 
#include "BESLog.h"
#include "BESDebug.h"
#include "BESInternalError.h"
#include "BESForbiddenError.h"
#include "AllowedHosts.h"
 
#include "DmrppCommon.h"
#include "DmrppNames.h"
#include "awsv4.h"
#include "CurlHandlePool.h"
#include "Chunk.h"
#include "CredentialsManager.h"
#include "AccessCredentials.h"
 
#define KEEP_ALIVE 1   // Reuse libcurl easy handles (1) or not (0).
#define CURL_VERBOSE 0  // Logs curl info to the bes.log
 
#define prolog std::string("CurlHandlePool::").append(__func__).append("() - ")
 
using namespace dmrpp;
using namespace std;
 
string pthread_error(unsigned int err){
    string error_msg;
    switch(err){
        case EINVAL:
            error_msg = "The mutex was either created with the "
                        "protocol attribute having the value "
                        "PTHREAD_PRIO_PROTECT and the calling "
                        "thread's priority is higher than the "
                        "mutex's current priority ceiling."
                        "OR The value specified by mutex does not "
                        "refer to an initialized mutex object.";
            break;
 
        case EBUSY:
            error_msg = "The mutex could not be acquired "
                        "because it was already locked.";
            break;
 
        case EAGAIN:
            error_msg = "The mutex could not be acquired because "
                        "the maximum number of recursive locks "
                        "for mutex has been exceeded.";
            break;
 
        case EDEADLK:
            error_msg = "The current thread already owns the mutex";
            break;
 
        case EPERM:
            error_msg = "The current thread does not own the mutex.";
            break;
 
        default:
            error_msg = "Unknown pthread error type.";
            break;
    }
 
    return error_msg;
}
 
#if 0
static
string dump(const char *text, unsigned char *ptr, size_t size)
{
    size_t i;
    size_t c;
    unsigned int width=0x10;
 
    ostringstream oss;
    oss << text << ", " << std::setw(10) << (long)size << std::setbase(16) << (long)size << endl;
 
    for(i=0; i<size; i+= width) {
    oss << std::setw(4) << (long)i;
    // fprintf(stream, "%4.4lx: ", (long)i);
 
    /* show hex to the left */
    for(c = 0; c < width; c++) {
        if(i+c < size) {
        oss << std::setw(2) << ptr[i+c];
        //fprintf(stream, "%02x ", ptr[i+c]);
        }
        else {
        oss << "   ";
        // fputs("   ", stream);
        }
    }
 
    /* show data on the right */
    for(c = 0; (c < width) && (i+c < size); c++) {
        char x = (ptr[i+c] >= 0x20 && ptr[i+c] < 0x80) ? ptr[i+c] : '.';
        // fputc(x, stream);
        oss << std::setw(1) << x;
    }
 
    // fputc('\n', stream); /* newline */
    oss << endl;
    }
 
    return oss.str();
}
#endif
 
#if CURL_VERBOSE
static
int curl_trace(CURL */*handle*/, curl_infotype type, char *data, size_t /*size*/, void */*userp*/)
{
    string text = "";
    switch (type) {
        // print info
        case CURLINFO_TEXT:
        case CURLINFO_HEADER_OUT:
        case CURLINFO_HEADER_IN: {
            text = data;
            size_t pos;
            while ((pos = text.find('\n')) != string::npos)
                text = text.substr(0, pos);
            break;
        }
 
        // Do not build up 'text' for the data transfers
        case CURLINFO_DATA_OUT:
        case CURLINFO_SSL_DATA_OUT:
        case CURLINFO_DATA_IN:
        case CURLINFO_SSL_DATA_IN:
        default: /* in case a new one is introduced to shock us */
            break;
    }
 
    switch (type) {
        // print info
        case CURLINFO_TEXT:
            LOG("libcurl == Info: " << text << endl);
            break;
 
        case CURLINFO_HEADER_OUT:
            LOG("libcurl == Send header: " << text << endl);
            break;
        case CURLINFO_HEADER_IN:
            LOG("libcurl == Recv header: " << text << endl);
            break;
 
        // Only print these if we're desperate and the above code has been hacked to match
        case CURLINFO_DATA_OUT:
        case CURLINFO_SSL_DATA_OUT:
        case CURLINFO_DATA_IN:
        case CURLINFO_SSL_DATA_IN:
        default:
            break;
     }
 
     return 0;
}
#endif
 
 // FIXME - This code does not make a cURL handle that follows links and I think that's a bug!
dmrpp_easy_handle::dmrpp_easy_handle() : d_url(nullptr), d_request_headers(nullptr) {
 
    CURLcode res;
 
    d_handle = curl_easy_init();
    if (!d_handle) throw BESInternalError("Could not allocate CURL handle", __FILE__, __LINE__);
 
    curl::set_error_buffer(d_handle, d_errbuf);
 
    res =  curl_easy_setopt(d_handle, CURLOPT_SSLVERSION, CURL_SSLVERSION_TLSv1_2);
    curl::eval_curl_easy_setopt_result(res, prolog, "CURLOPT_SSLVERSION", d_errbuf, __FILE__, __LINE__);
 
 
#if CURL_VERBOSE
    res = curl_easy_setopt(d_handle, CURLOPT_DEBUGFUNCTION, curl_trace);
    curl::check_setopt_result(res, prolog, "CURLOPT_DEBUGFUNCTION", d_errbuf, __FILE__, __LINE__);
   // Many tests fail with this option, but it's still useful to see how connections
   // are treated. jhrg 10/2/18
   res = curl_easy_setopt(d_handle, CURLOPT_VERBOSE, 1L);
   curl::check_setopt_result(res, prolog, "CURLOPT_VERBOSE", d_errbuf, __FILE__, __LINE__);
#endif
 
    res = curl_easy_setopt(d_handle, CURLOPT_HEADERFUNCTION, chunk_header_callback);
    curl::eval_curl_easy_setopt_result(res, prolog, "CURLOPT_HEADERFUNCTION", d_errbuf, __FILE__, __LINE__);
 
    // Pass all data to the 'write_data' function
    res = curl_easy_setopt(d_handle, CURLOPT_WRITEFUNCTION, chunk_write_data);
    curl::eval_curl_easy_setopt_result(res, prolog, "CURLOPT_WRITEFUNCTION", d_errbuf, __FILE__, __LINE__);
 
#ifdef CURLOPT_TCP_KEEPALIVE
    /* enable TCP keep-alive for this transfer */
    res = curl_easy_setopt(d_handle, CURLOPT_TCP_KEEPALIVE, 1L);
    curl::check_setopt_result(res, prolog, "CURLOPT_TCP_KEEPALIVE", d_errbuf, __FILE__, __LINE__);
#endif
 
#ifdef CURLOPT_TCP_KEEPIDLE
    /* keep-alive idle time to 120 seconds */
     res = curl_easy_setopt(d_handle, CURLOPT_TCP_KEEPIDLE, 120L);
    curl::check_setopt_result(res, prolog, "CURLOPT_TCP_KEEPIDLE", d_errbuf, __FILE__, __LINE__);
#endif
 
#ifdef CURLOPT_TCP_KEEPINTVL
    /* interval time between keep-alive probes: 120 seconds */
    res = curl_easy_setopt(d_handle, CURLOPT_TCP_KEEPINTVL, 120L)
    curl::check_setopt_result(res, prolog, "CURLOPT_TCP_KEEPINTVL", d_errbuf, __FILE__, __LINE__);
#endif
 
    d_in_use = false;
    d_chunk = 0;
}
 
dmrpp_easy_handle::~dmrpp_easy_handle() {
    if (d_handle) curl_easy_cleanup(d_handle);
    if (d_request_headers) curl_slist_free_all(d_request_headers);
}
 
void dmrpp_easy_handle::read_data() {
    // Treat HTTP/S requests specially; retry some kinds of failures.
    if (d_url->protocol() == HTTPS_PROTOCOL || d_url->protocol() == HTTP_PROTOCOL) {
        curl::super_easy_perform(d_handle);
    }
    else {
        CURLcode curl_code = curl_easy_perform(d_handle);
        if (CURLE_OK != curl_code) {
            string msg = prolog + "ERROR - Data transfer error: ";
            throw BESInternalError(msg.append(curl::error_message(curl_code, d_errbuf)), __FILE__, __LINE__);
        }
    }
 
    d_chunk->set_is_read(true);
}
 
CurlHandlePool::CurlHandlePool(unsigned int max_handles) : d_max_easy_handles(max_handles) {
    for (unsigned int i = 0; i < d_max_easy_handles; ++i) {
        d_easy_handles.push_back(new dmrpp_easy_handle());
    }
}
 
dmrpp_easy_handle *
CurlHandlePool::get_easy_handle(Chunk *chunk) {
    // Here we check to make sure that we are only going to
    // access an approved location with this easy_handle
    // TODO I don't think this belongs here. jhrg 5/13/22
    string reason = "The requested resource does not match any of the AllowedHost rules.";
    if (!http::AllowedHosts::theHosts()->is_allowed(chunk->get_data_url(),reason)) {
        stringstream ss;
        ss << "ERROR! The chunk url "<< chunk->get_data_url()->str() << " was rejected because: " << reason;
        throw BESForbiddenError(ss.str(), __FILE__, __LINE__);
    }
 
    std::lock_guard<std::recursive_mutex> lock_me(d_get_easy_handle_mutex);
 
    dmrpp_easy_handle *handle = 0;
    for (auto i = d_easy_handles.begin(), e = d_easy_handles.end(); i != e; ++i) {
        if (!(*i)->d_in_use) {
            handle = *i;
            break;
        }
    }
 
    if (handle) {
        // Once here, d_easy_handle holds a CURL* we can use.
        handle->d_in_use = true;
        handle->d_url = chunk->get_data_url();
 
        handle->d_chunk = chunk;
 
        CURLcode res = curl_easy_setopt(handle->d_handle, CURLOPT_URL, chunk->get_data_url()->str().c_str());
        curl::eval_curl_easy_setopt_result(res, prolog, "CURLOPT_URL", handle->d_errbuf, __FILE__, __LINE__);
 
        // get the offset to offset + size bytes
        res = curl_easy_setopt(handle->d_handle, CURLOPT_RANGE, chunk->get_curl_range_arg_string().c_str());
        curl::eval_curl_easy_setopt_result(res, prolog, "CURLOPT_RANGE", handle->d_errbuf, __FILE__, __LINE__);
 
        // Pass this to chunk_header_callback as the fourth argument
        res = curl_easy_setopt(handle->d_handle, CURLOPT_HEADERDATA, reinterpret_cast<void *>(chunk));
        curl::eval_curl_easy_setopt_result(res, prolog, "CURLOPT_HEADERDATA", handle->d_errbuf, __FILE__, __LINE__);
 
        // Pass this to chunk_write_data as the fourth argument
        res = curl_easy_setopt(handle->d_handle, CURLOPT_WRITEDATA, reinterpret_cast<void *>(chunk));
        curl::eval_curl_easy_setopt_result(res, prolog, "CURLOPT_WRITEDATA", handle->d_errbuf, __FILE__, __LINE__);
 
        // store the easy_handle so that we can call release_handle in multi_handle::read_data()
        res = curl_easy_setopt(handle->d_handle, CURLOPT_PRIVATE, reinterpret_cast<void *>(handle));
        curl::eval_curl_easy_setopt_result(res, prolog, "CURLOPT_PRIVATE", handle->d_errbuf, __FILE__, __LINE__);
 
        // Enabled cookies
        res = curl_easy_setopt(handle->d_handle, CURLOPT_COOKIEFILE, curl::get_cookie_filename().c_str());
        curl::eval_curl_easy_setopt_result(res, prolog, "CURLOPT_COOKIEFILE", handle->d_errbuf, __FILE__, __LINE__);
 
        res = curl_easy_setopt(handle->d_handle, CURLOPT_COOKIEJAR, curl::get_cookie_filename().c_str());
        curl::eval_curl_easy_setopt_result(res, prolog, "CURLOPT_COOKIEJAR", handle->d_errbuf, __FILE__, __LINE__);
 
        // Follow 302 (redirect) responses
        res = curl_easy_setopt(handle->d_handle, CURLOPT_FOLLOWLOCATION, 1);
        curl::eval_curl_easy_setopt_result(res, prolog, "CURLOPT_FOLLOWLOCATION", handle->d_errbuf, __FILE__, __LINE__);
 
        res = curl_easy_setopt(handle->d_handle, CURLOPT_MAXREDIRS, curl::max_redirects());
        curl::eval_curl_easy_setopt_result(res, prolog, "CURLOPT_MAXREDIRS", handle->d_errbuf, __FILE__, __LINE__);
 
        // Set the user agent something otherwise TEA will never redirect to URS.
        res = curl_easy_setopt(handle->d_handle, CURLOPT_USERAGENT, curl::hyrax_user_agent().c_str());
        curl::eval_curl_easy_setopt_result(res, prolog, "CURLOPT_USERAGENT", handle->d_errbuf, __FILE__, __LINE__);
 
        // This means libcurl will use Basic, Digest, GSS Negotiate, or NTLM,
        // choosing the the 'safest' one supported by the server.
        // This requires curl 7.10.6 which is still in pre-release. 07/25/03 jhrg
        res = curl_easy_setopt(handle->d_handle, CURLOPT_HTTPAUTH, (long) CURLAUTH_ANY);
        curl::eval_curl_easy_setopt_result(res, prolog, "CURLOPT_HTTPAUTH", handle->d_errbuf, __FILE__, __LINE__);
 
        // Enable using the .netrc credentials file.
        res = curl_easy_setopt(handle->d_handle, CURLOPT_NETRC, CURL_NETRC_OPTIONAL);
        curl::eval_curl_easy_setopt_result(res, prolog, "CURLOPT_NETRC", handle->d_errbuf, __FILE__, __LINE__);
 
        // If the configuration specifies a particular .netrc credentials file, use it.
        // TODO move this operation into constructor and stash the value.
        string netrc_file = curl::get_netrc_filename();
        if (!netrc_file.empty()) {
            res = curl_easy_setopt(handle->d_handle, CURLOPT_NETRC_FILE, netrc_file.c_str());
            curl::eval_curl_easy_setopt_result(res, prolog, "CURLOPT_NETRC_FILE", handle->d_errbuf, __FILE__, __LINE__);
        }
 
        AccessCredentials *credentials = CredentialsManager::theCM()->get(handle->d_url);
        if (credentials && credentials->is_s3_cred()) {
            BESDEBUG(DMRPP_CURL,
                     prolog << "Got AccessCredentials instance: " << endl << credentials->to_json() << endl);
            // If there are available credentials, and they are S3 credentials then we need to sign
            // the request
            const std::time_t request_time = std::time(0);
 
            const std::string auth_header =
                    AWSV4::compute_awsv4_signature(
                            handle->d_url,
                            request_time,
                            credentials->get(AccessCredentials::ID_KEY),
                            credentials->get(AccessCredentials::KEY_KEY),
                            credentials->get(AccessCredentials::REGION_KEY),
                            "s3");
 
 
            handle->d_request_headers = curl::append_http_header((curl_slist *)0, "Authorization", auth_header);
            handle->d_request_headers = curl::append_http_header(handle->d_request_headers, "x-amz-content-sha256",
                                                  "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855");
            handle->d_request_headers = curl::append_http_header(handle->d_request_headers, "x-amz-date", AWSV4::ISO8601_date(request_time));
 
            res = curl_easy_setopt(handle->d_handle, CURLOPT_HTTPHEADER, handle->d_request_headers);
            curl::eval_curl_easy_setopt_result(res, prolog, "CURLOPT_HTTPHEADER", handle->d_errbuf, __FILE__, __LINE__);
        }
    }
 
    return handle;
}
 
void CurlHandlePool::release_handle(dmrpp_easy_handle *handle) {
    // In get_easy_handle, it's possible that d_in_use could be false and d_chunk
    // could not be set to 0 (because a separate thread could be running these
    // methods). In that case, the thread running get_easy_handle could set d_chunk,
    // and then this thread could clear it (... unlikely, but an optimizing compiler is
    // free to reorder statements so long as they don't alter the function's behavior).
    // Timing tests indicate this lock does not cost anything that can be measured.
    // jhrg 8/21/18
    std::lock_guard<std::recursive_mutex> lock_me(d_get_easy_handle_mutex);
 
    // TODO Add a call to curl reset() here. jhrg 9/23/20
 
#if KEEP_ALIVE
    handle->d_url = nullptr;
    handle->d_chunk = 0;
    handle->d_in_use = false;
#else
    // This is to test the effect of libcurl Keep Alive support
    // Find the handle; erase from the vector; delete; allocate a new handle and push it back on
    for (std::vector<dmrpp_easy_handle *>::iterator i = d_easy_handles.begin(), e = d_easy_handles.end(); i != e; ++i) {
        if (*i == handle) {
            BESDEBUG("dmrpp:5", "Found a handle match for the " << i - d_easy_handles.begin() << "th easy handle." << endl);
            delete handle;
            *i = new dmrpp_easy_handle();
            break;
        }
    }
#endif
}
 
void CurlHandlePool::release_handle(Chunk *chunk) {
    for (std::vector<dmrpp_easy_handle *>::iterator i = d_easy_handles.begin(), e = d_easy_handles.end(); i != e; ++i) {
        if ((*i)->d_chunk == chunk) {
            release_handle(*i);
            break;
        }
    }
}
 
void CurlHandlePool::release_all_handles() {
    for (std::vector<dmrpp_easy_handle *>::iterator i = d_easy_handles.begin(), e = d_easy_handles.end(); i != e; ++i) {
        release_handle(*i);
    }
}