import flask import pytest from flask import url_for from flask_webtest import TestApp as WebtestApp from atlas_um import consts from atlas_um.app import create_app from atlas_um.auth.claims import Audience from atlas_um.consts import SystemEvents from atlas_um.helpers.either import Left, Right from atlas_um.pgdb import pgdb from atlas_um.pgdb.dna_account import DNAAccountStatuses, DNAAccount from tests.atlas_um.factories import DNAAccountFactory, ResourceGroupFactory from tests.atlas_um.pytest_helpers import get_cookie from atlas_um.usm.views import login class TestUSMLogin: FAKE_HOST = "atlas_um.test" FAKE_SUBDOMAIN_HOST = "sub.atlas_um.test" FAKE_AUTH_PROXY_DOMAIN = f"auth.{FAKE_HOST}" AUTH_PROXY_PARAM = f"https://{FAKE_AUTH_PROXY_DOMAIN}/process" @pytest.fixture def app(self, request): app = create_app() app.testing = True app.config["SERVER_NAME"] = self.FAKE_HOST app.config["ALLOWED_HOSTS"] = [ self.FAKE_HOST, self.FAKE_AUTH_PROXY_DOMAIN, ] ctx = app.app_context() ctx.push() @request.addfinalizer def teardown(): ctx.pop() return app @pytest.fixture def client(self, app): return WebtestApp(app) @pytest.fixture def pgdb_session(self, request, app): connection = pgdb.engine.connect() transaction = connection.begin() options = dict(bind=connection, binds={}) session = pgdb.create_scoped_session(options=options) pgdb.session = session @request.addfinalizer def teardown(): transaction.rollback() connection.close() session.remove() return session def test_login_no_info(self, app, faker, mocker): fake_state = faker.pystr() mocked_generate_state = mocker.patch( "atlas_um.usm.client.USMClient.generate_state" ) mocked_generate_state.return_value = fake_state fake_nonce = faker.pystr() mocked_generate_nonce = mocker.patch( "atlas_um.usm.client.USMClient.generate_nonce" ) mocked_generate_nonce.return_value = fake_nonce fake_authentication_url = faker.uri() mocked_auth_url = mocker.patch( "atlas_um.usm.client.USMClient.authentication_url" ) mocked_auth_url.return_value = fake_authentication_url with app.app_context(): req = url_for("usm_login.get_login") test_referrer = faker.url() with app.test_client() as client: resp = client.get(req, headers={"Referer": test_referrer}) session = flask.session.copy() assert resp.status_code == 302 assert resp.location == fake_authentication_url assert session[consts.STATE_SESSION_NAME] == fake_state assert session[consts.NONCE_SESSION_NAME] == fake_nonce assert mocked_generate_state.call_args_list == [()] assert mocked_generate_nonce.call_args_list == [()] assert mocked_auth_url.call_args_list == [ mocker.call(fake_state, fake_nonce, None) ] assert test_referrer == session.get(consts.REFERRER_SESSION_NAME) def test_login_with_prompt(self, app, faker, mocker): fake_state = faker.pystr() mocked_generate_state = mocker.patch( "atlas_um.usm.client.USMClient.generate_state" ) mocked_generate_state.return_value = fake_state fake_nonce = faker.pystr() fake_prompt = faker.pystr() mocked_generate_nonce = mocker.patch( "atlas_um.usm.client.USMClient.generate_nonce" ) mocked_generate_nonce.return_value = fake_nonce fake_authentication_url = faker.uri() mocked_auth_url = mocker.patch( "atlas_um.usm.client.USMClient.authentication_url" ) mocked_auth_url.return_value = fake_authentication_url with app.app_context(): req = url_for("usm_login.get_login", prompt=fake_prompt) test_referrer = faker.url() with app.test_client() as client: resp = client.get(req, headers={"Referer": test_referrer}) session = flask.session.copy() assert resp.status_code == 302 assert resp.location == fake_authentication_url assert session[consts.STATE_SESSION_NAME] == fake_state assert session[consts.NONCE_SESSION_NAME] == fake_nonce assert mocked_generate_state.call_args_list == [()] assert mocked_generate_nonce.call_args_list == [()] assert mocked_auth_url.call_args_list == [ mocker.call(fake_state, fake_nonce, fake_prompt) ] assert test_referrer == session.get(consts.REFERRER_SESSION_NAME) def test_login_with_nonce_param(self, app, faker, mocker): fake_nonce = faker.pystr() fake_state = faker.pystr() mocked_generate_state = mocker.patch( "atlas_um.usm.client.USMClient.generate_state" ) mocked_generate_state.return_value = fake_state fake_authentication_url = faker.uri() mocked_auth_url = mocker.patch( "atlas_um.usm.client.USMClient.authentication_url" ) mocked_auth_url.return_value = fake_authentication_url with app.app_context(): req = url_for("usm_login.get_login", nonce=fake_nonce) test_referrer = faker.url() with app.test_client() as client: resp = client.get(req, headers={"Referer": test_referrer}) session = flask.session.copy() assert resp.status_code == 302 assert resp.location == fake_authentication_url assert session[consts.STATE_SESSION_NAME] == fake_state assert session[consts.NONCE_SESSION_NAME] == fake_nonce assert mocked_generate_state.call_args_list == [()] assert mocked_auth_url.call_args_list == [ mocker.call(fake_state, fake_nonce, None) ] assert test_referrer == session.get(consts.REFERRER_SESSION_NAME) def test_login_no_mfa_disabled(self, app, faker, mocker): app.config["NO_MFA_ENABLED"] = False fake_nonce = faker.pystr() fake_state = faker.pystr() mocked_generate_state = mocker.patch( "atlas_um.usm.client.USMClient.generate_state" ) mocked_generate_state.return_value = fake_state fake_authentication_url = faker.uri() mocked_auth_url = mocker.patch( "atlas_um.usm.client.USMClient.authentication_url" ) mocked_auth_url.return_value = fake_authentication_url with app.app_context(): req = url_for("usm_login.get_login", nonce=fake_nonce) test_referrer = faker.url() with app.test_client() as client: resp = client.get(req, headers={"Referer": test_referrer}) session = flask.session.copy() assert resp.status_code == 302 assert resp.location == fake_authentication_url assert session[consts.STATE_SESSION_NAME] == fake_state assert session[consts.NONCE_SESSION_NAME] == fake_nonce assert mocked_generate_state.call_args_list == [()] assert mocked_auth_url.call_args_list == [ mocker.call(fake_state, fake_nonce, None) ] assert test_referrer == session.get(consts.REFERRER_SESSION_NAME) def test_login_no_mfa_enabled_without_referrer( self, app, faker, mocker, pgdb_session ): app.config["NO_MFA_ENABLED"] = True mocked_no_mfa_is_active = mocker.patch( "atlas_um.auth.utils.current_user" ) mocked_no_mfa_is_active.claims = [Audience(Audience.Values.no_mfa)] fake_token = faker.pystr() mocked_bearer_token = faker.pystr() mocked_refresh_token = faker.pystr() mocker.patch( "tldextract.extract" ).return_value.registered_domain = self.FAKE_HOST mocked_user_events = mocker.patch( "atlas_um.analytics.collectors.Collector.log" ) mocked_register_for_resource_group = mocker.patch( "atlas_um.pgdb.DNAAccountActivity.register_for_resource_group" ) mocked_create_bearer_token_service = mocker.patch( "atlas_um.tokens.services.CreateBearerTokenService.execute" ) mocked_create_bearer_token_service.return_value = Right( mocked_bearer_token ) mocked_create_refresh_token_service = mocker.patch( "atlas_um.tokens.services.CreateRefreshTokenService.execute" ) mocked_create_refresh_token_service.return_value = Right( mocked_refresh_token ) with app.test_client() as client: dna_account = DNAAccountFactory.build(no_mfa=True) resource_group = ResourceGroupFactory.build( service_url=f"https://{self.FAKE_HOST}", namespace_url="atlasum", ) pgdb_session.add(dna_account) pgdb_session.add(resource_group) pgdb_session.commit() req = url_for( "usm_login.get_login", **{ consts.DEV_TOKEN_PARAM: fake_token, consts.NO_MFA_ACCOUNT_PARAM: dna_account.email, }, ) resp = client.get(req) assert mocked_register_for_resource_group.call_args_list == [ mocker.call(dna_account, resource_group) ] assert mocked_create_refresh_token_service.call_args_list == [ mocker.call(dna_account, resource_group) ] assert resp.status_code == 302 bearer_cookie = get_cookie( resp, app.config.get("DNA_BEARER_TOKEN_COOKIE_NAME") ) assert bearer_cookie is not None assert "HttpOnly" in bearer_cookie assert "Secure" in bearer_cookie assert bearer_cookie.get("Domain") == self.FAKE_HOST assert ( int(bearer_cookie["Max-Age"]) == app.config["TOKEN_COOKIES_MAX_AGE"] ) assert ( bearer_cookie.get(app.config.get("DNA_BEARER_TOKEN_COOKIE_NAME")) == mocked_bearer_token ) refresh_cookie = get_cookie( resp, app.config.get("DNA_REFRESH_TOKEN_COOKIE_NAME") ) assert refresh_cookie is not None assert "HttpOnly" in refresh_cookie assert "Secure" in refresh_cookie assert refresh_cookie.get("Domain") == self.FAKE_HOST assert ( int(refresh_cookie["Max-Age"]) == app.config["TOKEN_COOKIES_MAX_AGE"] ) assert ( refresh_cookie.get(app.config.get("DNA_REFRESH_TOKEN_COOKIE_NAME")) == mocked_refresh_token ) assert mocked_user_events.call_args_list == [ mocker.call( SystemEvents.successful_login, domain=self.FAKE_HOST, user_id=dna_account.sub, ) ] with app.app_context(): assert resp.location == url_for("common.userinfo") def test_login_no_mfa_enabled_with_auth_proxy( self, app, faker, mocker, pgdb_session ): app.config["NO_MFA_ENABLED"] = True mocked_no_mfa_is_active = mocker.patch( "atlas_um.auth.utils.current_user" ) mocked_no_mfa_is_active.claims = [Audience(Audience.Values.no_mfa)] fake_token = faker.pystr() mocked_bearer_token = faker.pystr() mocked_refresh_token = faker.pystr() mocker.patch( "atlas_um.usm.views.login.get_allowed_root_domain" ).return_value = self.FAKE_HOST mocker.patch( "atlas_um.usm.views.login.get_allowed_subdomain" ).return_value = None mocker.patch( "atlas_um.usm.views.login.get_root_domain" ).return_value = self.FAKE_HOST mocked_user_events = mocker.patch( "atlas_um.analytics.collectors.Collector.log" ) mocked_register_for_resource_group = mocker.patch( "atlas_um.pgdb.DNAAccountActivity.register_for_resource_group" ) mocked_create_refresh_token_service = mocker.patch( "atlas_um.tokens.services.CreateRefreshTokenService.execute" ) mocked_create_refresh_token_service.return_value = Right( mocked_refresh_token ) mocked_create_bearer_token_service = mocker.patch( "atlas_um.tokens.services.CreateBearerTokenService.execute" ) mocked_create_bearer_token_service.return_value = Right( mocked_bearer_token ) mocker.spy(login, "render_template") with app.test_client() as client: dna_account = DNAAccountFactory.build(no_mfa=True) resource_group = ResourceGroupFactory.build( service_url=f"https://{self.FAKE_HOST}", namespace_url="atlasum", ) pgdb_session.add(dna_account) pgdb_session.add(resource_group) pgdb_session.commit() url = url_for( "usm_login.get_login", **{ consts.DEV_TOKEN_PARAM: fake_token, consts.NO_MFA_ACCOUNT_PARAM: dna_account.email, consts.AUTH_PROXY_SESSION_NAME: self.AUTH_PROXY_PARAM, }, ) resp = client.get(url) assert mocked_register_for_resource_group.call_args_list == [ mocker.call(dna_account, resource_group), mocker.call(dna_account, resource_group), ] assert mocked_create_refresh_token_service.call_args_list == [ mocker.call(dna_account, resource_group), mocker.call(dna_account, resource_group), ] assert resp.status_code == 200 assert login.render_template.call_args_list == [ mocker.call( "auth_proxy.html", auth_proxy=self.AUTH_PROXY_PARAM, access_token=mocked_bearer_token, refresh_token=mocked_refresh_token, back_to=url_for("common.userinfo"), ) ] assert self.AUTH_PROXY_PARAM in resp.data.decode() bearer_cookie = get_cookie( resp, app.config.get("DNA_BEARER_TOKEN_COOKIE_NAME") ) assert bearer_cookie is not None assert "HttpOnly" in bearer_cookie assert "Secure" in bearer_cookie assert bearer_cookie.get("Domain") == self.FAKE_HOST assert ( int(bearer_cookie["Max-Age"]) == app.config["TOKEN_COOKIES_MAX_AGE"] ) assert ( bearer_cookie.get(app.config.get("DNA_BEARER_TOKEN_COOKIE_NAME")) == mocked_bearer_token ) refresh_cookie = get_cookie( resp, app.config.get("DNA_REFRESH_TOKEN_COOKIE_NAME") ) assert refresh_cookie is not None assert "HttpOnly" in refresh_cookie assert "Secure" in refresh_cookie assert refresh_cookie.get("Domain") == self.FAKE_HOST assert ( int(refresh_cookie["Max-Age"]) == app.config["TOKEN_COOKIES_MAX_AGE"] ) assert ( refresh_cookie.get(app.config.get("DNA_REFRESH_TOKEN_COOKIE_NAME")) == mocked_refresh_token ) assert mocked_user_events.call_args_list == [ mocker.call( SystemEvents.successful_login, domain=self.FAKE_HOST, user_id=dna_account.sub, ) ] def test_login_no_mfa_enabled_with_auth_proxy_subdomain( self, app, faker, mocker, pgdb_session ): app.config["NO_MFA_ENABLED"] = True mocked_no_mfa_is_active = mocker.patch( "atlas_um.auth.utils.current_user" ) mocked_no_mfa_is_active.claims = [Audience(Audience.Values.no_mfa)] fake_token = faker.pystr() mocked_bearer_token = faker.pystr() mocked_refresh_token = faker.pystr() mocker.patch( "atlas_um.usm.views.login.get_allowed_root_domain" ).return_value = self.FAKE_HOST mocker.patch( "atlas_um.usm.views.login.get_allowed_subdomain" ).return_value = self.FAKE_SUBDOMAIN_HOST mocker.patch( "atlas_um.usm.views.login.get_root_domain" ).return_value = self.FAKE_HOST mocked_user_events = mocker.patch( "atlas_um.analytics.collectors.Collector.log" ) mocked_register_for_resource_group = mocker.patch( "atlas_um.pgdb.DNAAccountActivity.register_for_resource_group" ) mocked_create_refresh_token_service = mocker.patch( "atlas_um.tokens.services.CreateRefreshTokenService.execute" ) mocked_create_refresh_token_service.return_value = Right( mocked_refresh_token ) mocked_create_bearer_token_service = mocker.patch( "atlas_um.tokens.services.CreateBearerTokenService.execute" ) mocked_create_bearer_token_service.return_value = Right( mocked_bearer_token ) mocker.spy(login, "render_template") with app.test_client() as client: dna_account = DNAAccountFactory.build(no_mfa=True) resource_group = ResourceGroupFactory.build( service_url=f"https://{self.FAKE_HOST}", namespace_url="atlasum", ) resource_group_subdomain = ResourceGroupFactory.build( service_url=f"https://{self.FAKE_SUBDOMAIN_HOST}", namespace_url="atlasumsub", ) pgdb_session.add(dna_account) pgdb_session.add(resource_group) pgdb_session.add(resource_group_subdomain) pgdb_session.commit() url = url_for( "usm_login.get_login", **{ consts.DEV_TOKEN_PARAM: fake_token, consts.NO_MFA_ACCOUNT_PARAM: dna_account.email, consts.AUTH_PROXY_SESSION_NAME: self.AUTH_PROXY_PARAM, "next": self.FAKE_SUBDOMAIN_HOST, }, ) resp = client.get(url) assert mocked_register_for_resource_group.call_args_list == [ mocker.call(dna_account, resource_group_subdomain), mocker.call(dna_account, resource_group), ] assert mocked_create_refresh_token_service.call_args_list == [ mocker.call(dna_account, resource_group), mocker.call(dna_account, resource_group_subdomain), ] assert resp.status_code == 200 assert login.render_template.call_args_list == [ mocker.call( "auth_proxy.html", auth_proxy=self.AUTH_PROXY_PARAM, access_token=mocked_bearer_token, refresh_token=mocked_refresh_token, back_to=self.FAKE_SUBDOMAIN_HOST, ) ] assert self.AUTH_PROXY_PARAM in resp.data.decode() bearer_cookie = get_cookie( resp, app.config.get("DNA_BEARER_TOKEN_COOKIE_NAME") ) assert bearer_cookie is not None assert "HttpOnly" in bearer_cookie assert "Secure" in bearer_cookie assert bearer_cookie.get("Domain") == self.FAKE_HOST assert ( int(bearer_cookie["Max-Age"]) == app.config["TOKEN_COOKIES_MAX_AGE"] ) assert ( bearer_cookie.get(app.config.get("DNA_BEARER_TOKEN_COOKIE_NAME")) == mocked_bearer_token ) refresh_cookie = get_cookie( resp, app.config.get("DNA_REFRESH_TOKEN_COOKIE_NAME") ) assert refresh_cookie is not None assert "HttpOnly" in refresh_cookie assert "Secure" in refresh_cookie assert refresh_cookie.get("Domain") == self.FAKE_HOST assert ( int(refresh_cookie["Max-Age"]) == app.config["TOKEN_COOKIES_MAX_AGE"] ) assert ( refresh_cookie.get(app.config.get("DNA_REFRESH_TOKEN_COOKIE_NAME")) == mocked_refresh_token ) assert mocked_user_events.call_args_list == [ mocker.call( SystemEvents.successful_login, domain=self.FAKE_SUBDOMAIN_HOST, user_id=dna_account.sub, ) ] def test_login_no_mfa_enabled_with_referrer( self, app, faker, mocker, pgdb_session ): app.config["NO_MFA_ENABLED"] = True mocked_no_mfa_is_active = mocker.patch( "atlas_um.auth.utils.current_user" ) mocked_no_mfa_is_active.claims = [Audience(Audience.Values.no_mfa)] fake_token = faker.pystr() mocked_bearer_token = faker.pystr() mocked_refresh_token = faker.pystr() mocked_extract = mocker.patch("tldextract.extract").return_value mocked_extract.registered_domain = self.FAKE_HOST mocked_extract.fqdn = self.FAKE_HOST mocked_user_events = mocker.patch( "atlas_um.analytics.collectors.Collector.log" ) test_referrer = faker.url() mocked_register_for_resource_group = mocker.patch( "atlas_um.pgdb.DNAAccountActivity.register_for_resource_group" ) mocked_create_bearer_token_service = mocker.patch( "atlas_um.tokens.services.CreateBearerTokenService.execute" ) mocked_create_bearer_token_service.return_value = Right( mocked_bearer_token ) mocked_create_refresh_token_service = mocker.patch( "atlas_um.tokens.services.CreateRefreshTokenService.execute" ) mocked_create_refresh_token_service.return_value = Right( mocked_refresh_token ) with app.test_client() as client: dna_account = DNAAccountFactory.build(no_mfa=True) resource_group = ResourceGroupFactory.build( service_url=f"https://{self.FAKE_HOST}", namespace_url="atlasum", ) pgdb_session.add(dna_account) pgdb_session.add(resource_group) pgdb_session.commit() url = url_for( "usm_login.get_login", **{ consts.DEV_TOKEN_PARAM: fake_token, consts.NO_MFA_ACCOUNT_PARAM: dna_account.email, }, ) resp = client.get( url, headers={"Referer": test_referrer}, ) assert mocked_register_for_resource_group.call_args_list == [ mocker.call(dna_account, resource_group), ] assert mocked_create_refresh_token_service.call_args_list == [ mocker.call(dna_account, resource_group) ] assert resp.status_code == 302 bearer_cookie = get_cookie( resp, app.config.get("DNA_BEARER_TOKEN_COOKIE_NAME") ) assert bearer_cookie is not None assert "HttpOnly" in bearer_cookie assert "Secure" in bearer_cookie assert bearer_cookie.get("Domain") == self.FAKE_HOST assert ( int(bearer_cookie["Max-Age"]) == app.config["TOKEN_COOKIES_MAX_AGE"] ) assert ( bearer_cookie.get(app.config.get("DNA_BEARER_TOKEN_COOKIE_NAME")) == mocked_bearer_token ) refresh_cookie = get_cookie( resp, app.config.get("DNA_REFRESH_TOKEN_COOKIE_NAME") ) assert refresh_cookie is not None assert "HttpOnly" in refresh_cookie assert "Secure" in refresh_cookie assert refresh_cookie.get("Domain") == self.FAKE_HOST assert resp.location == test_referrer assert ( int(refresh_cookie["Max-Age"]) == app.config["TOKEN_COOKIES_MAX_AGE"] ) assert ( refresh_cookie.get(app.config.get("DNA_REFRESH_TOKEN_COOKIE_NAME")) == mocked_refresh_token ) assert mocked_user_events.call_args_list == [ mocker.call( SystemEvents.successful_login, domain=self.FAKE_HOST, user_id=dna_account.sub, ) ] assert resp.location == test_referrer # login_callback def test_login_callback_no_state_session(self, app, faker): fake_code = faker.pystr() fake_state = faker.pystr() with app.app_context(): req = url_for( "usm_login.get_login_callback", code=fake_code, state=fake_state, ) with app.test_client() as client: resp = client.get(req) assert resp.status_code == 400 def test_login_callback_no_state_param(self, app, faker): fake_code = faker.pystr() fake_state = faker.pystr() with app.app_context(): req = url_for("usm_login.get_login_callback", code=fake_code) with app.test_client() as client: with client.session_transaction() as session: session[consts.STATE_SESSION_NAME] = fake_state resp = client.get(req) assert resp.status_code == 400 def test_login_callback_mismatched_state(self, app, faker): fake_code = faker.pystr() fake_session_state = faker.pystr() fake_state_param = faker.pystr() assert fake_session_state != fake_state_param with app.app_context(): req = url_for( "usm_login.get_login_callback", code=fake_code, state=fake_state_param, ) with app.test_client() as client: with client.session_transaction() as session: session[consts.STATE_SESSION_NAME] = fake_session_state resp = client.get(req) assert resp.status_code == 400 def test_login_callback_no_code_param(self, app, faker): fake_state = faker.pystr() with app.app_context(): req = url_for("usm_login.get_login_callback", state=fake_state) with app.test_client() as client: with client.session_transaction() as session: session[consts.STATE_SESSION_NAME] = fake_state resp = client.get(req) assert resp.status_code == 400 def test_login_callback_bad_code(self, app, faker, mocker): fake_code = faker.pystr() fake_state = faker.pystr() mock_fetch_access_token = mocker.patch( "atlas_um.usm.client.USMClient.fetch_access_token" ) mock_fetch_access_token.return_value = Left("Bad code") with app.app_context(): req = url_for( "usm_login.get_login_callback", code=fake_code, state=fake_state, ) with app.test_client() as client: with client.session_transaction() as session: session[consts.STATE_SESSION_NAME] = fake_state resp = client.get(req) assert resp.status_code == 302 assert resp.location == url_for("common.userinfo") assert mock_fetch_access_token.call_args_list == [ mocker.call(fake_code) ] def test_login_callback_without_referrer( self, app, pgdb_session, faker, mocker ): fake_code = faker.pystr() fake_state = faker.pystr() fake_access_token = { "access_token": "d08trvm4h88c9de4i7gllb7lf6", "id_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJqYXNv" "bi53aGl0dGxlLnNtZUBzb255bXVzaWMuY29tIiwiYXVkIjoiUzNna" "ktuSTBVbTQxYUVkNVVWZG9hMmh0U2xjIiwiYXV0aF90aW1lIjoxNT" "k3OTQ0OTI5LCJpc3MiOiJodHRwczovL3VhdC11c20uc21lYW5hbHl" "0aWNzcG9ydGFsLmNvbS9vYXV0aDIvb3BlbmlkIiwicHJlZmVycmVk" "X3VzZXJuYW1lIjoiamFzb24ud2hpdHRsZS5zbWVAc29ueW11c2ljL" "mNvbSIsImdpdmVuX25hbWUiOiJKYXNvbiIsImV4cCI6MTU5Nzk0Nj" "EyOSwibm9uY2UiOiJaYkRhSkxwUyIsImZhbWlseV9uYW1lIjoiV2h" "pdHRsZSIsImlhdCI6MTU5Nzk0NDkyOSwiZW1haWwiOiJqYXNvbi53" "aGl0dGxlLnNtZUBzb255bXVzaWMuY29tIiwiY2lkIjoiUzNnaktuS" "TBVbTQxYUVkNVVWZG9hMmh0U2xjIn0=.77+977+9A8uFfu+/vRvvv" "70S77+9UO+/vSx777+9aFsJ77+9Q39677+977+977+977+936fvv7" "3vv70=", "token_type": "Bearer", "expires_in": 1200, } dna_account = DNAAccountFactory.build( email="jason.whittle.sme@sonymusic.com" ) resource_group = ResourceGroupFactory.build( service_url=f"https://{self.FAKE_HOST}", namespace_url="atlasum", ) pgdb_session.add(dna_account) pgdb_session.add(resource_group) pgdb_session.commit() mock_fetch_access_token = mocker.patch( "atlas_um.usm.client.USMClient.fetch_access_token" ) mock_fetch_access_token.return_value = Right(fake_access_token) mocked_sync_dna_account_with_usm = mocker.patch( "atlas_um.tasks.sync_dna_account_with_usm.delay" ) mocker.patch( "tldextract.extract" ).return_value.registered_domain = self.FAKE_HOST mocked_user_events = mocker.patch( "atlas_um.analytics.collectors.Collector.log" ) mocked_register_for_resource_group = mocker.patch( "atlas_um.pgdb.DNAAccountActivity.register_for_resource_group" ) mocked_bearer_token = faker.pystr() mocked_refresh_token = faker.pystr() mocked_create_bearer_token_service = mocker.patch( "atlas_um.tokens.services.CreateBearerTokenService.execute" ) mocked_create_bearer_token_service.return_value = Right( mocked_bearer_token ) mocked_create_refresh_token_service = mocker.patch( "atlas_um.tokens.services.CreateRefreshTokenService.execute" ) mocked_create_refresh_token_service.return_value = Right( mocked_refresh_token ) with app.app_context(): req = url_for( "usm_login.get_login_callback", code=fake_code, state=fake_state, ) expected_resource_group_id = resource_group.id expected_dna_account_id = dna_account.id with app.test_client() as client: with client.session_transaction() as session: session[consts.STATE_SESSION_NAME] = fake_state resp = client.get(req) args = mocked_register_for_resource_group.call_args_list[0].args actual_dna_account_id = args[0].id actual_resource_group_id = args[1].id assert actual_dna_account_id == expected_dna_account_id assert actual_resource_group_id == expected_resource_group_id args = mocked_create_refresh_token_service.call_args_list[0].args actual_dna_account_id = args[0].id actual_resource_group_id = args[1].id assert actual_dna_account_id == expected_dna_account_id assert actual_resource_group_id == expected_resource_group_id assert resp.status_code == 302 id_cookie = get_cookie(resp, consts.USM_ID_TOKEN_COOKIE_NAME) assert id_cookie is not None assert "HttpOnly" in id_cookie assert "Secure" in id_cookie assert int(id_cookie["Max-Age"]) == app.config["TOKEN_COOKIES_MAX_AGE"] bearer_cookie = get_cookie( resp, app.config.get("DNA_BEARER_TOKEN_COOKIE_NAME") ) assert bearer_cookie is not None assert "HttpOnly" in bearer_cookie assert "Secure" in bearer_cookie assert bearer_cookie.get("Domain") == self.FAKE_HOST assert ( int(bearer_cookie["Max-Age"]) == app.config["TOKEN_COOKIES_MAX_AGE"] ) assert ( bearer_cookie.get(app.config.get("DNA_BEARER_TOKEN_COOKIE_NAME")) == mocked_bearer_token ) refresh_cookie = get_cookie( resp, app.config.get("DNA_REFRESH_TOKEN_COOKIE_NAME") ) assert refresh_cookie is not None assert "HttpOnly" in refresh_cookie assert "Secure" in refresh_cookie assert refresh_cookie.get("Domain") == self.FAKE_HOST assert ( int(refresh_cookie["Max-Age"]) == app.config["TOKEN_COOKIES_MAX_AGE"] ) assert ( refresh_cookie.get(app.config.get("DNA_REFRESH_TOKEN_COOKIE_NAME")) == mocked_refresh_token ) dna_account = ( pgdb_session.query(DNAAccount) .filter_by(email="jason.whittle.sme@sonymusic.com") .one() ) assert mocked_sync_dna_account_with_usm.call_args_list == [ mocker.call(dna_account.id) ] assert mocked_user_events.call_args_list == [ mocker.call( SystemEvents.successful_login, domain=self.FAKE_HOST, user_id=dna_account.sub, ) ] with app.app_context(): assert resp.location == url_for("common.userinfo") def test_login_callback_with_auth_proxy( self, app, pgdb_session, faker, mocker ): fake_code = faker.pystr() fake_state = faker.pystr() fake_access_token = { "access_token": "d08trvm4h88c9de4i7gllb7lf6", "id_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJqYXNv" "bi53aGl0dGxlLnNtZUBzb255bXVzaWMuY29tIiwiYXVkIjoiUzNna" "ktuSTBVbTQxYUVkNVVWZG9hMmh0U2xjIiwiYXV0aF90aW1lIjoxNT" "k3OTQ0OTI5LCJpc3MiOiJodHRwczovL3VhdC11c20uc21lYW5hbHl" "0aWNzcG9ydGFsLmNvbS9vYXV0aDIvb3BlbmlkIiwicHJlZmVycmVk" "X3VzZXJuYW1lIjoiamFzb24ud2hpdHRsZS5zbWVAc29ueW11c2ljL" "mNvbSIsImdpdmVuX25hbWUiOiJKYXNvbiIsImV4cCI6MTU5Nzk0Nj" "EyOSwibm9uY2UiOiJaYkRhSkxwUyIsImZhbWlseV9uYW1lIjoiV2h" "pdHRsZSIsImlhdCI6MTU5Nzk0NDkyOSwiZW1haWwiOiJqYXNvbi53" "aGl0dGxlLnNtZUBzb255bXVzaWMuY29tIiwiY2lkIjoiUzNnaktuS" "TBVbTQxYUVkNVVWZG9hMmh0U2xjIn0=.77+977+9A8uFfu+/vRvvv" "70S77+9UO+/vSx777+9aFsJ77+9Q39677+977+977+977+936fvv7" "3vv70=", "token_type": "Bearer", "expires_in": 1200, } mock_fetch_access_token = mocker.patch( "atlas_um.usm.client.USMClient.fetch_access_token" ) mock_fetch_access_token.return_value = Right(fake_access_token) mocked_sync_dna_account_with_usm = mocker.patch( "atlas_um.tasks.sync_dna_account_with_usm.delay" ) mocker.patch( "atlas_um.usm.views.login.get_allowed_root_domain" ).return_value = self.FAKE_HOST mocker.patch( "atlas_um.usm.views.login.get_allowed_subdomain" ).return_value = None mocker.patch( "atlas_um.usm.views.login.get_root_domain" ).return_value = self.FAKE_HOST mocked_user_events = mocker.patch( "atlas_um.analytics.collectors.Collector.log" ) mocked_register_for_resource_group = mocker.patch( "atlas_um.pgdb.DNAAccountActivity.register_for_resource_group" ) mocked_bearer_token = faker.pystr() mocked_refresh_token = faker.pystr() mocked_create_bearer_token_service = mocker.patch( "atlas_um.tokens.services.CreateBearerTokenService.execute" ) mocked_create_bearer_token_service.return_value = Right( mocked_bearer_token ) mocked_create_refresh_token_service = mocker.patch( "atlas_um.tokens.services.CreateRefreshTokenService.execute" ) mocked_create_refresh_token_service.return_value = Right( mocked_refresh_token ) mocker.spy(login, "render_template") with app.app_context(): req = url_for( "usm_login.get_login_callback", code=fake_code, state=fake_state, ) with app.test_client() as client: dna_account = DNAAccountFactory.build( email="jason.whittle.sme@sonymusic.com" ) resource_group = ResourceGroupFactory.build( service_url=f"https://{self.FAKE_HOST}", namespace_url="atlasum", ) pgdb_session.add(dna_account) pgdb_session.add(resource_group) pgdb_session.commit() with client.session_transaction() as session: session[consts.STATE_SESSION_NAME] = fake_state session[consts.AUTH_PROXY_SESSION_NAME] = self.AUTH_PROXY_PARAM resp = client.get(req) assert mocked_register_for_resource_group.call_args_list == [ mocker.call(dna_account, resource_group), mocker.call(dna_account, resource_group), ] assert mocked_create_refresh_token_service.call_args_list == [ mocker.call(dna_account, resource_group), mocker.call(dna_account, resource_group), ] assert resp.status_code == 200 assert login.render_template.call_args_list == [ mocker.call( "auth_proxy.html", auth_proxy=self.AUTH_PROXY_PARAM, access_token=mocked_bearer_token, refresh_token=mocked_refresh_token, back_to=url_for("common.userinfo"), ) ] assert self.AUTH_PROXY_PARAM in resp.data.decode() id_cookie = get_cookie(resp, consts.USM_ID_TOKEN_COOKIE_NAME) assert id_cookie is not None assert "HttpOnly" in id_cookie assert "Secure" in id_cookie assert int(id_cookie["Max-Age"]) == app.config["TOKEN_COOKIES_MAX_AGE"] bearer_cookie = get_cookie( resp, app.config.get("DNA_BEARER_TOKEN_COOKIE_NAME") ) assert bearer_cookie is not None assert "HttpOnly" in bearer_cookie assert "Secure" in bearer_cookie assert bearer_cookie.get("Domain") == self.FAKE_HOST assert ( int(bearer_cookie["Max-Age"]) == app.config["TOKEN_COOKIES_MAX_AGE"] ) assert ( bearer_cookie.get(app.config.get("DNA_BEARER_TOKEN_COOKIE_NAME")) == mocked_bearer_token ) refresh_cookie = get_cookie( resp, app.config.get("DNA_REFRESH_TOKEN_COOKIE_NAME") ) assert refresh_cookie is not None assert "HttpOnly" in refresh_cookie assert "Secure" in refresh_cookie assert refresh_cookie.get("Domain") == self.FAKE_HOST assert ( int(refresh_cookie["Max-Age"]) == app.config["TOKEN_COOKIES_MAX_AGE"] ) assert ( refresh_cookie.get(app.config.get("DNA_REFRESH_TOKEN_COOKIE_NAME")) == mocked_refresh_token ) dna_account = ( pgdb_session.query(DNAAccount) .filter_by(email="jason.whittle.sme@sonymusic.com") .one() ) assert mocked_sync_dna_account_with_usm.call_args_list == [ mocker.call(dna_account.id) ] assert mocked_user_events.call_args_list == [ mocker.call( SystemEvents.successful_login, domain=self.FAKE_HOST, user_id=dna_account.sub, ) ] def test_login_callback_with_auth_proxy_subdomain( self, app, pgdb_session, faker, mocker ): fake_code = faker.pystr() fake_state = faker.pystr() fake_access_token = { "access_token": "d08trvm4h88c9de4i7gllb7lf6", "id_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJqYXNv" "bi53aGl0dGxlLnNtZUBzb255bXVzaWMuY29tIiwiYXVkIjoiUzNna" "ktuSTBVbTQxYUVkNVVWZG9hMmh0U2xjIiwiYXV0aF90aW1lIjoxNT" "k3OTQ0OTI5LCJpc3MiOiJodHRwczovL3VhdC11c20uc21lYW5hbHl" "0aWNzcG9ydGFsLmNvbS9vYXV0aDIvb3BlbmlkIiwicHJlZmVycmVk" "X3VzZXJuYW1lIjoiamFzb24ud2hpdHRsZS5zbWVAc29ueW11c2ljL" "mNvbSIsImdpdmVuX25hbWUiOiJKYXNvbiIsImV4cCI6MTU5Nzk0Nj" "EyOSwibm9uY2UiOiJaYkRhSkxwUyIsImZhbWlseV9uYW1lIjoiV2h" "pdHRsZSIsImlhdCI6MTU5Nzk0NDkyOSwiZW1haWwiOiJqYXNvbi53" "aGl0dGxlLnNtZUBzb255bXVzaWMuY29tIiwiY2lkIjoiUzNnaktuS" "TBVbTQxYUVkNVVWZG9hMmh0U2xjIn0=.77+977+9A8uFfu+/vRvvv" "70S77+9UO+/vSx777+9aFsJ77+9Q39677+977+977+977+936fvv7" "3vv70=", "token_type": "Bearer", "expires_in": 1200, } mock_fetch_access_token = mocker.patch( "atlas_um.usm.client.USMClient.fetch_access_token" ) mock_fetch_access_token.return_value = Right(fake_access_token) mocked_sync_dna_account_with_usm = mocker.patch( "atlas_um.tasks.sync_dna_account_with_usm.delay" ) mocker.patch( "atlas_um.usm.views.login.get_allowed_root_domain" ).return_value = self.FAKE_HOST mocker.patch( "atlas_um.usm.views.login.get_allowed_subdomain" ).return_value = self.FAKE_SUBDOMAIN_HOST mocker.patch( "atlas_um.usm.views.login.get_root_domain" ).return_value = self.FAKE_HOST mocked_user_events = mocker.patch( "atlas_um.analytics.collectors.Collector.log" ) mocked_register_for_resource_group = mocker.patch( "atlas_um.pgdb.DNAAccountActivity.register_for_resource_group" ) mocked_bearer_token = faker.pystr() mocked_refresh_token = faker.pystr() mocked_create_bearer_token_service = mocker.patch( "atlas_um.tokens.services.CreateBearerTokenService.execute" ) mocked_create_bearer_token_service.return_value = Right( mocked_bearer_token ) mocked_create_refresh_token_service = mocker.patch( "atlas_um.tokens.services.CreateRefreshTokenService.execute" ) mocked_create_refresh_token_service.return_value = Right( mocked_refresh_token ) mocker.spy(login, "render_template") with app.app_context(): req = url_for( "usm_login.get_login_callback", code=fake_code, state=fake_state, ) with app.test_client() as client: dna_account = DNAAccountFactory.build( email="jason.whittle.sme@sonymusic.com" ) resource_group = ResourceGroupFactory.build( service_url=f"https://{self.FAKE_HOST}", namespace_url="atlasum", ) resource_group_subdomain = ResourceGroupFactory.build( service_url=f"https://{self.FAKE_SUBDOMAIN_HOST}", namespace_url="atlasumsub", ) pgdb_session.add(dna_account) pgdb_session.add(resource_group) pgdb_session.add(resource_group_subdomain) pgdb_session.commit() with client.session_transaction() as session: session[consts.STATE_SESSION_NAME] = fake_state session[consts.AUTH_PROXY_SESSION_NAME] = self.AUTH_PROXY_PARAM session[ consts.REFERRER_SESSION_NAME ] = self.FAKE_SUBDOMAIN_HOST resp = client.get(req) assert mocked_register_for_resource_group.call_args_list == [ mocker.call(dna_account, resource_group_subdomain), mocker.call(dna_account, resource_group), ] assert mocked_create_refresh_token_service.call_args_list == [ mocker.call(dna_account, resource_group), mocker.call(dna_account, resource_group_subdomain), ] assert resp.status_code == 200 assert login.render_template.call_args_list == [ mocker.call( "auth_proxy.html", auth_proxy=self.AUTH_PROXY_PARAM, access_token=mocked_bearer_token, refresh_token=mocked_refresh_token, back_to=self.FAKE_SUBDOMAIN_HOST, ) ] assert self.AUTH_PROXY_PARAM in resp.data.decode() id_cookie = get_cookie(resp, consts.USM_ID_TOKEN_COOKIE_NAME) assert id_cookie is not None assert "HttpOnly" in id_cookie assert "Secure" in id_cookie assert int(id_cookie["Max-Age"]) == app.config["TOKEN_COOKIES_MAX_AGE"] bearer_cookie = get_cookie( resp, app.config.get("DNA_BEARER_TOKEN_COOKIE_NAME") ) assert bearer_cookie is not None assert "HttpOnly" in bearer_cookie assert "Secure" in bearer_cookie assert bearer_cookie.get("Domain") == self.FAKE_HOST assert ( int(bearer_cookie["Max-Age"]) == app.config["TOKEN_COOKIES_MAX_AGE"] ) assert ( bearer_cookie.get(app.config.get("DNA_BEARER_TOKEN_COOKIE_NAME")) == mocked_bearer_token ) refresh_cookie = get_cookie( resp, app.config.get("DNA_REFRESH_TOKEN_COOKIE_NAME") ) assert refresh_cookie is not None assert "HttpOnly" in refresh_cookie assert "Secure" in refresh_cookie assert refresh_cookie.get("Domain") == self.FAKE_HOST assert ( int(refresh_cookie["Max-Age"]) == app.config["TOKEN_COOKIES_MAX_AGE"] ) assert ( refresh_cookie.get(app.config.get("DNA_REFRESH_TOKEN_COOKIE_NAME")) == mocked_refresh_token ) dna_account = ( pgdb_session.query(DNAAccount) .filter_by(email="jason.whittle.sme@sonymusic.com") .one() ) assert mocked_sync_dna_account_with_usm.call_args_list == [ mocker.call(dna_account.id) ] assert mocked_user_events.call_args_list == [ mocker.call( SystemEvents.successful_login, domain=self.FAKE_SUBDOMAIN_HOST, user_id=dna_account.sub, ) ] def test_login_callback_with_referrer( self, app, pgdb_session, faker, mocker ): fake_code = faker.pystr() fake_state = faker.pystr() fake_access_token = { "access_token": "d08trvm4h88c9de4i7gllb7lf6", "id_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJqYXNv" "bi53aGl0dGxlLnNtZUBzb255bXVzaWMuY29tIiwiYXVkIjoiUzNna" "ktuSTBVbTQxYUVkNVVWZG9hMmh0U2xjIiwiYXV0aF90aW1lIjoxNT" "k3OTQ0OTI5LCJpc3MiOiJodHRwczovL3VhdC11c20uc21lYW5hbHl" "0aWNzcG9ydGFsLmNvbS9vYXV0aDIvb3BlbmlkIiwicHJlZmVycmVk" "X3VzZXJuYW1lIjoiamFzb24ud2hpdHRsZS5zbWVAc29ueW11c2ljL" "mNvbSIsImdpdmVuX25hbWUiOiJKYXNvbiIsImV4cCI6MTU5Nzk0Nj" "EyOSwibm9uY2UiOiJaYkRhSkxwUyIsImZhbWlseV9uYW1lIjoiV2h" "pdHRsZSIsImlhdCI6MTU5Nzk0NDkyOSwiZW1haWwiOiJqYXNvbi53" "aGl0dGxlLnNtZUBzb255bXVzaWMuY29tIiwiY2lkIjoiUzNnaktuS" "TBVbTQxYUVkNVVWZG9hMmh0U2xjIn0=.77+977+9A8uFfu+/vRvvv" "70S77+9UO+/vSx777+9aFsJ77+9Q39677+977+977+977+936fvv7" "3vv70=", "token_type": "Bearer", "expires_in": 1200, } dna_account = DNAAccountFactory.build( email="jason.whittle.sme@sonymusic.com" ) resource_group = ResourceGroupFactory.build( service_url=f"https://{self.FAKE_HOST}", namespace_url="atlasum", ) pgdb_session.add(dna_account) pgdb_session.add(resource_group) pgdb_session.commit() mock_fetch_access_token = mocker.patch( "atlas_um.usm.client.USMClient.fetch_access_token" ) mock_fetch_access_token.return_value = Right(fake_access_token) mocked_sync_dna_account_with_usm = mocker.patch( "atlas_um.tasks.sync_dna_account_with_usm.delay" ) req = url_for( "usm_login.get_login_callback", code=fake_code, state=fake_state ) mocked_extract = mocker.patch("tldextract.extract").return_value mocked_extract.registered_domain = self.FAKE_HOST mocked_extract.fqdn = self.FAKE_HOST mocked_user_events = mocker.patch( "atlas_um.analytics.collectors.Collector.log" ) mocked_register_for_resource_group = mocker.patch( "atlas_um.pgdb.DNAAccountActivity.register_for_resource_group" ) mocked_bearer_token = faker.pystr() mocked_refresh_token = faker.pystr() mocked_create_bearer_token_service = mocker.patch( "atlas_um.tokens.services.CreateBearerTokenService.execute" ) mocked_create_bearer_token_service.return_value = Right( mocked_bearer_token ) mocked_create_refresh_token_service = mocker.patch( "atlas_um.tokens.services.CreateRefreshTokenService.execute" ) mocked_create_refresh_token_service.return_value = Right( mocked_refresh_token ) with app.test_client() as client: with client.session_transaction() as session: session[consts.REFERRER_SESSION_NAME] = "http://other.test/abc" session[consts.STATE_SESSION_NAME] = fake_state resp = client.get(req) assert mocked_register_for_resource_group.call_args_list == [ mocker.call(dna_account, resource_group), ] assert mocked_create_refresh_token_service.call_args_list == [ mocker.call(dna_account, resource_group) ] assert resp.status_code == 302 id_cookie = get_cookie(resp, consts.USM_ID_TOKEN_COOKIE_NAME) assert id_cookie is not None assert "HttpOnly" in id_cookie assert "Secure" in id_cookie assert id_cookie.get("Domain") == self.FAKE_HOST assert int(id_cookie["Max-Age"]) == app.config["TOKEN_COOKIES_MAX_AGE"] bearer_cookie = get_cookie( resp, app.config.get("DNA_BEARER_TOKEN_COOKIE_NAME") ) assert bearer_cookie is not None assert "HttpOnly" in bearer_cookie assert "Secure" in bearer_cookie assert bearer_cookie.get("Domain") == self.FAKE_HOST assert ( int(bearer_cookie["Max-Age"]) == app.config["TOKEN_COOKIES_MAX_AGE"] ) assert ( bearer_cookie.get(app.config.get("DNA_BEARER_TOKEN_COOKIE_NAME")) == mocked_bearer_token ) refresh_cookie = get_cookie( resp, app.config.get("DNA_REFRESH_TOKEN_COOKIE_NAME") ) assert refresh_cookie is not None assert "HttpOnly" in refresh_cookie assert "Secure" in refresh_cookie assert refresh_cookie.get("Domain") == self.FAKE_HOST assert ( int(refresh_cookie["Max-Age"]) == app.config["TOKEN_COOKIES_MAX_AGE"] ) assert ( refresh_cookie.get(app.config.get("DNA_REFRESH_TOKEN_COOKIE_NAME")) == mocked_refresh_token ) dna_account = ( pgdb_session.query(DNAAccount) .filter_by(email="jason.whittle.sme@sonymusic.com") .one() ) assert mocked_sync_dna_account_with_usm.call_args_list == [ mocker.call(dna_account.id) ] assert resp.location == "http://other.test/abc" assert mocked_user_events.call_args_list == [ mocker.call( SystemEvents.successful_login, domain=self.FAKE_HOST, user_id=dna_account.sub, ) ] def test_login_callback_with_subdomain_product_referrer( self, app, pgdb_session, faker, mocker ): fake_code = faker.pystr() fake_state = faker.pystr() fake_access_token = { "access_token": "d08trvm4h88c9de4i7gllb7lf6", "id_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJqYXNv" "bi53aGl0dGxlLnNtZUBzb255bXVzaWMuY29tIiwiYXVkIjoiUzNna" "ktuSTBVbTQxYUVkNVVWZG9hMmh0U2xjIiwiYXV0aF90aW1lIjoxNT" "k3OTQ0OTI5LCJpc3MiOiJodHRwczovL3VhdC11c20uc21lYW5hbHl" "0aWNzcG9ydGFsLmNvbS9vYXV0aDIvb3BlbmlkIiwicHJlZmVycmVk" "X3VzZXJuYW1lIjoiamFzb24ud2hpdHRsZS5zbWVAc29ueW11c2ljL" "mNvbSIsImdpdmVuX25hbWUiOiJKYXNvbiIsImV4cCI6MTU5Nzk0Nj" "EyOSwibm9uY2UiOiJaYkRhSkxwUyIsImZhbWlseV9uYW1lIjoiV2h" "pdHRsZSIsImlhdCI6MTU5Nzk0NDkyOSwiZW1haWwiOiJqYXNvbi53" "aGl0dGxlLnNtZUBzb255bXVzaWMuY29tIiwiY2lkIjoiUzNnaktuS" "TBVbTQxYUVkNVVWZG9hMmh0U2xjIn0=.77+977+9A8uFfu+/vRvvv" "70S77+9UO+/vSx777+9aFsJ77+9Q39677+977+977+977+936fvv7" "3vv70=", "token_type": "Bearer", "expires_in": 1200, } dna_account = DNAAccountFactory.build( email="jason.whittle.sme@sonymusic.com" ) resource_group = ResourceGroupFactory.build( service_url=f"https://{self.FAKE_HOST}", namespace_url="atlasum", ) product_domain = f"product_subdomain.{self.FAKE_HOST}" product_url = f"https://{product_domain}" product_resource_group = ResourceGroupFactory.build( service_url=product_url, namespace_url="product", ) pgdb_session.add(dna_account) pgdb_session.add(resource_group) pgdb_session.add(product_resource_group) pgdb_session.commit() mock_fetch_access_token = mocker.patch( "atlas_um.usm.client.USMClient.fetch_access_token" ) mock_fetch_access_token.return_value = Right(fake_access_token) mocked_sync_dna_account_with_usm = mocker.patch( "atlas_um.tasks.sync_dna_account_with_usm.delay" ) req = url_for( "usm_login.get_login_callback", code=fake_code, state=fake_state ) mocked_extract = mocker.patch("tldextract.extract").return_value mocked_extract.registered_domain = self.FAKE_HOST mocked_extract.fqdn = product_domain mocked_user_events = mocker.patch( "atlas_um.analytics.collectors.Collector.log" ) mocked_register_for_resource_group = mocker.patch( "atlas_um.pgdb.DNAAccountActivity.register_for_resource_group" ) mocked_bearer_token = faker.pystr() mocked_refresh_token = faker.pystr() mocked_create_bearer_token_service = mocker.patch( "atlas_um.tokens.services.CreateBearerTokenService.execute" ) mocked_create_bearer_token_service.return_value = Right( mocked_bearer_token ) mocked_create_refresh_token_service = mocker.patch( "atlas_um.tokens.services.CreateRefreshTokenService.execute" ) mocked_create_refresh_token_service.return_value = Right( mocked_refresh_token ) with app.test_client() as client: with client.session_transaction() as session: session[consts.REFERRER_SESSION_NAME] = product_url session[consts.STATE_SESSION_NAME] = fake_state resp = client.get(req) assert mocked_register_for_resource_group.call_args_list == [ mocker.call(dna_account, product_resource_group), mocker.call(dna_account, resource_group), ] assert mocked_create_refresh_token_service.call_args_list == [ mocker.call(dna_account, resource_group) ] assert resp.status_code == 302 id_cookie = get_cookie(resp, consts.USM_ID_TOKEN_COOKIE_NAME) assert id_cookie is not None assert "HttpOnly" in id_cookie assert "Secure" in id_cookie assert id_cookie.get("Domain") == self.FAKE_HOST assert int(id_cookie["Max-Age"]) == app.config["TOKEN_COOKIES_MAX_AGE"] bearer_cookie = get_cookie( resp, app.config.get("DNA_BEARER_TOKEN_COOKIE_NAME") ) assert bearer_cookie is not None assert "HttpOnly" in bearer_cookie assert "Secure" in bearer_cookie assert bearer_cookie.get("Domain") == self.FAKE_HOST assert ( int(bearer_cookie["Max-Age"]) == app.config["TOKEN_COOKIES_MAX_AGE"] ) assert ( bearer_cookie.get(app.config.get("DNA_BEARER_TOKEN_COOKIE_NAME")) == mocked_bearer_token ) refresh_cookie = get_cookie( resp, app.config.get("DNA_REFRESH_TOKEN_COOKIE_NAME") ) assert refresh_cookie is not None assert "HttpOnly" in refresh_cookie assert "Secure" in refresh_cookie assert refresh_cookie.get("Domain") == self.FAKE_HOST assert ( int(refresh_cookie["Max-Age"]) == app.config["TOKEN_COOKIES_MAX_AGE"] ) assert ( refresh_cookie.get(app.config.get("DNA_REFRESH_TOKEN_COOKIE_NAME")) == mocked_refresh_token ) dna_account = ( pgdb_session.query(DNAAccount) .filter_by(email="jason.whittle.sme@sonymusic.com") .one() ) assert mocked_sync_dna_account_with_usm.call_args_list == [ mocker.call(dna_account.id) ] assert resp.location == product_url assert mocked_user_events.call_args_list == [ mocker.call( SystemEvents.successful_login, domain=product_domain, user_id=dna_account.sub, ) ] def test_login_callback_suspended_account( self, app, pgdb_session, faker, mocker ): fake_code = faker.pystr() fake_state = faker.pystr() fake_access_token = { "access_token": "d08trvm4h88c9de4i7gllb7lf6", "id_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9" ".eyJzdWIiOiJqYXNvbi53aGl0dGxlLnNtZUBzb255bXVzaWMuY29tI" "iwiYXVkIjoiUzNnaktuSTBVbTQxYUVkNVVWZG9hMmh0U2xjIiwiYXV" "0aF90aW1lIjoxNTk3OTQ0OTI5LCJpc3MiOiJodHRwczovL3VhdC11c" "20uc21lYW5hbHl0aWNzcG9ydGFsLmNvbS9vYXV0aDIvb3BlbmlkIiw" "icHJlZmVycmVkX3VzZXJuYW1lIjoiamFzb24ud2hpdHRsZS5zbWVAc" "29ueW11c2ljLmNvbSIsImdpdmVuX25hbWUiOiJKYXNvbiIsImV4cCI" "6MTU5Nzk0NjEyOSwibm9uY2UiOiJaYkRhSkxwUyIsImZhbWlseV9uY" "W1lIjoiV2hpdHRsZSIsImlhdCI6MTU5Nzk0NDkyOSwiZW1haWwiOiJ" "qYXNvbi53aGl0dGxlLnNtZUBzb255bXVzaWMuY29tIiwiY2lkIjoiU" "zNnaktuSTBVbTQxYUVkNVVWZG9hMmh0U2xjIn0=.77+977+9A8uFfu" "+/vRvvv70S77+9UO+/vSx777+9aFsJ77+9Q39677+977+977+977+9" "36fvv73vv70=", "token_type": "Bearer", "expires_in": 1200, } mocked_account = mocker.patch( "atlas_um.pgdb.dna_account.DNAAccount.selsert_from_usm_claimset" ).return_value mocked_account.status = DNAAccountStatuses.SUSPENDED mocked_account.expiration_date = faker.past_date() mock_fetch_access_token = mocker.patch( "atlas_um.usm.client.USMClient.fetch_access_token" ) mock_fetch_access_token.return_value = Right(fake_access_token) req = url_for( "usm_login.get_login_callback", code=fake_code, state=fake_state ) with app.test_client() as client: with client.session_transaction() as session: session[consts.STATE_SESSION_NAME] = fake_state resp = client.get(req) assert resp.status_code == 401 def test_login_callback_non_existing_account( self, app, pgdb_session, faker, mocker ): fake_code = faker.pystr() fake_state = faker.pystr() fake_access_token = { "access_token": "d08trvm4h88c9de4i7gllb7lf6", "id_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9" ".eyJzdWIiOiJqYXNvbi53aGl0dGxlLnNtZUBzb255bXVzaWMuY29tI" "iwiYXVkIjoiUzNnaktuSTBVbTQxYUVkNVVWZG9hMmh0U2xjIiwiYXV" "0aF90aW1lIjoxNTk3OTQ0OTI5LCJpc3MiOiJodHRwczovL3VhdC11c" "20uc21lYW5hbHl0aWNzcG9ydGFsLmNvbS9vYXV0aDIvb3BlbmlkIiw" "icHJlZmVycmVkX3VzZXJuYW1lIjoiamFzb24ud2hpdHRsZS5zbWVAc" "29ueW11c2ljLmNvbSIsImdpdmVuX25hbWUiOiJKYXNvbiIsImV4cCI" "6MTU5Nzk0NjEyOSwibm9uY2UiOiJaYkRhSkxwUyIsImZhbWlseV9uY" "W1lIjoiV2hpdHRsZSIsImlhdCI6MTU5Nzk0NDkyOSwiZW1haWwiOiJ" "qYXNvbi53aGl0dGxlLnNtZUBzb255bXVzaWMuY29tIiwiY2lkIjoiU" "zNnaktuSTBVbTQxYUVkNVVWZG9hMmh0U2xjIn0=.77+977+9A8uFfu" "+/vRvvv70S77+9UO+/vSx777+9aFsJ77+9Q39677+977+977+977+9" "36fvv73vv70=", "token_type": "Bearer", "expires_in": 1200, } mock_fetch_access_token = mocker.patch( "atlas_um.usm.client.USMClient.fetch_access_token" ) mock_fetch_access_token.return_value = Right(fake_access_token) req = url_for( "usm_login.get_login_callback", code=fake_code, state=fake_state ) with app.test_client() as client: with client.session_transaction() as session: session[consts.STATE_SESSION_NAME] = fake_state resp = client.get(req) assert DNAAccount.query.count() == 0 assert resp.status_code == 401