1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
|
import logging
import ssl
import urllib
import hmac
import hashlib
import base64
import xmltodict
from http import client as http_client
from urllib import parse as urlparse
from time import gmtime, strftime
import boto3
from botocore.client import Config
import os
import subprocess
log = logging.getLogger('bucket_notification.tests')
NO_HTTP_BODY = ''
def put_object_tagging(conn, bucket_name, key, tags):
client = boto3.client('s3',
endpoint_url='http://'+conn.host+':'+str(conn.port),
aws_access_key_id=conn.aws_access_key_id,
aws_secret_access_key=conn.aws_secret_access_key)
return client.put_object(Body='aaaaaaaaaaa', Bucket=bucket_name, Key=key, Tagging=tags)
def make_request(conn, method, resource, parameters=None, sign_parameters=False, extra_parameters=None):
"""generic request sending to pubsub radogw
should cover: topics, notificatios and subscriptions
"""
url_params = ''
if parameters is not None:
url_params = urlparse.urlencode(parameters)
# remove 'None' from keys with no values
url_params = url_params.replace('=None', '')
url_params = '?' + url_params
if extra_parameters is not None:
url_params = url_params + '&' + extra_parameters
string_date = strftime("%a, %d %b %Y %H:%M:%S +0000", gmtime())
string_to_sign = method + '\n\n\n' + string_date + '\n' + resource
if sign_parameters:
string_to_sign += url_params
signature = base64.b64encode(hmac.new(conn.aws_secret_access_key.encode('utf-8'),
string_to_sign.encode('utf-8'),
hashlib.sha1).digest()).decode('ascii')
headers = {'Authorization': 'AWS '+conn.aws_access_key_id+':'+signature,
'Date': string_date,
'Host': conn.host+':'+str(conn.port)}
http_conn = http_client.HTTPConnection(conn.host, conn.port)
if log.getEffectiveLevel() <= 10:
http_conn.set_debuglevel(5)
http_conn.request(method, resource+url_params, NO_HTTP_BODY, headers)
response = http_conn.getresponse()
data = response.read()
status = response.status
http_conn.close()
return data.decode('utf-8'), status
def delete_all_objects(conn, bucket_name):
client = boto3.client('s3',
endpoint_url='http://'+conn.host+':'+str(conn.port),
aws_access_key_id=conn.aws_access_key_id,
aws_secret_access_key=conn.aws_secret_access_key)
objects = []
for key in client.list_objects(Bucket=bucket_name)['Contents']:
objects.append({'Key': key['Key']})
# delete objects from the bucket
response = client.delete_objects(Bucket=bucket_name,
Delete={'Objects': objects})
class PSTopicS3:
"""class to set/list/get/delete a topic
POST ?Action=CreateTopic&Name=<topic name>[&OpaqueData=<data>[&push-endpoint=<endpoint>&[<arg1>=<value1>...]]]
POST ?Action=ListTopics
POST ?Action=GetTopic&TopicArn=<topic-arn>
POST ?Action=DeleteTopic&TopicArn=<topic-arn>
"""
def __init__(self, conn, topic_name, region, endpoint_args=None, opaque_data=None):
self.conn = conn
self.topic_name = topic_name.strip()
assert self.topic_name
self.topic_arn = ''
self.attributes = {}
if endpoint_args is not None:
self.attributes = {nvp[0] : nvp[1] for nvp in urlparse.parse_qsl(endpoint_args, keep_blank_values=True)}
if opaque_data is not None:
self.attributes['OpaqueData'] = opaque_data
protocol = 'https' if conn.is_secure else 'http'
self.client = boto3.client('sns',
endpoint_url=protocol+'://'+conn.host+':'+str(conn.port),
aws_access_key_id=conn.aws_access_key_id,
aws_secret_access_key=conn.aws_secret_access_key,
region_name=region,
verify='./cert.pem')
def get_config(self):
"""get topic info"""
parameters = {'Action': 'GetTopic', 'TopicArn': self.topic_arn}
body = urlparse.urlencode(parameters)
string_date = strftime("%a, %d %b %Y %H:%M:%S +0000", gmtime())
content_type = 'application/x-www-form-urlencoded; charset=utf-8'
resource = '/'
method = 'POST'
string_to_sign = method + '\n\n' + content_type + '\n' + string_date + '\n' + resource
log.debug('StringTosign: %s', string_to_sign)
signature = base64.b64encode(hmac.new(self.conn.aws_secret_access_key.encode('utf-8'),
string_to_sign.encode('utf-8'),
hashlib.sha1).digest()).decode('ascii')
headers = {'Authorization': 'AWS '+self.conn.aws_access_key_id+':'+signature,
'Date': string_date,
'Host': self.conn.host+':'+str(self.conn.port),
'Content-Type': content_type}
if self.conn.is_secure:
http_conn = http_client.HTTPSConnection(self.conn.host, self.conn.port,
context=ssl.create_default_context(cafile='./cert.pem'))
else:
http_conn = http_client.HTTPConnection(self.conn.host, self.conn.port)
http_conn.request(method, resource, body, headers)
response = http_conn.getresponse()
data = response.read()
status = response.status
http_conn.close()
dict_response = xmltodict.parse(data)
return dict_response, status
def set_config(self):
"""set topic"""
result = self.client.create_topic(Name=self.topic_name, Attributes=self.attributes)
self.topic_arn = result['TopicArn']
return self.topic_arn
def del_config(self, topic_arn=None):
"""delete topic"""
result = self.client.delete_topic(TopicArn=(topic_arn if topic_arn is not None else self.topic_arn))
return result['ResponseMetadata']['HTTPStatusCode']
def get_list(self):
"""list all topics"""
# note that boto3 supports list_topics(), however, the result only show ARNs
parameters = {'Action': 'ListTopics'}
body = urlparse.urlencode(parameters)
string_date = strftime("%a, %d %b %Y %H:%M:%S +0000", gmtime())
content_type = 'application/x-www-form-urlencoded; charset=utf-8'
resource = '/'
method = 'POST'
string_to_sign = method + '\n\n' + content_type + '\n' + string_date + '\n' + resource
log.debug('StringTosign: %s', string_to_sign)
signature = base64.b64encode(hmac.new(self.conn.aws_secret_access_key.encode('utf-8'),
string_to_sign.encode('utf-8'),
hashlib.sha1).digest()).decode('ascii')
headers = {'Authorization': 'AWS '+self.conn.aws_access_key_id+':'+signature,
'Date': string_date,
'Host': self.conn.host+':'+str(self.conn.port),
'Content-Type': content_type}
if self.conn.is_secure:
http_conn = http_client.HTTPSConnection(self.conn.host, self.conn.port,
context=ssl.create_default_context(cafile='./cert.pem'))
else:
http_conn = http_client.HTTPConnection(self.conn.host, self.conn.port)
http_conn.request(method, resource, body, headers)
response = http_conn.getresponse()
data = response.read()
status = response.status
http_conn.close()
dict_response = xmltodict.parse(data)
return dict_response, status
class PSNotificationS3:
"""class to set/get/delete an S3 notification
PUT /<bucket>?notification
GET /<bucket>?notification[=<notification>]
DELETE /<bucket>?notification[=<notification>]
"""
def __init__(self, conn, bucket_name, topic_conf_list):
self.conn = conn
assert bucket_name.strip()
self.bucket_name = bucket_name
self.resource = '/'+bucket_name
self.topic_conf_list = topic_conf_list
self.client = boto3.client('s3',
endpoint_url='http://'+conn.host+':'+str(conn.port),
aws_access_key_id=conn.aws_access_key_id,
aws_secret_access_key=conn.aws_secret_access_key)
def send_request(self, method, parameters=None):
"""send request to radosgw"""
return make_request(self.conn, method, self.resource,
parameters=parameters, sign_parameters=True)
def get_config(self, notification=None):
"""get notification info"""
parameters = None
if notification is None:
response = self.client.get_bucket_notification_configuration(Bucket=self.bucket_name)
status = response['ResponseMetadata']['HTTPStatusCode']
return response, status
parameters = {'notification': notification}
response, status = self.send_request('GET', parameters=parameters)
dict_response = xmltodict.parse(response)
return dict_response, status
def set_config(self):
"""set notification"""
response = self.client.put_bucket_notification_configuration(Bucket=self.bucket_name,
NotificationConfiguration={
'TopicConfigurations': self.topic_conf_list
})
status = response['ResponseMetadata']['HTTPStatusCode']
return response, status
def del_config(self, notification=None):
"""delete notification"""
parameters = {'notification': notification}
return self.send_request('DELETE', parameters)
test_path = os.path.normpath(os.path.dirname(os.path.realpath(__file__))) + '/../'
def bash(cmd, **kwargs):
log.debug('running command: %s', ' '.join(cmd))
kwargs['stdout'] = subprocess.PIPE
process = subprocess.Popen(cmd, **kwargs)
s = process.communicate()[0].decode('utf-8')
return (s, process.returncode)
def admin(args, **kwargs):
""" radosgw-admin command """
cmd = [test_path + 'test-rgw-call.sh', 'call_rgw_admin', 'noname'] + args
return bash(cmd, **kwargs)
|
