~drizzle-trunk/drizzle/development

/* Copyright (c) 2009 PrimeBase Technologies GmbH, Germany

 *

 * PrimeBase Media Stream for MySQL

 *

 * This program is free software; you can redistribute it and/or modify

 * it under the terms of the GNU General Public License as published by

 * the Free Software Foundation; either version 2 of the License, or

 * (at your option) any later version.

 *

 * This program 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 General Public License for more details.

 *

 * You should have received a copy of the GNU General Public License

 * along with this program; if not, write to the Free Software

 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA

 *

 *  Created by Barry Leslie on 10/02/09.

 *

 */

#include "CSConfig.h"

#include <inttypes.h>

#include <stdlib.h>

#include <curl/curl.h>

#include "CSGlobal.h"

#include "CSString.h"

#include "CSStrUtil.h"

#include "CSEncode.h"

#include "CSS3Protocol.h"

#include "CSXML.h"

#ifdef S3_UNIT_TEST

//#define SHOW_SIGNING

// Uncomment this line to trace network action during request. Very Usefull!!

//#define DEBUG_CURL

#define DUMP_ERRORS

#endif

//#define DEBUG_CURL

#define DUMP_ERRORS

//#define SHOW_SIGNING

#define HEX_CHECKSUM_VALUE_SIZE (2 *CHECKSUM_VALUE_SIZE)

#define THROW_CURL_IF(v) { if (v) CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, ms_curl_error);}

//-------------------------------

static const char *retryCodes[] = {

	"ExpiredToken",

	"InternalError",

	"OperationAborted",

	"RequestTimeout",

	"SlowDown",

	NULL

};

//======================================

static size_t receive_data(void *ptr, size_t size, size_t nmemb, void *stream);

static size_t receive_header(void *ptr, size_t size, size_t nmemb, void *stream);

static size_t send_callback(void *ptr, size_t size, size_t nmemb, void *stream);

class S3ProtocolCon : CSXMLBuffer, public CSObject {

	private:

	virtual bool openNode(char *path, char *value) {

		if (value && *value && (strcmp(path,"/error/code/") == 0)) {

			printf("S3 ERROR Code: %s\n", value);

			for (int i = 0; retryCodes[i] && !ms_retry; i++)

				ms_retry = (strcmp(value, retryCodes[i]) == 0);

			if (ms_retry && !strcmp("slowdown", value)) 

				ms_slowDown = true;

		} else if (value && *value && (strcmp(path,"/error/message/") == 0)) {

			printf("S3 ERROR MESSAGE: %s\n", value);

		}

		return true;

	}

	virtual bool closeNode(char *path) {

		(void)path;

		return true;

	}

	virtual bool addAttribute(char *path, char *name, char *value) {

		(void)path;

		(void)name;

		(void)value;

		return true;

	}

	//-------------------------------

	void parse_s3_error()

	{

		enter_();

		if (!ms_errorReply)

			CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "Missing HTTP reply: possible S3 connection failure.");

	#ifdef DUMP_ERRORS

		printf("ms_errorReply:\n===========\n%s\n===========\n", ms_errorReply->getCString());

	#endif

		if (!parseData(ms_errorReply->getCString(), ms_errorReply->length(), 0)){

			int		err;

			char	*msg;

			getError(&err, &msg);

			CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, msg);

		}

		exit_();

	}

	public:

	CSHTTPHeaders	ms_reply_headers;

	CSStringBuffer	ms_buffer; // A scratch buffer

	CURL			*ms_curl;	

	struct curl_slist	*ms_header_list;	// A curl list of headers to be sent with the next request.

	CSInputStream	*ms_inputStream;

	CSOutputStream	*ms_outputStream;

	CSMd5			ms_md5;

	char			ms_s3Checksum[HEX_CHECKSUM_VALUE_SIZE +1];

	bool			ms_notFound; // True if the object could not be found

	bool			ms_retry; // True if the request failed with a retry error.

	bool			ms_slowDown;

	CSStringBuffer	*ms_errorReply;

	char			ms_curl_error[CURL_ERROR_SIZE];

	off64_t			ms_data_size;

	unsigned int	ms_replyStatus;

	bool			ms_throw_error;	// Gets set if an exception occurs in a callback.

	bool			ms_old_libcurl;

	char			*ms_safe_url;

	S3ProtocolCon():

		ms_curl(NULL),

		ms_header_list(NULL),

		ms_inputStream(NULL),

		ms_outputStream(NULL),

		ms_notFound(false),

		ms_retry(false),

		ms_slowDown(false),

		ms_errorReply(NULL),

		ms_data_size(0),

		ms_replyStatus(0),

		ms_throw_error(false),

		ms_old_libcurl(false),

		ms_safe_url(NULL)

	{

		ms_curl = curl_easy_init();

		if (!ms_curl)

			CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "curl_easy_init() failed.");

		curl_version_info_data *curl_ver = curl_version_info(CURLVERSION_NOW); 

		// libCurl versions prior to 7.17.0 did not make copies of strings passed into curl_easy_setopt()

		// If this version requirement is a problem I can do this myself, if I have to, I guess. :(

		if (curl_ver->version_num < 0X071700 ) {

			ms_old_libcurl = true;

			//char msg[200];

			//snprintf(msg, 200, "libcurl version %s is too old, require version 7.17.0 or newer.", curl_ver->version);

			//CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, msg);

		}

		if (curl_easy_setopt(ms_curl, CURLOPT_ERRORBUFFER, ms_curl_error))

			CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "curl_easy_setopt(CURLOPT_ERRORBUFFER) failed.");

#ifdef DEBUG_CURL

		curl_easy_setopt(ms_curl, CURLOPT_VERBOSE, 1L);

#endif		

		//THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_TCP_NODELAY, 1L));

		THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_NOPROGRESS, 1L));

		THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_WRITEFUNCTION, receive_data));

		THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_READFUNCTION, send_callback));	

		THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_HEADERFUNCTION, receive_header));

		THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_WRITEDATA, this));

		THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_READDATA, this));

		THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_WRITEHEADER, this));

		THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_FOLLOWLOCATION, 1L)); // Follow redirects.

	}

	~S3ProtocolCon()

	{

		if (ms_curl) 

			curl_easy_cleanup(ms_curl);			

		if (ms_header_list) 

			curl_slist_free_all(ms_header_list);			

		if (ms_inputStream)

			ms_inputStream->release();

		if (ms_outputStream)

			ms_outputStream->release();

		if (ms_errorReply)

			ms_errorReply->release();

		ms_reply_headers.clearHeaders();

		if (ms_safe_url)

			cs_free(ms_safe_url);

	}

	inline void check_reply_status() 

	{

		if (ms_replyStatus > 199 && ms_replyStatus < 300)

			return;

		switch (ms_replyStatus) {

			case 200:

			case 204:	// No Content

			//case 301: // Moved Permanently

			//case 307: // Temporary Redirect

				break;

			case 404:	// Not Found

				ms_notFound = true;

				break;

			case 500:	

				ms_retry = true;

				break;

			default: {

				parse_s3_error();

				if (!ms_retry) {

					enter_();

					CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, ms_errorReply->getCString());

					outer_();

				} else if (ms_slowDown) {

					enter_();

					CSException::logException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "S3 slow down request.");

					self->sleep(10); // sleep for 1/100 second.

					outer_();

				}

			}

		}

	}

	inline void ms_reset()

	{

		// Remove any old headers

		if (ms_header_list) {

			curl_slist_free_all(ms_header_list);

			ms_header_list = NULL;

		}

		ms_reply_headers.clearHeaders();

		ms_replyStatus = 0;

		ms_throw_error = false;

		if (ms_errorReply)

			ms_errorReply->setLength(0);

		ms_s3Checksum[0] = 0;

		ms_notFound = false;

		ms_retry = false;

		if (ms_outputStream) {

			ms_outputStream->release();

			ms_outputStream = NULL;

		}

		if (ms_inputStream) {

			ms_inputStream->release();

			ms_inputStream = NULL;

		}

		if (ms_safe_url) {

			cs_free(ms_safe_url);

			ms_safe_url = NULL;

		}

	}

	inline void ms_setHeader(const char *header)

	{

		ms_header_list = curl_slist_append(ms_header_list, header);

		if (!ms_header_list) 

			CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "curl_slist_append() failed.");

	}

private:	

	inline const char *safe_url(const char *url)

	{

		if (ms_old_libcurl == false)

			return url;

		if (ms_safe_url) {

			cs_free(ms_safe_url);

			ms_safe_url = NULL;

		}

		ms_safe_url = cs_strdup(url);

		return ms_safe_url;

	}

public:	

	inline void ms_setURL(const char *url)

	{

		//printf("URL: \"%s\n", url);

		THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_URL, safe_url(url)));

	}

	inline void ms_execute_delete_request()

	{

		CURLcode rtc;

		THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_HTTPHEADER, ms_header_list));

		THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_CUSTOMREQUEST, "DELETE"));

		rtc = curl_easy_perform(ms_curl);

		THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_CUSTOMREQUEST, NULL)); // IMPORTANT: Reset this to it's default value

		if (rtc && !ms_throw_error)

			CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, ms_curl_error);

		if (ms_throw_error) {

			enter_();

			throw_();

			outer_();

		}

		check_reply_status();

	}

	inline void ms_execute_copy_request()

	{

		CURLcode rtc;

		THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_HTTPHEADER, ms_header_list));

		THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_INFILESIZE_LARGE, 0));

		THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_UPLOAD, 1L));

		rtc = curl_easy_perform(ms_curl);

		if (rtc && !ms_throw_error)

			CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, ms_curl_error);

		if (ms_throw_error) {

			enter_();

			throw_();

			outer_();

		}

		check_reply_status();

	}

	inline void ms_execute_get_request(CSOutputStream *output)

	{

		enter_();

		push_(output);	

		THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_HTTPHEADER, ms_header_list));

		THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_HTTPGET, 1L));

		ms_outputStream = output;	

		if (curl_easy_perform(ms_curl) && !ms_throw_error) {

			ms_outputStream = NULL;	

			CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, ms_curl_error);

		}

		ms_outputStream = NULL;	

		release_(output);	

		if (ms_throw_error) 

			throw_();

		check_reply_status();

		exit_();		

	}

	inline void ms_execute_put_request(CSInputStream *input, off64_t size, Md5Digest *digest)

	{

		enter_();

		push_(input);	

		THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_HTTPHEADER, ms_header_list));

		THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_INFILESIZE_LARGE, size));

		THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_UPLOAD, 1L));

		//THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_POSTFIELDSIZE_LARGE, size));

		//THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_POST, 1L));

		ms_md5.md5_init();

		ms_data_size = size;

		ms_inputStream = input;	

		if (curl_easy_perform(ms_curl) && !ms_throw_error) {

			ms_inputStream = NULL;	

			CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, ms_curl_error);

		}

		ms_inputStream = NULL;

		release_(input);	

		if (ms_throw_error)

			throw_();

		check_reply_status();

		// Check that the checksums agree

		char checksum[HEX_CHECKSUM_VALUE_SIZE +1];

		ms_md5.md5_digest(digest);

		cs_bin_to_hex(HEX_CHECKSUM_VALUE_SIZE, checksum, CHECKSUM_VALUE_SIZE, digest->val);

		checksum[HEX_CHECKSUM_VALUE_SIZE] = 0;

		cs_strToUpper(ms_s3Checksum);

		if (strcmp(checksum, ms_s3Checksum)) {

			// The request should be restarted in this case.

			ms_retry = true;

			CSException::logException(CS_CONTEXT, CS_ERR_CHECKSUM_ERROR, "Calculated checksum did not match S3 checksum");

		}

		exit_();		

	}

};

//======================================

//======================================

CSS3Protocol::CSS3Protocol():

	s3_server(NULL),

	s3_public_key(NULL),

	s3_private_key(NULL),

	s3_maxRetrys(5)

{

	new_(s3_server, CSStringBuffer());

	s3_server->append("s3.amazonaws.com/");

	s3_public_key = CSString::newString("");

	s3_private_key = CSString::newString("");

}

//------------------

CSS3Protocol::~CSS3Protocol()

{

	if (s3_server)

		s3_server->release();

	if (s3_public_key)

		s3_public_key->release();

	if (s3_private_key)

		s3_private_key->release();

}

//------------------

CSString *CSS3Protocol::s3_getSignature(const char *verb, 

										const char *md5, 

										const char *content_type, 

										const char *date, 

										const char *bucket, 

										const char *key,

										CSString *headers 

									)

{

	CSStringBuffer *s3_buffer;

	enter_();

	if (headers)

		push_(headers);

	new_(s3_buffer, CSStringBuffer());

	push_(s3_buffer);

	s3_buffer->setLength(0);

	s3_buffer->append(verb);	

	s3_buffer->append("\n");	

	if (md5) s3_buffer->append(md5);	

	s3_buffer->append("\n");	

	if (content_type) s3_buffer->append(content_type);	

	s3_buffer->append("\n");	

	s3_buffer->append(date);

	if (headers) { 

		// Note: headers are assumed to be in lower case, sorted, and containing no white space.

		s3_buffer->append("\n");	

		s3_buffer->append(headers->getCString());

	}

	s3_buffer->append("\n/");

	s3_buffer->append(bucket);

	s3_buffer->append("/");

	s3_buffer->append(key);

#ifdef SHOW_SIGNING

printf("signing:\n=================\n%s\n=================\n", 	s3_buffer->getCString());

if(0){

	const char *ptr = s3_buffer->getCString();

	while (*ptr) {

		printf("%x ", *ptr); ptr++;

	}

	printf("\n");

}

#endif

	CSString *sig = signature(s3_buffer->getCString(), s3_private_key->getCString());

	release_(s3_buffer);

	if (headers) 

		release_(headers);

	return_(sig);

}

//----------------------

// CURL callback functions:

////////////////////////////

//----------------------

static size_t send_callback(void *ptr, size_t objs, size_t obj_size, void *v_con)

{

	S3ProtocolCon *con = (S3ProtocolCon*) v_con;

	size_t data_sent, buffer_size = objs * obj_size;

	if (!con->ms_data_size)

		return 0;

	enter_();

	try_(a) {

		data_sent = con->ms_inputStream->read((char*)ptr, buffer_size);

		if (data_sent <= con->ms_data_size) {

			con->ms_data_size -= data_sent;

			if (data_sent)

				con->ms_md5.md5_append((u_char*)ptr, data_sent); // Calculating the checksum for the data sent.

		} else if (data_sent > con->ms_data_size) 

			CSException::RecordException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "Blob larger than expected.");

		else if (con->ms_data_size && !data_sent)

			CSException::RecordException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "Blob smaller than expected.");

	}

	catch_(a);

	con->ms_throw_error = true;

	data_sent = SIZE_MAX;

	cont_(a);

	return data_sent;

}

//----------------------

static size_t receive_data(void *vptr, size_t objs, size_t obj_size, void *v_con)

{

	S3ProtocolCon *con = (S3ProtocolCon*) v_con;

	size_t data_len = objs * obj_size;

	enter_();

	try_(a) {

		if (con->ms_replyStatus >= 400) { // Collect the error reply.

			if (!con->ms_errorReply)

				con->ms_errorReply = new CSStringBuffer(50);		

			con->ms_errorReply->append((char*)vptr, data_len);

		} else if (	con->ms_outputStream)

			con->ms_outputStream->write((char*)vptr, data_len);

	}

	catch_(a);

	con->ms_throw_error = true;

	data_len = SIZE_MAX; 

	cont_(a);

	return_(data_len);	

}

#define IS_REDIRECT(s) ((s >= 300) && (s < 400))

//----------------------

static size_t receive_header(void *header, size_t objs, size_t obj_size, void *v_con)

{

	S3ProtocolCon *con = (S3ProtocolCon*) v_con;

	size_t size = objs * obj_size;

	char *end, *ptr = (char*) header, *name, *value = NULL;

	uint32_t name_len =0, value_len = 0;

	CLOBBER_PROTECT(con);

	CLOBBER_PROTECT(size);

	CLOBBER_PROTECT(ptr);

	CLOBBER_PROTECT(value);

	CLOBBER_PROTECT(value_len);

	CLOBBER_PROTECT(name_len);

//printf(	"receive_header: %s\n", ptr);

	end = ptr + size;

	if (*(end -2) == '\r' && *(end -1) == '\n')

		end -=2;

	while ((end != ptr) && (*ptr == ' ')) ptr++;

	if (end == ptr)

		return size;

	// Get the reply status.

	// Status 100 = Continue

	if (((!con->ms_replyStatus) || (con->ms_replyStatus == 100) || IS_REDIRECT(con->ms_replyStatus) ) 

			&& !strncasecmp(ptr, "HTTP", 4)

		) {

		char status[4];

		while ((end != ptr) && (*ptr != ' ')) ptr++; // skip HTTP stuff

		while ((end != ptr) && (*ptr == ' ')) ptr++; // find the start of eh status code.

		if (end == ptr)

			return size;

		if (end < (ptr +3)) // expecting a 3 digit status code.

			return size;

		memcpy(status, ptr, 3);

		status[3] = 0;

		con->ms_replyStatus = atoi(status);

	}

	name = ptr;

	while ((end != ptr) && (*ptr != ':')) ptr++;

	if (end == ptr)

		return size;

	name_len = ptr - name;

	ptr++; 

	while ((end != ptr) && (*ptr == ' ')) ptr++;

	if (end == ptr)

		return size;

	value = ptr;

	value_len = end - value;

	while (name[name_len-1] == ' ') name_len--;

	while (value[value_len-1] == ' ') value_len--;

	if (!strncasecmp(name, "ETag", 4)) {

		value++; value_len -=2; // Strip quotation marks from checksum string.

		if (value_len == HEX_CHECKSUM_VALUE_SIZE) {

			memcpy(con->ms_s3Checksum, value, value_len);

			con->ms_s3Checksum[value_len] = 0;

		}

	}

	enter_();

	try_(a) {

		con->ms_reply_headers.addHeader(name, name_len, value, value_len);

	}

	catch_(a);

	con->ms_throw_error = true;

	size = (size_t)-1;

	cont_(a);

	return_(size);

}

//----------------------

#define SET_DATE_FROM_TIME(t, d) {strftime(d, sizeof(d), "%a, %d %b %Y %H:%M:%S GMT", gmtime(&t));}

#define SET_DATE(d) {time_t t = time(NULL); SET_DATE_FROM_TIME(t, d);}

bool CSS3Protocol::s3_delete(const char *bucket, const char *key)

{

 	CSStringBuffer *s3_buffer;

	char date[64];

	CSString *signed_str;

	uint32_t retry_count = 0;

	S3ProtocolCon *con_data;

	enter_();

	new_(s3_buffer, CSStringBuffer());

	push_(s3_buffer);

	new_(con_data, S3ProtocolCon());

	push_(con_data);

retry:

	// Clear old settings. 

	con_data->ms_reset();	

	SET_DATE(date);

	// Build the URL

	s3_buffer->setLength(0);

	s3_buffer->append("http://");

	s3_buffer->append(bucket);

	s3_buffer->append(".");	

	s3_buffer->append(s3_server->getCString());

	s3_buffer->append(key);

	con_data->ms_setURL(s3_buffer->getCString());

	// Add the 'DATE' header

	s3_buffer->setLength(0);

	s3_buffer->append("Date: ");	

	s3_buffer->append(date);

	con_data->ms_setHeader(s3_buffer->getCString());

	// Create the authentication signature and add the 'Authorization' header

	signed_str = s3_getSignature("DELETE", NULL, NULL, date, bucket, key);

	push_(signed_str);

	s3_buffer->setLength(0);

	s3_buffer->append("Authorization: AWS ");	

	s3_buffer->append(s3_public_key->getCString());

	s3_buffer->append(":");	

	s3_buffer->append(signed_str->getCString());	

	release_(signed_str); signed_str = NULL;

	con_data->ms_setHeader(s3_buffer->getCString());

	con_data->ms_execute_delete_request();

	if (con_data->ms_retry) {

		if (retry_count == s3_maxRetrys) {

			CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "S3 operation aborted after max retries.");

		}

	printf("RETRY: s3_delete()\n");

		retry_count++;

		goto retry;

	}

	bool notFound = con_data->ms_notFound;

	release_(con_data);

	release_(s3_buffer);

	return_(!notFound);

}

//-------------------------------

void CSS3Protocol::s3_copy(const char *dest_server, const char *dest_bucket, const char *dest_key, const char *src_bucket, const char *src_key)

{

  	CSStringBuffer *s3_buffer;

	char date[64];

	CSString *signed_str;

	uint32_t retry_count = 0;

	S3ProtocolCon *con_data;

	enter_();

	new_(s3_buffer, CSStringBuffer());

	push_(s3_buffer);

	new_(con_data, S3ProtocolCon());

	push_(con_data);

	if (!dest_server)

		dest_server = s3_server->getCString();

retry:

	// Clear old settings. 

	con_data->ms_reset();	

	SET_DATE(date);

	// Build the URL

	s3_buffer->setLength(0);

	s3_buffer->append("http://");

	s3_buffer->append(dest_bucket);

	s3_buffer->append(".");	

	s3_buffer->append(s3_server->getCString());

	s3_buffer->append(dest_key);

	con_data->ms_setURL(s3_buffer->getCString());

	// Add the destination location

	s3_buffer->setLength(0);