| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133 |
- # Copyright 2012-2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
- #
- # Modifications made by Cloudera are:
- # Copyright (c) 2016 Cloudera, Inc. All rights reserved.
- #
- # Licensed under the Apache License, Version 2.0 (the "License"). You
- # may not use this file except in compliance with the License. A copy of
- # the License is located at
- #
- # http://aws.amazon.com/apache2.0/
- #
- # or in the "license" file accompanying this file. This file 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.
- from base64 import urlsafe_b64encode
- from email.utils import formatdate
- import logging
- from altuscli.compat import json
- from altuscli.compat import OrderedDict
- from altuscli.compat import urlsplit
- from altuscli.exceptions import NoCredentialsError
- from asn1crypto import keys, pem
- import rsa
- LOG = logging.getLogger('altuscli.auth')
- class BaseSigner(object):
- def add_auth(self, request):
- raise NotImplementedError("add_auth")
- class RSAv1Auth(BaseSigner):
- """
- RSA signing with a SHA-256 hash returning a base64 encoded signature.
- """
- AUTH_METHOD_NAME = 'rsav1'
- def __init__(self, credentials):
- self.credentials = credentials
- def _sign_string(self, string_to_sign):
- try:
- # We expect the private key to be the an PKCS8 pem formatted string.
- pem_bytes = self.credentials.private_key.encode('utf-8')
- if pem.detect(pem_bytes):
- _, _, der_bytes = pem.unarmor(pem_bytes)
- # In PKCS8 the key is wrapped in a container that describes it
- info = keys.PrivateKeyInfo.load(der_bytes, strict=True)
- # The unwrapped key is equivalent to pkcs1 contents
- key = rsa.PrivateKey.load_pkcs1(info.unwrap().dump(), 'DER')
- else:
- raise Exception('Not a PEM file')
- except:
- message = \
- "Failed to import private key from: '%s'. The private key is " \
- "corrupted or it is not in PKCS8 PEM format. The private key " \
- "was extracted either from 'env' (environment variables), " \
- "'shared-credentials-file' (a profile in the shared " \
- "credential file, by default under ~/.altus/credentials), or " \
- "'auth-config-file' (a file containing the credentials whose " \
- "location was supplied on the command line.)" % \
- self.credentials.method
- LOG.debug(message, exc_info=True)
- raise Exception(message)
- # We sign the hash.
- signature = rsa.sign(string_to_sign.encode('utf-8'), key, 'SHA-256')
- return urlsafe_b64encode(signature).strip().decode('utf-8')
- def _canonical_standard_headers(self, headers):
- interesting_headers = ['content-type', 'x-altus-date']
- hoi = []
- if 'x-altus-date' in headers:
- raise Exception("x-altus-date found in headers!")
- headers['x-altus-date'] = self._get_date()
- for ih in interesting_headers:
- found = False
- for key in headers:
- lk = key.lower()
- if headers[key] is not None and lk == ih:
- hoi.append(headers[key].strip())
- found = True
- if not found:
- hoi.append('')
- return '\n'.join(hoi)
- def _canonical_string(self, method, split, headers):
- cs = method.upper() + '\n'
- cs += self._canonical_standard_headers(headers) + '\n'
- cs += split.path + '\n'
- cs += RSAv1Auth.AUTH_METHOD_NAME
- return cs
- def _get_signature(self, method, split, headers):
- string_to_sign = self._canonical_string(method, split, headers)
- LOG.debug('StringToSign:\n%s', string_to_sign)
- return self._sign_string(string_to_sign)
- def add_auth(self, request):
- if self.credentials is None:
- raise NoCredentialsError
- LOG.debug("Calculating signature using RSAv1Auth.")
- LOG.debug('HTTP request method: %s', request.method)
- split = urlsplit(request.url)
- signature = self._get_signature(request.method,
- split,
- request.headers)
- self._inject_signature(request, signature)
- def _get_date(self):
- return formatdate(usegmt=True)
- def _inject_signature(self, request, signature):
- if 'x-altus-auth' in request.headers:
- raise Exception("x-altus-auth found in headers!")
- request.headers['x-altus-auth'] = self._get_signature_header(signature)
- def _get_signature_header(self, signature):
- auth_params = OrderedDict()
- auth_params['access_key_id'] = self.credentials.access_key_id
- auth_params['auth_method'] = RSAv1Auth.AUTH_METHOD_NAME
- encoded_auth_params = json.dumps(auth_params).encode('utf-8')
- return "%s.%s" % (
- urlsafe_b64encode(encoded_auth_params).strip().decode('utf-8'),
- signature)
- AUTH_TYPE_MAPS = {
- RSAv1Auth.AUTH_METHOD_NAME: RSAv1Auth,
- }
|