2692903695

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/dowjones/tokendito/commit/<a class=hub.com/dowjones/tokendito/commit/1934788631ab41e9f06815c4ca4485eaf0359849">193478863<a href="https://github.com/dowjones/tokendito/commit/1934788631ab41e9f06815c4ca4485eaf0359849">">Merge </a><a class="double-link" href="https://github.com/dowjones/tokendito/commit/<a class="double-link" href="https://github.com/dowjones/tokendito/commit/770a670dac2d88d38da8023331c385a208053ec0">770a670da</a>">770a670da</a><a href="https://github.com/dowjones/tokendito/commit/1934788631ab41e9f06815c4ca4485eaf0359849"> into fa346f889">fa346f889</a>

Pull Request Pull Request #74: Cse-3121 discover aws url

Run Details

87 of 167 new or added lines in 4 files covered. (52.1%)

670 of 1005 relevant lines covered (66.67%)

1.97 hits per line

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

73.2

/tokendito/aws.py

# vim: set filetype=python ts=4 sw=4
# -*- coding: utf-8 -*-
"""This module handles the all aws workflow operations.

Tasks include:
1. Aws Authentication with SAML
2. Updating the AWS Credentials
3. Updating the AWS Config

"""
import codecs
import logging
import sys

from botocore import UNSIGNED
from botocore.client import Config
from botocore.exceptions import ClientError
import botocore.session
import requests
from tokendito import user


logger = logging.getLogger(__name__)


def authenticate_to_roles(secret_session_token, urls, cookies=None):
    """Authenticate AWS user with saml.

    :param secret_session_token: secret session token
    :param urls: list of tuples or tuple, with apps info
    :param cookies: html cookies
    :return: response text

    """
    payload = {"onetimetoken": secret_session_token}
    url_list = [urls] if isinstance(urls, tuple) else urls
    responses = []

    for url, label in url_list:
        try:
            logger.debug(f"Authenticate AWS user with SAML URL [{url}]")

            response = requests.get(url, params=payload, cookies=cookies)
            saml_response_string = response.text
            if response.status_code == 400 or response.status_code == 401:
                errmsg = "Invalid Credentials."
                logger.error(f"{errmsg}\nExiting with code:{response.status_code}")
                sys.exit(2)
            elif response.status_code == 404:
                errmsg = "Invalid Okta application URL. Please verify your configuration."
                logger.error(f"{errmsg}")
                sys.exit(2)
            elif response.status_code >= 500 and response.status_code < 504:
                errmsg = (
                    "Unable to establish connection with Okta. Verify Okta Org URL and try again."
                )
                logger.error(f"{errmsg}\nExiting with code:{response.status_code}")
                sys.exit(2)
            elif response.status_code != 200:
                logger.error(f"Exiting with code:{response.status_code}")
                logger.debug(saml_response_string)
                sys.exit(2)

        except Exception as error:
            errmsg = f"Okta auth failed:\n{error}"
            logger.error(errmsg)
            sys.exit(1)

        saml_xml = user.validate_saml_response(saml_response_string)
        responses.append((url, saml_response_string, saml_xml, label))

    return responses


def assume_role(role_arn, provider_arn, saml):
    """Return AssumeRoleWithSaml API response.

    :param role_arn: IAM role arn to assume
    :param provider_arn: ARN of saml-provider resource
    :param saml: decoded saml response from okta
    :return: AssumeRoleWithSaml API response

    """
    default_error = (
        "\nUnable to assume role with the following details:\n- Role ARN: {}\n- Error: {}\n"
    )

    encoded_xml = codecs.encode(saml.encode("utf-8"), "base64")
    assume_role_response = None
    # Attempt to assume a role with the following durations:
    # 12h, 8h, 6h, 4h, 2h, 1h, 30m, 15m
    session_times = [43200, 28800, 21600, 14400, 7200, 3600, 1800, 900, "exit"]
    for duration in session_times:
        if duration == "exit":
            logger.error(
                default_error.format(
                    role_arn,
                    f"IAM role session time is not within set: {session_times[:-1]}",
                )
            )
            sys.exit(2)

        assume_role_response = handle_assume_role(
            role_arn, provider_arn, encoded_xml, duration, default_error
        )
        if "Credentials" in assume_role_response:
            break

    return assume_role_response


def handle_assume_role(role_arn, provider_arn, encoded_xml, duration, default_error):
    """Handle assume role with saml.

    :param role_arn: IAM role arn to assume
    :param provider_arn: ARN of saml-provider resource
    :param saml: decoded saml response from okta
    :return: AssumeRoleWithSaml API responses
    """
    logger.debug(f"Attempting session time [{duration}]")
    try:
        session = botocore.session.get_session()
        client = session.create_client("sts", config=Config(signature_version=UNSIGNED))
        assume_role_response = client.assume_role_with_saml(
            RoleArn=role_arn,
            PrincipalArn=provider_arn,
            SAMLAssertion=encoded_xml.decode(),
            DurationSeconds=duration,
        )
        # Client Exceptions
    except ClientError as error:
        if error.response["Error"]["Code"] == "ValidationError":
            logger.info(
                f"AssumeRoleWithSaml failed with {error.response['Error']['Code']} "
                f"for duration {duration}"
            )
            assume_role_response = "continue"
        elif error.response["Error"]["Code"] == "AccessDenied":
            errmsg = f"Error assuming intermediate {provider_arn} SAML role"
            logger.error(errmsg)
            sys.exit(2)
        else:
            logger.error(default_error.format(role_arn, str(error)))
            sys.exit(1)
        # Service Exceptions
    except Exception as error:
        logger.error(default_error.format(role_arn, str(error)))
        sys.exit(1)

    return assume_role_response


def ensure_keys_work(assume_role_response):
    """Validate the temporary AWS credentials.

    :param aws_access_key: AWS access key
    :param aws_secret_key: AWS secret key
    :param aws_session_token: AWS session token
    :return:

    """
    logger.debug("Validate the temporary AWS credentials")

    aws_access_key = assume_role_response["Credentials"]["AccessKeyId"]
    aws_secret_key = assume_role_response["Credentials"]["SecretAccessKey"]
    aws_session_token = assume_role_response["Credentials"]["SessionToken"]

    try:
        session = botocore.session.get_session()

        client = session.create_client(
            "sts",
            aws_access_key_id=aws_access_key,
            aws_secret_access_key=aws_secret_key,
            aws_session_token=aws_session_token,
        )
        client.get_caller_identity()
    except Exception as auth_error:
        logger.error(f"There was an error authenticating your keys with AWS: {auth_error}")
        sys.exit(1)


def select_assumeable_role(apps):
    """Select the role to perform the AssumeRoleWithSaml.

    # :param apps: apps metadata, list of tuples
    # :return: AWS AssumeRoleWithSaml response, role name, tuple
    """
    authenticated_aps = {}
    for url, saml_response, saml, label in apps:
        roles_and_providers = user.extract_arns(saml)
        authenticated_aps[url] = {
            "roles": list(roles_and_providers.keys()),
            "saml": saml,
            "saml_response_string": saml_response,
            "roles_and_providers": roles_and_providers,
            "label": label,
        }

    role_arn, _id = user.select_role_arn(authenticated_aps)
    role_name = role_arn.split("/")[-1]

    assume_role_response = assume_role(
        role_arn,
        authenticated_aps[_id]["roles_and_providers"][role_arn],
        authenticated_aps[_id]["saml"],
    )

    return assume_role_response, role_name

1	# vim: set filetype=python ts=4 sw=4
2	# -- coding: utf-8 --
3	"""This module handles the all aws workflow operations.	4✔
4
5	Tasks include:
6	1. Aws Authentication with SAML
7	2. Updating the AWS Credentials
8	3. Updating the AWS Config
9
10	"""
11	import codecs	4✔
12	import logging	4✔
13	import sys	4✔
14
15	from botocore import UNSIGNED	4✔
16	from botocore.client import Config	4✔
17	from botocore.exceptions import ClientError	4✔
18	import botocore.session	4✔
19	import requests	4✔
20	from tokendito import user	4✔
21
22
23	logger = logging.getLogger(__name__)	4✔
24
25
26	def authenticate_to_roles(secret_session_token, urls, cookies=None):	4✔
27	"""Authenticate AWS user with saml.
28
29	:param secret_session_token: secret session token
30	:param urls: list of tuples or tuple, with apps info
31	:param cookies: html cookies
32	:return: response text
33
34	"""
35	payload = {"onetimetoken": secret_session_token}	4✔
36	url_list = [urls] if isinstance(urls, tuple) else urls	4✔
37	responses = []	4✔
38
39	for url, label in url_list:	4✔
40	try:	4✔
41	logger.debug(f"Authenticate AWS user with SAML URL [{url}]")	4✔
42
43	response = requests.get(url, params=payload, cookies=cookies)	4✔
44	saml_response_string = response.text	1✔
45	if response.status_code == 400 or response.status_code == 401:	1✔
NEW 46	errmsg = "Invalid Credentials."	×
NEW 47	logger.error(f"{errmsg}\nExiting with code:{response.status_code}")	×
NEW 48	sys.exit(2)	×
49	elif response.status_code == 404:	1✔
NEW 50	errmsg = "Invalid Okta application URL. Please verify your configuration."	×
NEW 51	logger.error(f"{errmsg}")	×
NEW 52	sys.exit(2)	×
53	elif response.status_code >= 500 and response.status_code < 504:	1✔
NEW 54	errmsg = (	×
55	"Unable to establish connection with Okta. Verify Okta Org URL and try again."
56	)
NEW 57	logger.error(f"{errmsg}\nExiting with code:{response.status_code}")	×
NEW 58	sys.exit(2)	×
59	elif response.status_code != 200:	1✔
NEW 60	logger.error(f"Exiting with code:{response.status_code}")	×
NEW 61	logger.debug(saml_response_string)	×
NEW 62	sys.exit(2)	×
63
64	except Exception as error:	3✔
65	errmsg = f"Okta auth failed:\n{error}"	3✔
66	logger.error(errmsg)	3✔
67	sys.exit(1)	3✔
68
69	saml_xml = user.validate_saml_response(saml_response_string)	1✔
70	responses.append((url, saml_response_string, saml_xml, label))	1✔
71
72	return responses	1✔
73
74
75	def assume_role(role_arn, provider_arn, saml):	4✔
76	"""Return AssumeRoleWithSaml API response.
77
78	:param role_arn: IAM role arn to assume
79	:param provider_arn: ARN of saml-provider resource
80	:param saml: decoded saml response from okta
81	:return: AssumeRoleWithSaml API response
82
83	"""
84	default_error = (	1✔
85	"\nUnable to assume role with the following details:\n- Role ARN: {}\n- Error: {}\n"
86	)
87
88	encoded_xml = codecs.encode(saml.encode("utf-8"), "base64")	1✔
89	assume_role_response = None	1✔
90	# Attempt to assume a role with the following durations:
91	# 12h, 8h, 6h, 4h, 2h, 1h, 30m, 15m
92	session_times = [43200, 28800, 21600, 14400, 7200, 3600, 1800, 900, "exit"]	1✔
93	for duration in session_times:	1✔
94	if duration == "exit":	1✔
95	logger.error(	×
96	default_error.format(
97	role_arn,
98	f"IAM role session time is not within set: {session_times[:-1]}",
99	)
100	)
101	sys.exit(2)	×
102
103	assume_role_response = handle_assume_role(	1✔
104	role_arn, provider_arn, encoded_xml, duration, default_error
105	)
106	if "Credentials" in assume_role_response:	1✔
107	break	1✔
108
109	return assume_role_response	1✔
110
111
112	def handle_assume_role(role_arn, provider_arn, encoded_xml, duration, default_error):	4✔
113	"""Handle assume role with saml.
114
115	:param role_arn: IAM role arn to assume
116	:param provider_arn: ARN of saml-provider resource
117	:param saml: decoded saml response from okta
118	:return: AssumeRoleWithSaml API responses
119	"""
120	logger.debug(f"Attempting session time [{duration}]")	1✔
121	try:	1✔
122	session = botocore.session.get_session()	1✔
123	client = session.create_client("sts", config=Config(signature_version=UNSIGNED))	1✔
124	assume_role_response = client.assume_role_with_saml(	1✔
125	RoleArn=role_arn,
126	PrincipalArn=provider_arn,
127	SAMLAssertion=encoded_xml.decode(),
128	DurationSeconds=duration,
129	)
130	# Client Exceptions
131	except ClientError as error:	1✔
132	if error.response["Error"]["Code"] == "ValidationError":	1✔
133	logger.info(	1✔
134	f"AssumeRoleWithSaml failed with {error.response['Error']['Code']} "
135	f"for duration {duration}"
136	)
137	assume_role_response = "continue"	1✔
138	elif error.response["Error"]["Code"] == "AccessDenied":	×
139	errmsg = f"Error assuming intermediate {provider_arn} SAML role"	×
140	logger.error(errmsg)	×
141	sys.exit(2)	×
142	else:
143	logger.error(default_error.format(role_arn, str(error)))	×
144	sys.exit(1)	×
145	# Service Exceptions
146	except Exception as error:	×
147	logger.error(default_error.format(role_arn, str(error)))	×
148	sys.exit(1)	×
149
150	return assume_role_response	1✔
151
152
153	def ensure_keys_work(assume_role_response):	4✔
154	"""Validate the temporary AWS credentials.
155
156	:param aws_access_key: AWS access key
157	:param aws_secret_key: AWS secret key
158	:param aws_session_token: AWS session token
159	:return:
160
161	"""
162	logger.debug("Validate the temporary AWS credentials")	1✔
163
164	aws_access_key = assume_role_response["Credentials"]["AccessKeyId"]	1✔
165	aws_secret_key = assume_role_response["Credentials"]["SecretAccessKey"]	1✔
166	aws_session_token = assume_role_response["Credentials"]["SessionToken"]	1✔
167
168	try:	1✔
169	session = botocore.session.get_session()	1✔
170
171	client = session.create_client(	1✔
172	"sts",
173	aws_access_key_id=aws_access_key,
174	aws_secret_access_key=aws_secret_key,
175	aws_session_token=aws_session_token,
176	)
177	client.get_caller_identity()	1✔
178	except Exception as auth_error:	×
179	logger.error(f"There was an error authenticating your keys with AWS: {auth_error}")	×
180	sys.exit(1)	×
181
182
183	def select_assumeable_role(apps):	4✔
184	"""Select the role to perform the AssumeRoleWithSaml.
185
186	# :param apps: apps metadata, list of tuples
187	# :return: AWS AssumeRoleWithSaml response, role name, tuple
188	"""
189	authenticated_aps = {}	1✔
190	for url, saml_response, saml, label in apps:	1✔
191	roles_and_providers = user.extract_arns(saml)	1✔
192	authenticated_aps[url] = {	1✔
193	"roles": list(roles_and_providers.keys()),
194	"saml": saml,
195	"saml_response_string": saml_response,
196	"roles_and_providers": roles_and_providers,
197	"label": label,
198	}
199
200	role_arn, _id = user.select_role_arn(authenticated_aps)	1✔
201	role_name = role_arn.split("/")[-1]	1✔
202
203	assume_role_response = assume_role(	1✔
204	role_arn,
205	authenticated_aps[_id]["roles_and_providers"][role_arn],
206	authenticated_aps[_id]["saml"],
207	)
208
209	return assume_role_response, role_name	1✔

dowjones / tokendito / 2692903695

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