-
Notifications
You must be signed in to change notification settings - Fork 0
/
controltower.py
61 lines (52 loc) · 1.97 KB
/
controltower.py
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
import boto3
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
from botocore.exceptions import ClientError
from urllib import request, error
import os
import json
from functools import partial
REGION = os.environ["AWS_REGION"]
SERVICE = "controltower"
SERVICE_URL = f'https://prod.{REGION}.blackbeard.aws.a2z.com/'
def make_signed_headers(method, url, data=None, params=None, headers=None):
session = boto3.Session()
credentials = session.get_credentials()
creds = credentials.get_frozen_credentials()
request = AWSRequest(method=method, url=url, data=data, params=params, headers=headers)
SigV4Auth(creds, SERVICE, REGION).add_auth(request)
return dict(request.headers)
def signed_request(method, url, data=None, headers=None):
signed_headers = make_signed_headers(method=method, url=url, data=data, headers=headers)
return request.Request(url, method=method, headers=signed_headers, data=data)
def execute_control_tower_operation(operation, data):
headers = {
'Content-Type': 'application/x-amz-json-1.0',
'X-Amz-Target': f'AWSBlackbeardService.{operation}',
}
data = json.dumps(data).encode()
try:
_request = signed_request(
method='POST',
url=SERVICE_URL,
headers=headers,
data=data
)
response = request.urlopen(_request)
except error.HTTPError as e:
err = json.load(e.fp)
raise ClientError(
operation_name=operation,
error_response={
'Error': {
'Code': err['__type'].split('#')[-1],
'Message': err.get('Message', "")
}
}
) from e
return json.loads(response.read())
def exec_wrapper(operation, **kwargs):
return execute_control_tower_operation(operation, kwargs)
def __getattr__(param):
operation = ''.join(map(lambda x: x.title(), param.split('_')))
return partial(exec_wrapper, operation)