diff --git a/.gitignore b/.gitignore index 2954c5dd..6c6ba063 100644 --- a/.gitignore +++ b/.gitignore @@ -91,6 +91,7 @@ venv/ ENV/ env.bak/ venv.bak/ +.vscode/ # Spyder project settings .spyderproject diff --git a/tokendito/__main__.py b/tokendito/__main__.py index 68fa0fba..ebe335b2 100755 --- a/tokendito/__main__.py +++ b/tokendito/__main__.py @@ -12,7 +12,7 @@ def main(args=None): # needed for console script path = os.path.dirname(os.path.dirname(__file__)) sys.path[0:0] = [path] - from tokendito.tool import cli + from tokendito.user import cli try: return cli(args) diff --git a/tokendito/okta.py b/tokendito/okta.py index cc0ef1c6..527a59dc 100644 --- a/tokendito/okta.py +++ b/tokendito/okta.py @@ -16,6 +16,9 @@ import time import bs4 +import base64 +import os +import hashlib from bs4 import BeautifulSoup import requests from tokendito import duo @@ -73,6 +76,7 @@ def api_error_code_parser(status=None): param status: Response status return message: status message """ + logger.debug(f"api_error_code_parser({status})") if status and status in _status_dict.keys(): message = f"Okta auth failed: {_status_dict[status]}" else: @@ -82,17 +86,41 @@ def api_error_code_parser(status=None): return message +def get_auth_pipeline(url=None): + """Get auth pipeline version.""" + logger.debug(f"get_auth_pipeline({url})") + headers = {"accept": "application/json"} + # https://developer.okta.com/docs/api/openapi/okta-management/management/tag/OrgSetting/ + url = f"{url}/.well-known/okta-organization" + + response = user.request_wrapper("GET", url, headers=headers) + try: + ret = response.json() + except (KeyError, ValueError) as e: + logger.error(f"Failed to parse type in {url}:{str(e)}") + logger.debug(f"Response: {response.text}") + sys.exit(1) + logger.debug(f"we have {ret}") + auth_pipeline = ret.get("pipeline", None) + if auth_pipeline != "idx" and auth_pipeline != "v1": + logger.error(f"unsupported auth pipeline version {auth_pipeline}") + sys.exit(1) + logger.debug(f"Pipeline is of type {auth_pipeline}") + return auth_pipeline + + def get_auth_properties(userid=None, url=None): - """Make a call to the webfinger endpoint. + """Make a call to the webfinger endpoint to get the auth properties metadata :param userid: User for which we are requesting an auth endpoint. :param url: Site where we are looking up the user. :returns: dictionary with authentication properties. """ + logger.debug(f"get_auth_properies({userid}, {url})") payload = {"resource": f"okta:acct:{userid}", "rel": "okta:idp"} + # payload = {"resource": f"okta:acct:{userid}"} headers = {"accept": "application/jrd+json"} url = f"{url}/.well-known/webfinger" - logger.debug(f"Looking up auth endpoint for {userid} in {url}") response = user.request_wrapper("GET", url, headers=headers, params=payload) @@ -195,7 +223,6 @@ def get_session_token(config, primary_auth, headers): :param primary_auth: Primary authentication :return: Session Token from JSON response """ - status = None try: status = primary_auth.get("status", None) except AttributeError: @@ -207,42 +234,234 @@ def get_session_token(config, primary_auth, headers): session_token = mfa_challenge(config, headers, primary_auth) else: logger.debug(f"Error parsing response: {json.dumps(primary_auth)}") - logger.error("Okta auth failed: unknown status.") + logger.error(f"Okta auth failed: unknown status {status}") sys.exit(1) user.add_sensitive_value_to_be_masked(session_token) return session_token +def get_oauth_token(token_endpoint_url, authz_code): + """Get OAuth token from Okta. + + :param url: URL of the Okta OAuth token endpoint + todo:, authz_code in payload, put call etc. + :return: OAuth token + """ + # payload = {"code": f"{authz_code}"} + # payload = {"resource": f"okta:acct:{userid}"} + # headers = {"accept": "application/jrd+json"} + # response = user.request_wrapper("GET", token_endpoint_url , headers=headers, params=payload) + + headers = {"accept": "text/html,application/xhtml+xml,application/xml"} + payload = {"code": authz_code} + response = user.request_wrapper("POST", token_endpoint_url, headers=headers, data=payload) + return response.json() + +def extract_authz_code(response_text): + """extract authorization code from response text + :param response_text: response text from /authorize call + :return: authorization code + """ + authz_code = re.search(r"(?<=code=)[^&]+", response_text).group(0) + return authz_code + +def get_client_id(): + """TODO""" + return "UNIQ_CLIENT_ID_1123" + +def get_redirect_uri(org_url): + """TODO""" + breakpoint() + return f"{org_url}/enduser/callback" + +def get_response_type(): + """TODO + https://developer.okta.com/docs/guides/implement-grant-type/authcode/main/#flow-specifics + https://www.oauth.com/oauth2-servers/pkce/authorization-request/ + https://developer.okta.com/docs/guides/implement-grant-type/authcodepkce/main/#next-steps + """"" + return "code" + +def get_authorize_scope(): + """TODO + https://developer.okta.com/docs/reference/api/oidc/#access-token-scopes-and-claims + https://developer.okta.com/docs/guides/implement-grant-type/authcodepkce/main/#next-steps + """ + return "openid" + +def get_oauth_state(): + """generate a random string for state + https://developer.okta.com/docs/guides/implement-grant-type/authcode/main/#flow-specifics + https://developer.okta.com/docs/guides/implement-grant-type/authcodepkce/main/#next-steps + """ + return "ramdom_state" + +def get_pkce_code_challenge_method(): + + """TODO""" + return "S256" + + +def get_pkce_code_challenge(code_verifier=None): + """ + get_pkce_code_challenge + + Base64-URL-encoded string of the SHA256 hash of the code verifier + https://www.oauth.com/oauth2-servers/pkce/authorization-request/ + + :param: code_verifier + :return: code_challenge + """ + if code_verifier is None: + code_verifier = get_pkce_code_verifier() + code_challenge = base64.urlsafe_b64encode( + hashlib.sha256(code_verifier.encode('utf-8')).digest() + ).decode('utf-8') + return code_challenge + +def get_pkce_code_verifier(): + """ + to review + """ + code_verifier = base64.urlsafe_b64encode(os.urandom(32)).decode('utf-8') + return code_verifier + +def pkce_enabled(): + """TODO""" + return True + +def oauth_authz_code_request(authz_server_url, authn_sid): + """implements authorization code request + calls /authorize endpoint with authenticated sid. + https://developer.okta.com/docs/reference/api/oidc/#_2-okta-as-the-identity-platform-for-your-app-or-api + :param + :return: authorization code, needed for /token call + """ + logger.debug(f"authz_code_reqquest({authz_server_url}, {authn_sid})") + headers = {"accept": "text/html,application/xhtml+xml,application/xml"} + # headers = {"accept": "application/json"} + client_id = get_client_id() + redirect_uri = get_redirect_uri(authz_server_url) + response_type = get_response_type() + scope = get_authorize_scope() + oauth_state = get_oauth_state() + + if pkce_enabled(): + code_verifier = get_pkce_code_verifier() + code_challenge = get_pkce_code_challenge(code_verifier) + code_challenge_method = get_pkce_code_challenge_method() + payload = {"client_id": client_id, "redirect_url": redirect_uri, "response_type": response_type, "scope": scope, "state": oauth_state, "code_challenge": code_challenge, "code_challenge_method": code_challenge_method} + # todo; authn_sid? + breakpoint() + response = user.request_wrapper("GET", authz_server_url, headers=headers, data=payload) + authz_code = extract_authz_code(response.text) + +def authorization_code_workflow(config, authn_sid): + # Authorization Code flow (see + # https://developer.okta.com/docs/guides/implement-grant-type/authcode/main/#about-the-authorization-code-grant + # ) + logger.debug(f"authorization_code_worflow({config},{authn_sid})") + authz_endpoint_url, token_endpoint_url = get_authorization_server_endpoints(config.okta["org"]) + authz_code = oauth_authz_code_request(authz_endpoint_url, authn_sid) + user.add_sensitive_value_to_be_masked(authz_code) + + authz_token = get_oauth_token(token_endpoint_url, authz_code) + user.add_sensitive_value_to_be_masked(authz_token) + return authz_token + +def authorization_code_enabled(config): + """determines if authorization code grant is enabled + todo + """ + return True + +def get_authorization_server_endpoints(url=None): + """Get /authorize and /token endpoints from Okta -def authenticate(config): - """Authenticate user. + :param url: URL of the Okta + :return: tuple of URLs of the /authorize and /token endpoints + """ + url = f"{url}/.well-known/oauth-authorization-server" + headers = {"accept": "application/json"} + response = user.request_wrapper("GET", url, headers=headers) + logger.info(f"Authorization Server info: {response.json()}") + # todo: handle errors. + return response.json()["authorization_endpoint"], response.json()["token_endpoint"] + +def oie_auth(config, authn_sid): + logger.debug(f"oie_auth({config}, {authn_sid})") + # get url where to authorize + if authorization_code_enabled(config): + sid = authorization_code_workflow(config, authn_sid) + else: + logger.warning("Authorization Code is not enabled, basically skipping oie.") + sid = authn_sid + return sid + + +def idp_auth(config): + """authenticate and authorize with the idp :param config: Config object :return: session ID cookie. """ + + logger.debug(f"idp_auth({config})") auth_properties = get_auth_properties(userid=config.okta["username"], url=config.okta["org"]) + if "type" not in auth_properties: logger.error("Okta auth failed: unknown type.") sys.exit(1) + sid = None - if is_local_auth(auth_properties): - session_token = local_auth(config) + + if is_local_authn(auth_properties): + session_token = local_authn(config) sid = user.request_cookies(config.okta["org"], session_token) elif is_saml2_auth(auth_properties): sid = saml2_auth(config, auth_properties) else: logger.error(f"{auth_properties['type']} login via IdP Discovery is not curretly supported") sys.exit(1) + + if oie_enabled(config.okta["org"]): + sid = oie_auth(config, sid) + return sid +def oie_enabled(url): + """ + Determines if OIE is enabled. + :pamam url: okta org url + :return: True if OIE is enabled, False otherwise + """ + if get_auth_pipeline(url) == "idx": # oie + return True + else: + return False + + +def local_authn(config): + """Authenticate user on local okta instance. + + :param config: Config object + :return: auth session ID cookie. + """ + + url = config.okta["org"] + + authn_sid = idp_authn(config) + user.add_sensitive_value_to_be_masked(authn_sid) + return authn_sid + -def is_local_auth(auth_properties): - """Check whether authentication happens locally. +def is_local_authn(auth_properties): + """Check whether authentication happens on the current instance. :param auth_properties: auth_properties dict - :return: True for local auth, False otherwise. + :return: True if this is the place to authenticate, False otherwise. """ try: if auth_properties["type"] == "OKTA": @@ -266,12 +485,13 @@ def is_saml2_auth(auth_properties): return False -def local_auth(config): - """Authenticate local user with okta credential. +def idp_authn(config): + """Authenticate with okta. :param config: Config object :return: MFA session with options """ + logger.debug(f"idp_authn({config}") session_token = None headers = {"content-type": "application/json", "accept": "application/json"} payload = {"username": config.okta["username"], "password": config.okta["password"]} @@ -303,7 +523,7 @@ def saml2_auth(config, auth_properties): logger.info(f"Authentication is being redirected to {saml2_config.okta['org']}.") # Try to authenticate using the new configuration. This could cause # recursive calls, which allows for IdP chaining. - session_cookies = authenticate(saml2_config) + session_cookies = idp_authn(saml2_config) # Once we are authenticated, send the SAML request to the IdP. # This call requires session cookies. diff --git a/tokendito/tokendito.py b/tokendito/tokendito.py index 26c04fbb..07e867b5 100755 --- a/tokendito/tokendito.py +++ b/tokendito/tokendito.py @@ -1,7 +1,8 @@ #!/usr/bin/env python # vim: set filetype=python ts=4 sw=4 # -*- coding: utf-8 -*- -"""tokendito cli entry point.""" +"""tokendito entry point.""" +import logging import sys @@ -12,8 +13,7 @@ def main(args=None): # needed for console script path = os.path.dirname(os.path.dirname(__file__)) sys.path[0:0] = [path] - from tokendito.tool import cli - + from tokendito.user import cli return cli(args) diff --git a/tokendito/tool.py b/tokendito/tool.py deleted file mode 100644 index 26125579..00000000 --- a/tokendito/tool.py +++ /dev/null @@ -1,70 +0,0 @@ -# vim: set filetype=python ts=4 sw=4 -# -*- coding: utf-8 -*- -"""This module retrieves AWS credentials after authenticating with Okta.""" -import logging -import sys - -from tokendito import aws -from tokendito import config -from tokendito import okta -from tokendito import user - -logger = logging.getLogger(__name__) - - -def cli(args): - """Tokendito retrieves AWS credentials after authenticating with Okta.""" - args = user.parse_cli_args(args) - - # Early logging, in case the user requests debugging via env/CLI - user.setup_early_logging(args) - - # Set some required initial values - user.process_options(args) - - # Late logging (default) - user.setup_logging(config.user) - - # Validate configuration - message = user.validate_configuration(config) - if message: - quiet_msg = "" - if config.user["quiet"] is not False: - quiet_msg = " to run in quiet mode" - logger.error( - f"Could not validate configuration{quiet_msg}: {'. '.join(message)}. " - "Please check your settings, and try again." - ) - sys.exit(1) - - # Authenticate to okta - session_cookies = okta.authenticate(config) - - if config.okta["tile"]: - tile_label = "" - config.okta["tile"] = (config.okta["tile"], tile_label) - else: - config.okta["tile"] = user.discover_tiles(config.okta["org"], session_cookies) - - # Authenticate to AWS roles - auth_tiles = aws.authenticate_to_roles(config.okta["tile"], cookies=session_cookies) - - (role_response, role_name) = aws.select_assumeable_role(auth_tiles) - - identity = aws.assert_credentials(role_response=role_response) - if "Arn" not in identity and "UserId" not in identity: - logger.error( - f"There was an error retrieving and verifying AWS credentials: {role_response}" - ) - sys.exit(1) - - user.set_profile_name(config, role_name) - - user.set_local_credentials( - response=role_response, - role=config.aws["profile"], - region=config.aws["region"], - output=config.aws["output"], - ) - - user.display_selected_role(profile_name=config.aws["profile"], role_response=role_response) diff --git a/tokendito/user.py b/tokendito/user.py index 9fff2fb8..dce5c16d 100644 --- a/tokendito/user.py +++ b/tokendito/user.py @@ -21,10 +21,11 @@ from bs4 import __version__ as __bs4_version__ # type: ignore (bs4 does not have PEP 561 support) from bs4 import BeautifulSoup import requests -from tokendito import __version__ +from tokendito import config from tokendito import aws +from tokendito import okta +from tokendito import __version__ from tokendito import Config -from tokendito import config as config # Unfortunately, readline is only available in non-Windows systems. There is no substitution. try: @@ -34,9 +35,64 @@ logger = logging.getLogger(__name__) - mask_items = [] +def cli(args): + """Tokendito retrieves AWS credentials after authenticating with Okta.""" + args = parse_cli_args(args) + + # Early logging, in case the user requests debugging via env/CLI + setup_early_logging(args) + + # Set some required initial values + process_options(args) + + # Late logging (default) + setup_logging(config.user) + + # Validate configuration + message = validate_configuration(config) + if message: + quiet_msg = "" + if config.user["quiet"] is not False: + quiet_msg = " to run in quiet mode" + logger.error( + f"Could not validate configuration{quiet_msg}: {'. '.join(message)}. " + "Please check your settings, and try again." + ) + sys.exit(1) + + # get authentication and authorization cookies from okta + session_cookies = okta.idp_auth(config) + + if config.okta["tile"]: + tile_label = "" + config.okta["tile"] = (config.okta["tile"], tile_label) + else: + config.okta["tile"] = discover_tiles(config.okta["org"], session_cookies) + + # Authenticate to AWS roles + auth_tiles = aws.authenticate_to_roles(config.okta["tile"], cookies=session_cookies) + + (role_response, role_name) = aws.select_assumeable_role(auth_tiles) + + identity = aws.assert_credentials(role_response=role_response) + if "Arn" not in identity and "UserId" not in identity: + logger.error( + f"There was an error retrieving and verifying AWS credentials: {role_response}" + ) + sys.exit(1) + + set_profile_name(config, role_name) + + set_local_credentials( + response=role_response, + role=config.aws["profile"], + region=config.aws["region"], + output=config.aws["output"], + ) + + display_selected_role(profile_name=config.aws["profile"], role_response=role_response) class MaskLoggerSecret(logging.Filter): """Masks secrets in logger messages.""" @@ -54,6 +110,8 @@ def filter(self, record): return True + + def parse_cli_args(args): """Parse command line arguments. @@ -1192,6 +1250,7 @@ def sanitize_config_values(config): config.aws["shared_credentials_file"] = os.path.expanduser( config.aws["shared_credentials_file"] ) + return config