#!/usr/bin/python

# Licensed to the Apache Software Foundation (ASF) under one

# or more contributor license agreements. See the NOTICE file

# distributed with this work for additional information

# regarding copyright ownership. The ASF licenses this file

# to you under the Apache License, Version 2.0 (the

# "License"); you may not use this file except in compliance

# with the License. You may obtain a copy of the License at

#

# http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing,

# software distributed under the License is distributed on an

# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

# KIND, either express or implied. See the License for the

# specific language governing permissions and limitations

# under the License.

# Version @VERSION@

#

# A plugin for executing script needed by CloudStack

from copy import copy

from datetime import datetime

from httplib import *

from string import join

from string import split

import os

import sys

import time

import md5 as md5mod

import sha

import base64

import hmac

import traceback

import urllib2

from xml.dom.minidom import parseString

import XenAPIPlugin

sys.path.extend(["/opt/xensource/sm/"])

import util

import cloudstack_pluginlib as lib

import logging

lib.setup_logging("/var/log/cloud/s3xen.log")

NULL = 'null'

# Value conversion utility functions ...

def to_none(value):

if value is None:

return None

if isinstance(value, basestring) and value.strip().lower() == NULL:

return None

return value

def to_bool(value):

if to_none(value) is None:

return False

if isinstance(value, basestring) and value.strip().lower() == 'true':

return True

if isinstance(value, int) and value:

return True

return False

def to_integer(value, default):

if to_none(value) is None or not isinstance(value, int):

return default

return int(value)

def optional_str_value(value, default):

if is_not_blank(value):

return value

return default

def is_blank(value):

return not is_not_blank(value)

def is_not_blank(value):

if to_none(value) is None or not isinstance(value, basestring):

return True

if value.strip == '':

return False

return True

def get_optional_key(map, key, default=''):

if key in map:

return map[key]

return default

def log(message):

logging.debug('#### VMOPS %s ####' % message)

def echo(fn):

def wrapped(*v, **k):

name = fn.__name__

log("enter %s ####" % name)

res = fn(*v, **k)

log("exit %s with result %s" % (name, res))

return res

return wrapped

def require_str_value(value, error_message):

if is_not_blank(value):

return value

raise ValueError(error_message)

def retry(max_attempts, fn):

attempts = 1

while attempts <= max_attempts:

log("Attempting execution " + str(attempts) + "/" + str(

max_attempts) + " of " + fn.__name__)

try:

return fn()

except:

if (attempts >= max_attempts):

raise

attempts = attempts + 1

def compute_md5(filename, buffer_size=8192):

hasher = md5mod.md5()

file = open(filename, 'rb')

try:

data = file.read(buffer_size)

while data != "":

hasher.update(data)

data = file.read(buffer_size)

return base64.encodestring(hasher.digest())[:-1]

finally:

file.close()

class S3Client(object):

DEFAULT_END_POINT = 's3.amazonaws.com'

DEFAULT_CONNECTION_TIMEOUT = 50000

DEFAULT_SOCKET_TIMEOUT = 50000

DEFAULT_MAX_ERROR_RETRY = 3

HEADER_CONTENT_MD5 = 'Content-MD5'

HEADER_CONTENT_TYPE = 'Content-Type'

HEADER_CONTENT_LENGTH = 'Content-Length'

def __init__(self, access_key, secret_key, end_point=None,

https_flag=None, connection_timeout=None, socket_timeout=None,

max_error_retry=None):

self.access_key = require_str_value(

access_key, 'An access key must be specified.')

self.secret_key = require_str_value(

secret_key, 'A secret key must be specified.')

self.end_point = optional_str_value(end_point, self.DEFAULT_END_POINT)

self.https_flag = to_bool(https_flag)

self.connection_timeout = to_integer(

connection_timeout, self.DEFAULT_CONNECTION_TIMEOUT)

self.socket_timeout = to_integer(

socket_timeout, self.DEFAULT_SOCKET_TIMEOUT)

self.max_error_retry = to_integer(

max_error_retry, self.DEFAULT_MAX_ERROR_RETRY)

def build_canocialized_resource(self, bucket, key):

if not key.startswith("/"):

uri = bucket + "/" + key

else:

uri = bucket + key

return "/" + uri

def noop_send_body(connection):

pass

def noop_read(response):

return response.read()

def do_operation(

self, method, bucket, key, input_headers={},

fn_send_body=noop_send_body, fn_read=noop_read):

headers = copy(input_headers)

headers['Expect'] = '100-continue'

uri = self.build_canocialized_resource(bucket, key)

signature, request_date = self.sign_request(method, uri, headers)

headers['Authorization'] = "AWS " + self.access_key + ":" + signature

headers['Date'] = request_date

def perform_request():

connection = None

if self.https_flag:

connection = HTTPSConnection(self.end_point)

else:

connection = HTTPConnection(self.end_point)

try:

connection.timeout = self.socket_timeout

connection.putrequest(method, uri)

for k, v in headers.items():

connection.putheader(k, v)

connection.endheaders()

fn_send_body(connection)

response = connection.getresponse()

log("Sent " + method + " request to " + self.end_point +

uri + " with headers " + str(headers) +

". Received response status " + str(response.status) +

": " + response.reason)

return fn_read(response)

finally:

connection.close()

return retry(self.max_error_retry, perform_request)

'''

See http://bit.ly/MMC5de for more information regarding the creation of

AWS authorization tokens and header signing

'''

def sign_request(self, operation, canocialized_resource, headers):

request_date = datetime.utcnow(

).strftime('%a, %d %b %Y %H:%M:%S +0000')

content_hash = get_optional_key(headers, self.HEADER_CONTENT_MD5)

content_type = get_optional_key(headers, self.HEADER_CONTENT_TYPE)

string_to_sign = join(

[operation, content_hash, content_type, request_date,

canocialized_resource], '\n')

signature = base64.encodestring(

hmac.new(self.secret_key, string_to_sign.encode('utf8'),

sha).digest())[:-1]

return signature, request_date

def getText(self, nodelist):

rc = []

for node in nodelist:

if node.nodeType == node.TEXT_NODE:

rc.append(node.data)

return ''.join(rc)

def multiUpload(self, bucket, key, src_fileName, chunkSize=5 * 1024 * 1024):

uploadId={}

def readInitalMultipart(response):

data = response.read()

xmlResult = parseString(data)

result = xmlResult.getElementsByTagName("InitiateMultipartUploadResult")[0]

upload = result.getElementsByTagName("UploadId")[0]

uploadId["0"] = upload.childNodes[0].data

self.do_operation('POST', bucket, key + "?uploads", fn_read=readInitalMultipart)

fileSize = os.path.getsize(src_fileName)

parts = fileSize / chunkSize + ((fileSize % chunkSize) and 1)

part = 1

srcFile = open(src_fileName, 'rb')

etags = []

while part <= parts:

offset = part - 1

size = min(fileSize - offset * chunkSize, chunkSize)

headers = {

self.HEADER_CONTENT_LENGTH: size

}

def send_body(connection):

srcFile.seek(offset * chunkSize)

block = srcFile.read(size)

connection.send(block)

def read_multiPart(response):

etag = response.getheader('ETag')

etags.append((part, etag))

self.do_operation("PUT", bucket, "%s?partNumber=%s&uploadId=%s"%(key, part, uploadId["0"]), headers, send_body, read_multiPart)

part = part + 1

srcFile.close()

data = []

partXml = "<Part><PartNumber>%i</PartNumber><ETag>%s</ETag></Part>"

for etag in etags:

data.append(partXml%etag)

msg = "<CompleteMultipartUpload>%s</CompleteMultipartUpload>"%("".join(data))

size = len(msg)

headers = {

self.HEADER_CONTENT_LENGTH: size

}

def send_complete_multipart(connection):

connection.send(msg)

self.do_operation("POST", bucket, "%s?uploadId=%s"%(key, uploadId["0"]), headers, send_complete_multipart)

def put(self, bucket, key, src_filename, maxSingleUpload):

if not os.path.isfile(src_filename):

raise Exception(

"Attempt to put " + src_filename + " that does not exist.")

size = os.path.getsize(src_filename)

if size > maxSingleUpload or maxSingleUpload == 0:

return self.multiUpload(bucket, key, src_filename)

headers = {

self.HEADER_CONTENT_MD5: compute_md5(src_filename),

self.HEADER_CONTENT_TYPE: 'application/octet-stream',

self.HEADER_CONTENT_LENGTH: str(os.stat(src_filename).st_size),

}

def send_body(connection):

src_file = open(src_filename, 'rb')

try:

while True:

block = src_file.read(8192)

if not block:

break

connection.send(block)

except:

src_file.close()

self.do_operation('PUT', bucket, key, headers, send_body)

def get(self, bucket, key, target_filename):

def read(response):

file = open(target_filename, 'wb')

try:

while True:

block = response.read(8192)

if not block:

break

file.write(block)

except:

file.close()

return self.do_operation('GET', bucket, key, fn_read=read)

def delete(self, bucket, key):

return self.do_operation('DELETE', bucket, key)

def parseArguments(args):

# The keys in the args map will correspond to the properties defined on

# the com.cloud.utils.S3Utils#ClientOptions interface

client = S3Client(

args['accessKey'], args['secretKey'], args['endPoint'],

args['https'], args['connectionTimeout'], args['socketTimeout'])

operation = args['operation']

bucket = args['bucket']

key = args['key']

filename = args['filename']

maxSingleUploadBytes = int(args["maxSingleUploadSizeInBytes"])

if is_blank(operation):

raise ValueError('An operation must be specified.')

if is_blank(bucket):

raise ValueError('A bucket must be specified.')

if is_blank(key):

raise ValueError('A value must be specified.')

if is_blank(filename):

raise ValueError('A filename must be specified.')

return client, operation, bucket, key, filename, maxSingleUploadBytes

@echo

def s3(session, args):

client, operation, bucket, key, filename, maxSingleUploadBytes = parseArguments(args)

try:

if operation == 'put':

client.put(bucket, key, filename, maxSingleUploadBytes)

elif operation == 'get':

client.get(bucket, key, filename)

elif operation == 'delete':

client.delete(bucket, key, filename)

else:

raise RuntimeError(

"S3 plugin does not support operation " + operation)

return 'true'

except:

log("Operation " + operation + " on file " + filename +

" from/in bucket " + bucket + " key " + key)

log(traceback.format_exc())

return 'false'

if __name__ == "__main__":

XenAPIPlugin.dispatch({"s3": s3})