From 2a97bce08534cadb06a103cdf1ee476bab8d25f5 Mon Sep 17 00:00:00 2001 From: Jean-Pierre Sevigny <41591249+sevignyj@users.noreply.github.com> Date: Wed, 16 Aug 2023 10:36:47 -0400 Subject: [PATCH] oie implementation WIP, oauth2 only, authorization code flow. --- .gitignore | 1 + tokendito/__init__.py | 1 + tokendito/__main__.py | 2 +- tokendito/okta.py | 303 ++++++++++++++++++++++++++++++++++++++--- tokendito/tokendito.py | 5 +- tokendito/tool.py | 70 ---------- tokendito/user.py | 93 ++++++++++++- 7 files changed, 379 insertions(+), 96 deletions(-) delete mode 100644 tokendito/tool.py 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/__init__.py b/tokendito/__init__.py index 5d2fab86..1ae94640 100644 --- a/tokendito/__init__.py +++ b/tokendito/__init__.py @@ -53,6 +53,7 @@ class Config(object): password="", mfa=None, mfa_response=None, + oauth_client_id=None, tile=None, org=None, ), 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..7e5851c0 100644 --- a/tokendito/okta.py +++ b/tokendito/okta.py @@ -14,8 +14,10 @@ import re import sys import time - +import base64 import bs4 +import os +import hashlib from bs4 import BeautifulSoup import requests from tokendito import duo @@ -73,6 +75,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 +85,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 +222,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,7 +233,7 @@ 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) @@ -215,34 +241,272 @@ def get_session_token(config, primary_auth, headers): return session_token -def authenticate(config): - """Authenticate user. +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(config): + """TODO + https://datatracker.ietf.org/doc/html/rfc6749#section-2.2 + 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 + """ + + if config.okta['oauth_client_id'] is None: + logger.error("oauth client_id is not set.") + sys.exit(1) + + return config.okta['oauth_client_id'] + + +def get_redirect_uri(config): + """TODO""" + # return "http://localhost:8080/authorization-code/callback" + url = f"{config.okta['org']}enduser/callback" + return url + +def get_response_type(): + """TODO""" + 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 profile email okta.users.read.self okta.users.manage.self okta.internal.enduser.read okta.internal.enduser.manage okta.enduser.dashboard.read okta.enduser.dashboard.manage" + + +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 + """ + state = hashlib.sha256(os.urandom(1024)).hexdigest() + return 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 + """ + + code_challenge = hashlib.sha256(code_verifier.encode('utf-8')).digest() + code_challenge = base64.urlsafe_b64encode(code_challenge).decode('utf-8') + code_challenge = code_challenge.replace('=', '') + return code_challenge + #return base64.urlsafe_b64encode(hashlib.sha256(code_verifier).digest()) + + + +# +# 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") + # code_verifier = base64.urlsafe_b64encode(os.urandom(50)).rstrip(b'=') + + code_verifier = base64.urlsafe_b64encode(os.urandom(40)).decode('utf-8') + code_verifier = re.sub('[^a-zA-Z0-9]+', '', code_verifier) + return code_verifier + + +def pkce_enabled(): + """TODO""" + return True + + +def oauth_authz_code_request(config, 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(config) + redirect_uri = get_redirect_uri(config) + response_type = get_response_type() + scope = get_authorize_scope() + oauth_state = get_oauth_state() + + session_cookies = None + + 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_uri": redirect_uri, + "response_type": response_type, + "scope": scope, + "state": oauth_state, + "code_challenge": code_challenge, + "code_challenge_method": code_challenge_method, + } + response = user.request_wrapper("GET", authz_server_url, params=payload) + session_cookies = response.cookies + logger.debug(f"Have session cookies: {session_cookies}") + + return session_cookies + #logger.debug(f"authz_code_reqquest({authz_server_url}, {authn_sid}) response: {response.text}") + #authz_code = extract_authz_code(response.text) + + +def authorization_code_flow(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(config, 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 + + :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_authz(config, authn_sid): + logger.debug(f"oie_auth({config}, {authn_sid})") + # get url where to authorize + if authorization_code_enabled(config): + sid = authorization_code_flow(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. Authorize heppens if oie is enabled. :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) - sid = user.request_cookies(config.okta["org"], session_token) + session_cookies = None + + if is_local_authn(auth_properties): + session_token = local_authn(config) + session_cookies = user.request_cookies(config.okta["org"], session_token) elif is_saml2_auth(auth_properties): - sid = saml2_auth(config, auth_properties) + session_cookies = saml2_auth(config, auth_properties) else: logger.error(f"{auth_properties['type']} login via IdP Discovery is not curretly supported") sys.exit(1) - return sid + + if oie_enabled(config.okta["org"]): + session_cookies = oie_authz(config, session_cookies) + + return session_cookies + + +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. + """ + + 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 +530,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 +568,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..5eaf9b60 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,7 +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..2ee9b879 100644 --- a/tokendito/user.py +++ b/tokendito/user.py @@ -17,14 +17,19 @@ import sys from urllib.parse import urlparse + +# to debug http messages +import http.client as http_client + from botocore import __version__ as __botocore_version__ 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,10 +39,68 @@ 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.""" @@ -131,6 +194,12 @@ def parse_cli_args(args): "--okta-tile", help="Okta tile URL to use.", ) + # oauth_client_id + + parser.add_argument( + "--okta-oauth-client-id", + help="Sets the Okta client ID needed in oauth2.", + ) parser.add_argument( "--okta-mfa", help="Sets the MFA method. You " @@ -151,6 +220,7 @@ def parse_cli_args(args): parsed_args = parser.parse_args(args) + return parsed_args @@ -624,7 +694,9 @@ def process_arguments(args): res = dict() pattern = re.compile(r"^(.*?)_(.*)") + for key, val in vars(args).items(): + logger.debug(f"key is {key} and val is {val}") match = re.search(pattern, key.lower()) if match: if match.group(1) not in get_submodule_names(): @@ -636,6 +708,7 @@ def process_arguments(args): add_sensitive_value_to_be_masked(val, match.group(2)) logger.debug(f"Found arguments: {res}") + try: config_args = Config(**res) @@ -1192,6 +1265,7 @@ def sanitize_config_values(config): config.aws["shared_credentials_file"] = os.path.expanduser( config.aws["shared_credentials_file"] ) + return config @@ -1267,7 +1341,18 @@ def request_wrapper(method, url, headers=None, **kwargs): if headers is None: headers = {"content-type": "application/json", "accept": "application/json"} - logger.debug(f"Issuing {method} request to {url} with {headers} and {kwargs}") + if logging.getLogger(__name__).level == logging.DEBUG: + # TODO : secrets are not masked in debug mode. + # set http log level to debug if we're in debug + pass # uncomment the next 6 lines to debug http requests + #import http.client as httplib + + #http_client.HTTPConnection.debuglevel = 1 + #requests_log = logging.getLogger("requests.packages.urllib3") + #requests_log.setLevel(logging.DEBUG) + #requests_log.propagate = True + + logger.debug(f"Issuing response = requests.request({method}, {url}, {headers}, {kwargs})") try: response = requests.request(method=method, url=url, headers=headers, **kwargs) response.raise_for_status()