import json import logging from urllib.parse import parse_qs, urlsplit import jwt import mockserver import nose.tools as nt from behave import step from mockserver import times from atlas_um.tokens.bearer_tokens import DNABearerToken logger = logging.getLogger(__name__) @step("I should get redirected to the USM authorization URL with") def step_should_get_redirected_to_usm_authorization_url_with(context): nt.assert_equal(context.response.status_code, 302) (scheme, netloc, path, query, fragment) = urlsplit( context.response.headers["Location"] ) context.usm_authorization_url_query_part = query nt.assert_equal(scheme, "https") nt.assert_equal(netloc, context.atlas_um_app.config["USM_HOSTNAME"]) nt.assert_equal(path, context.atlas_um_app.config["USM_AUTHORIZE_PATH"]) qs = parse_qs(query) for row in context.table: if row["value"] == "": nt.assert_in(row["query param"], qs) else: nt.assert_equal(qs[row["query param"]], [row["value"]]) nt.assert_equal(len(context.table.rows), len(qs)) nt.assert_equal(fragment, "") def fake_id_token(): return ( "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJqYXNvbi53aGl0dGxlLnNt" "ZUBzb255bXVzaWMuY29tIiwiYXVkIjoiUzNnaktuSTBVbTQxYUVkNVVWZG9hMmh0U2xjI" "iwiYXV0aF90aW1lIjoxNTk3OTQ0OTI5LCJpc3MiOiJodHRwczovL3VhdC11c20uc21lYW" "5hbHl0aWNzcG9ydGFsLmNvbS9vYXV0aDIvb3BlbmlkIiwicHJlZmVycmVkX3VzZXJuYW1" "lIjoiamFzb24ud2hpdHRsZS5zbWVAc29ueW11c2ljLmNvbSIsImdpdmVuX25hbWUiOiJK" "YXNvbiIsImV4cCI6MTU5Nzk0NjEyOSwibm9uY2UiOiJaYkRhSkxwUyIsImZhbWlseV9uY" "W1lIjoiV2hpdHRsZSIsImlhdCI6MTU5Nzk0NDkyOSwiZW1haWwiOiJqYXNvbi53aGl0dG" "xlLnNtZUBzb255bXVzaWMuY29tIiwiY2lkIjoiUzNnaktuSTBVbTQxYUVkNVVWZG9hMmh" "0U2xjIn0=.77+977+9A8uFfu+/vRvvv70S77+9UO+/vSx777+9aFsJ77+9Q39677+977+" "977+977+936fvv73vv70=" ) def fake_id_token_response(): return { "access_token": "d08trvm4h88c9de4i7gllb7lf6", "id_token": fake_id_token(), "token_type": "Bearer", "expires_in": 1200, } @step("USM is prepared to issue an ID token for me") def step_usm_is_prepared_to_issue_an_id_token_for_me(context): m_req = mockserver.request(method="POST", path="/oauth2/openid/v1/token") m_res = mockserver.response( code=200, body=json.dumps(fake_id_token_response()) ) context.mock_usm_client.expect(m_req, m_res, times(1)) @step("USM graph has the following data for me") def step_USM_graph_has_the_following_data_for_me(context): m_req = mockserver.request(method="POST", path="/graph/v1/getUser") response_data = { "success": True, "data": {row["field"]: row["value"] for row in context.table}, } m_res = mockserver.response(code=200, body=json.dumps(response_data)) context.mock_usm_client.expect(m_req, m_res, times(1)) @step("I complete my USM login and return to the login callback") def step_complete_my_usm_login_return_to_login_callback(context): """Depends on step_should_get_redirected_to_usm_authorization_url_with to set context.usm_authorization_url_query_part """ qs = parse_qs(context.usm_authorization_url_query_part) redirect_uri = qs["redirect_uri"][0] redirect_path = urlsplit(redirect_uri).path fake_code = context.faker.pystr() params = {"state": qs["state"], "code": fake_code} context.response = context.atlas_um_client.get( redirect_path, params, expect_errors=True ) @step("I should be assigned a valid DNA bearer token") def step_should_be_assigned_dna_bearer_token(context): dna_token = context.atlas_um_client.cookies["dna_bearer_token"] key = context.atlas_um_app.config["DNA_IDENTITY_PUBLIC_KEY"] validation_options = {"require": ["sub", "iat", "nbf", "exp"]} jwt.decode(dna_token, key, algorithms="RS256", options=validation_options) @step("No DNA bearer token should be assigned") def step_no_dna_bearer_token_should_be_assigned(context): nt.assert_not_in("dna_bearer_token", context.atlas_um_client.cookies) @step("I have a USM id token for that USM account") def step_have_usm_id_token_for_usm_account(context): context.atlas_um_client.set_cookie("usm_id_token", fake_id_token()) @step("I have a DNA bearer token for that DNA account") def step_have_dna_bearer_token_for_dna_account(context): """Depends on step_there_exists_dna_account to set context.dna_account""" nt.assert_is_not_none(context.dna_account) dna_bearer_token = DNABearerToken(context.dna_account).encode() context.atlas_um_client.set_cookie("dna_bearer_token", dna_bearer_token) @step("I should get redirected to the USM logout URL with") def step_should_get_redirected_to_usm_logout_url_with(context): nt.assert_equal(context.response.status_code, 302) (scheme, netloc, path, query, fragment) = urlsplit( context.response.headers["Location"] ) context.usm_logout_url_query_part = query nt.assert_equal(scheme, "https") nt.assert_equal(netloc, context.atlas_um_app.config["USM_HOSTNAME"]) nt.assert_equal(path, context.atlas_um_app.config["USM_LOGOUT_PATH"]) qs = parse_qs(query) for row in context.table: if row["value"] == "": nt.assert_equal(qs[row["query param"]], [fake_id_token()]) elif row["value"] == "": nt.assert_in(row["query param"], qs) else: nt.assert_equal(qs[row["query param"]], [row["value"]]) nt.assert_equal(len(context.table.rows), len(qs)) nt.assert_equal(fragment, "") @step("I complete my USM logout and return to the logout callback") def step_complete_my_usm_logout_return_to_logout_callback(context): """Depends on step_should_get_redirected_to_usm_logout_url_with to set context.usm_logout_url_query_part """ qs = parse_qs(context.usm_logout_url_query_part) redirect_uri = qs["post_logout_redirect_uri"][0] redirect_path = urlsplit(redirect_uri).path params = {"state": qs["state"]} context.response = context.atlas_um_client.get(redirect_path, params) @step("I should not have a USM id token") def step_should_not_have_usm_id_token(context): nt.assert_not_in("usm_id_token", context.atlas_um_client.cookies) @step("I should not have a DNA bearer token") def step_should_not_have_dna_bearer_token(context): nt.assert_not_in("dna_bearer_token", context.atlas_um_client.cookies)