cevoaustralia / aws-google-auth / 504

pending completion

Build # 504

Build Type

Pull #199

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/cevoaustralia/aws-google-auth/commit/<a class=hub.com/cevoaustralia/aws-google-auth/commit/eef0b343892e7bc8f535ad6c8f560e403d64ef35">eef0b3438<a href="https://github.com/cevoaustralia/aws-google-auth/commit/eef0b343892e7bc8f535ad6c8f560e403d64ef35">">Merge </a><a class="double-link" href="https://github.com/cevoaustralia/aws-google-auth/commit/<a class="double-link" href="https://github.com/cevoaustralia/aws-google-auth/commit/4be928655ee54b56e55b3c96878892568b3a1a60">4be928655</a>">4be928655</a><a href="https://github.com/cevoaustralia/aws-google-auth/commit/eef0b343892e7bc8f535ad6c8f560e403d64ef35"> into d473d67b0">d473d67b0</a>

Pull Request Pull Request #199: fix role/principal ARN position switch

Run Details

5 of 5 new or added lines in 1 file covered. (100.0%)

434 of 918 relevant lines covered (47.28%)

1.42 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

50.0

/aws_google_auth/amazon.py

#!/usr/bin/env python

import base64
import boto3
import os
import re

from datetime import datetime
from threading import Thread

from botocore.exceptions import ClientError, ProfileNotFound
from lxml import etree

from aws_google_auth.google import ExpectedGoogleException


class Amazon:

    def __init__(self, config, saml_xml):
        self.config = config
        self.saml_xml = saml_xml
        self.__token = None

    @property
    def sts_client(self):
        try:
            profile = os.environ.get('AWS_PROFILE')
            if profile is not None:
                del os.environ['AWS_PROFILE']
            client = boto3.client('sts', region_name=self.config.region)
            if profile is not None:
                os.environ['AWS_PROFILE'] = profile
            return client
        except ProfileNotFound as ex:
            raise ExpectedGoogleException("Error : {}.".format(ex))

    @property
    def base64_encoded_saml(self):
        return base64.b64encode(self.saml_xml).decode("utf-8")

    @property
    def token(self):
        if self.__token is None:
            self.__token = self.assume_role(self.config.role_arn,
                                            self.config.provider,
                                            self.base64_encoded_saml,
                                            self.config.duration)
        return self.__token

    @property
    def access_key_id(self):
        return self.token['Credentials']['AccessKeyId']

    @property
    def secret_access_key(self):
        return self.token['Credentials']['SecretAccessKey']

    @property
    def session_token(self):
        return self.token['Credentials']['SessionToken']

    @property
    def expiration(self):
        return self.token['Credentials']['Expiration']

    def print_export_line(self):
        export_template = "export AWS_ACCESS_KEY_ID='{}' AWS_SECRET_ACCESS_KEY='{}' AWS_SESSION_TOKEN='{}' AWS_SESSION_EXPIRATION='{}'"

        formatted = export_template.format(
            self.access_key_id,
            self.secret_access_key,
            self.session_token,
            self.expiration.strftime('%Y-%m-%dT%H:%M:%S%z'))

        print(formatted)

    @property
    def roles(self):
        doc = etree.fromstring(self.saml_xml)
        roles = {}
        for x in doc.xpath('//*[@Name = "https://aws.amazon.com/SAML/Attributes/Role"]//text()'):
            if "arn:aws:iam:" in x or "arn:aws-us-gov:iam:" in x:
                res = x.split(',')
                roles[res[0]] = res[1]
        return roles

    def assume_role(self, role, principal, saml_assertion, duration=None, auto_duration=True):

        # There is a possiblity that the role and the principal ARN might switch positions
        # as mentioned in the below issue:
        # https://github.com/cevoaustralia/aws-google-auth/issues/163#issuecomment-599113325
        #
        # To resolve this, a check is first made to see if the role contains the string "role"
        # and similarly if principal contains the string "provider"; if yes, then the ordering
        # is as expected, else switch them
        #
        # This switching logic is based on the ARN naming format as provided by AWS in their docs:
        # https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_identifiers.html#identifiers-arns
        if "role" in role and "provider" in principal:
            _role = role
            _principal = principal
        else:
            _role = principal
            _principal = role

        sts_call_vars = {
            'RoleArn': _role,
            'PrincipalArn': _principal,
            'SAMLAssertion': saml_assertion
        }

        # Try the maximum duration of 12 hours, if it fails try to use the
        # maximum duration indicated by the error
        if self.config.auto_duration and auto_duration:
            sts_call_vars['DurationSeconds'] = self.config.max_duration
            try:
                res = self.sts_client.assume_role_with_saml(**sts_call_vars)
            except ClientError as err:
                if (err.response.get('Error', []).get('Code') == 'ValidationError' and err.response.get('Error', []).get('Message')):
                    m = re.search(
                        'Member must have value less than or equal to ([0-9]{3,5})',
                        err.response['Error']['Message']
                    )
                    if m is not None and m.group(1):
                        new_duration = int(m.group(1))
                        return self.assume_role(role, principal,
                                                saml_assertion,
                                                duration=new_duration,
                                                auto_duration=False)
                # Unknown error or no max time returned in error message
                raise
        elif duration:
            sts_call_vars['DurationSeconds'] = duration

        res = self.sts_client.assume_role_with_saml(**sts_call_vars)

        return res

    def resolve_aws_aliases(self, roles):
        def resolve_aws_alias(role, principal, aws_dict):
            session = boto3.session.Session(region_name=self.config.region)

            sts = session.client('sts')
            saml = sts.assume_role_with_saml(RoleArn=role,
                                             PrincipalArn=principal,
                                             SAMLAssertion=self.base64_encoded_saml)

            iam = session.client('iam',
                                 aws_access_key_id=saml['Credentials']['AccessKeyId'],
                                 aws_secret_access_key=saml['Credentials']['SecretAccessKey'],
                                 aws_session_token=saml['Credentials']['SessionToken'])
            try:
                response = iam.list_account_aliases()
                account_alias = response['AccountAliases'][0]
                aws_dict[role.split(':')[4]] = account_alias
            except:
                sts = session.client('sts',
                                     aws_access_key_id=saml['Credentials']['AccessKeyId'],
                                     aws_secret_access_key=saml['Credentials']['SecretAccessKey'],
                                     aws_session_token=saml['Credentials']['SessionToken'])

                account_id = sts.get_caller_identity().get('Account')
                aws_dict[role.split(':')[4]] = '{}'.format(account_id)

        threads = []
        aws_id_alias = {}
        for number, (role, principal) in enumerate(roles.items()):
            t = Thread(target=resolve_aws_alias, args=(role, principal, aws_id_alias))
            t.start()
            threads.append(t)

        for t in threads:
            t.join()

        return aws_id_alias

    @staticmethod
    def is_valid_saml_assertion(saml_xml):
        if saml_xml is None:
            return False

        try:
            doc = etree.fromstring(saml_xml)
            conditions = list(doc.iter(tag='{urn:oasis:names:tc:SAML:2.0:assertion}Conditions'))
            not_before_str = conditions[0].get('NotBefore')
            not_on_or_after_str = conditions[0].get('NotOnOrAfter')

            now = datetime.utcnow()
            not_before = datetime.strptime(not_before_str, "%Y-%m-%dT%H:%M:%S.%fZ")
            not_on_or_after = datetime.strptime(not_on_or_after_str, "%Y-%m-%dT%H:%M:%S.%fZ")

            if not_before <= now < not_on_or_after:
                return True
            else:
                return False
        except Exception:
            return False

1	#!/usr/bin/env python
2
3	import base64	3✔
4	import boto3	3✔
5	import os	3✔
6	import re	3✔
7
8	from datetime import datetime	3✔
9	from threading import Thread	3✔
10
11	from botocore.exceptions import ClientError, ProfileNotFound	3✔
12	from lxml import etree	3✔
13
14	from aws_google_auth.google import ExpectedGoogleException	3✔
15
16
17	class Amazon:	3✔
18
19	def __init__(self, config, saml_xml):	3✔
20	self.config = config	3✔
21	self.saml_xml = saml_xml	3✔
22	self.__token = None	3✔
23
24	@property	3✔
25	def sts_client(self):
26	try:	3✔
27	profile = os.environ.get('AWS_PROFILE')	3✔
28	if profile is not None:	3✔
29	del os.environ['AWS_PROFILE']	3✔
30	client = boto3.client('sts', region_name=self.config.region)	3✔
31	if profile is not None:	3✔
32	os.environ['AWS_PROFILE'] = profile	3✔
33	return client	3✔
34	except ProfileNotFound as ex:	×
35	raise ExpectedGoogleException("Error : {}.".format(ex))	×
36
37	@property	3✔
38	def base64_encoded_saml(self):
39	return base64.b64encode(self.saml_xml).decode("utf-8")	×
40
41	@property	3✔
42	def token(self):
43	if self.__token is None:	×
44	self.__token = self.assume_role(self.config.role_arn,	×
45	self.config.provider,
46	self.base64_encoded_saml,
47	self.config.duration)
48	return self.__token	×
49
50	@property	3✔
51	def access_key_id(self):
52	return self.token['Credentials']['AccessKeyId']	×
53
54	@property	3✔
55	def secret_access_key(self):
56	return self.token['Credentials']['SecretAccessKey']	×
57
58	@property	3✔
59	def session_token(self):
60	return self.token['Credentials']['SessionToken']	×
61
62	@property	3✔
63	def expiration(self):
64	return self.token['Credentials']['Expiration']	×
65
66	def print_export_line(self):	3✔
67	export_template = "export AWS_ACCESS_KEY_ID='{}' AWS_SECRET_ACCESS_KEY='{}' AWS_SESSION_TOKEN='{}' AWS_SESSION_EXPIRATION='{}'"	×
68
69	formatted = export_template.format(	×
70	self.access_key_id,
71	self.secret_access_key,
72	self.session_token,
73	self.expiration.strftime('%Y-%m-%dT%H:%M:%S%z'))
74
75	print(formatted)	×
76
77	@property	3✔
78	def roles(self):
79	doc = etree.fromstring(self.saml_xml)	3✔
80	roles = {}	3✔
81	for x in doc.xpath('//*[@Name = "https://aws.amazon.com/SAML/Attributes/Role"]//text()'):	3✔
82	if "arn:aws:iam:" in x or "arn:aws-us-gov:iam:" in x:	3✔
83	res = x.split(',')	3✔
84	roles[res[0]] = res[1]	3✔
85	return roles	3✔
86
87	def assume_role(self, role, principal, saml_assertion, duration=None, auto_duration=True):	3✔
88
89	# There is a possiblity that the role and the principal ARN might switch positions
90	# as mentioned in the below issue:
91	# https://github.com/cevoaustralia/aws-google-auth/issues/163#issuecomment-599113325
92	#
93	# To resolve this, a check is first made to see if the role contains the string "role"
94	# and similarly if principal contains the string "provider"; if yes, then the ordering
95	# is as expected, else switch them
96	#
97	# This switching logic is based on the ARN naming format as provided by AWS in their docs:
98	# https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_identifiers.html#identifiers-arns
99	if "role" in role and "provider" in principal:	×
100	_role = role	×
101	_principal = principal	×
102	else:
103	_role = principal	×
104	_principal = role	×
105
106	sts_call_vars = {	×
107	'RoleArn': _role,
108	'PrincipalArn': _principal,
109	'SAMLAssertion': saml_assertion
110	}
111
112	# Try the maximum duration of 12 hours, if it fails try to use the
113	# maximum duration indicated by the error
114	if self.config.auto_duration and auto_duration:	×
115	sts_call_vars['DurationSeconds'] = self.config.max_duration	×
116	try:	×
117	res = self.sts_client.assume_role_with_saml(**sts_call_vars)	×
118	except ClientError as err:	×
119	if (err.response.get('Error', []).get('Code') == 'ValidationError' and err.response.get('Error', []).get('Message')):	×
120	m = re.search(	×
121	'Member must have value less than or equal to ([0-9]{3,5})',
122	err.response['Error']['Message']
123	)
124	if m is not None and m.group(1):	×
125	new_duration = int(m.group(1))	×
126	return self.assume_role(role, principal,	×
127	saml_assertion,
128	duration=new_duration,
129	auto_duration=False)
130	# Unknown error or no max time returned in error message
131	raise	×
132	elif duration:	×
133	sts_call_vars['DurationSeconds'] = duration	×
134
135	res = self.sts_client.assume_role_with_saml(**sts_call_vars)	×
136
137	return res	×
138
139	def resolve_aws_aliases(self, roles):	3✔
140	def resolve_aws_alias(role, principal, aws_dict):	×
141	session = boto3.session.Session(region_name=self.config.region)	×
142
143	sts = session.client('sts')	×
144	saml = sts.assume_role_with_saml(RoleArn=role,	×
145	PrincipalArn=principal,
146	SAMLAssertion=self.base64_encoded_saml)
147
148	iam = session.client('iam',	×
149	aws_access_key_id=saml['Credentials']['AccessKeyId'],
150	aws_secret_access_key=saml['Credentials']['SecretAccessKey'],
151	aws_session_token=saml['Credentials']['SessionToken'])
152	try:	×
153	response = iam.list_account_aliases()	×
154	account_alias = response['AccountAliases'][0]	×
155	aws_dict[role.split(':')[4]] = account_alias	×
156	except:	×
157	sts = session.client('sts',	×
158	aws_access_key_id=saml['Credentials']['AccessKeyId'],
159	aws_secret_access_key=saml['Credentials']['SecretAccessKey'],
160	aws_session_token=saml['Credentials']['SessionToken'])
161
162	account_id = sts.get_caller_identity().get('Account')	×
163	aws_dict[role.split(':')[4]] = '{}'.format(account_id)	×
164
165	threads = []	×
166	aws_id_alias = {}	×
167	for number, (role, principal) in enumerate(roles.items()):	×
168	t = Thread(target=resolve_aws_alias, args=(role, principal, aws_id_alias))	×
169	t.start()	×
170	threads.append(t)	×
171
172	for t in threads:	×
173	t.join()	×
174
175	return aws_id_alias	×
176
177	@staticmethod	3✔
178	def is_valid_saml_assertion(saml_xml):
179	if saml_xml is None:	3✔
180	return False	3✔
181
182	try:	3✔
183	doc = etree.fromstring(saml_xml)	3✔
184	conditions = list(doc.iter(tag='{urn:oasis:names:tc:SAML:2.0:assertion}Conditions'))	3✔
185	not_before_str = conditions[0].get('NotBefore')	3✔
186	not_on_or_after_str = conditions[0].get('NotOnOrAfter')	3✔
187
188	now = datetime.utcnow()	3✔
189	not_before = datetime.strptime(not_before_str, "%Y-%m-%dT%H:%M:%S.%fZ")	3✔
190	not_on_or_after = datetime.strptime(not_on_or_after_str, "%Y-%m-%dT%H:%M:%S.%fZ")	3✔
191
192	if not_before <= now < not_on_or_after:	3✔
193	return True	3✔
194	else:
195	return False	3✔
196	except Exception:	3✔
197	return False	3✔

cevoaustralia / aws-google-auth / 504

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous