"""Test Access.""" from unittest.mock import MagicMock, patch import pytest from grass import access, config from grass.consts import headers from grass.consts.headers import JWT_USER_METADATA from grass.logic import api_token, endpoint_rules, session, user from grass.models.profile_user import ProfileUser from grass.utils import response from tests.unit.fixtures import fixture_request def test_access_confirm_denied(): """Test an access is denied. When all the access checks have been denied, the method returns false. """ assert not access.confirm(fixture_request.RequestHandler()) def test_access_confirm_options(): """Test allow_session_access_for_routes is allowed for OPTIONS request.""" request_handler = fixture_request.RequestHandler(request_method='OPTIONS') assert access.confirm(request_handler) def test_access_confirm_with_unsupported_values(): """Test the access method by passing unsupported values. It should raise a custom exception (ExceptionAccessList) because none of those values are supported. """ request_handler = fixture_request.RequestHandler() for unsupported_value in [(False,), 'Unsupported', 10]: try: access.confirm(request_handler, unsupported_value) assert False except access.ExceptionAccessList: assert True def test_client_access_denied(): """Test client access. If information is missing or if the token is not valid, the method should fail. """ request_handler = fixture_request.RequestHandler() wrapper = access.allow_client() assert not wrapper(request_handler) request_handler.arguments.update(client='fake', token='fake', user='alw:fake') assert not wrapper(request_handler) request_handler.arguments.update(client='fake', token='fake', user=None) assert not wrapper(request_handler) request_handler.arguments.update(client='fake', token=None, user='alw:fake') assert not wrapper(request_handler) request_handler.arguments.update(client=None, token='fake', user='alw:fake') assert not wrapper(request_handler) def test_client_access_with_list(monkeypatch): """Test client access.""" is_token_valid = MagicMock(return_value=True) monkeypatch.setattr(api_token, 'is_token_valid', is_token_valid) client_id = 'fake' request_handler = fixture_request.RequestHandler( arguments=dict(client=client_id, token='fake', user='alw:fake') ) wrapper = access.allow_client(client_id) assert wrapper(request_handler) assert is_token_valid.called def test_client_access_with_list_denied(monkeypatch): """Test client access.""" is_token_valid = MagicMock(return_value=True) monkeypatch.setattr(api_token, 'is_token_valid', is_token_valid) client_id = 'fake' request_handler = fixture_request.RequestHandler( arguments=dict(client=client_id, token='fake', user='alw:fake') ) wrapper = access.allow_client('anotherclient') assert not wrapper(request_handler) assert not is_token_valid.called @pytest.fixture def fixture_user_token_response(): """Get a tuple representation of a user token response.""" return ( { 'user_id': 'alw:20982', 'client_id': '5656789871', 'roles': [1], 'roles_by_name': ['admin', 'catalog'], }, 200, ) @pytest.fixture def fixture_test_rules_user_token_response(): """Get a tuple representation of a user token response.""" return ({'user_id': 'alw:16888', 'client_id': '5656789871', 'roles': [1]}, 200) @pytest.fixture def fixture_oa_user_token_response(): """Get a tuple representation of OA user token response.""" return ( { 'user_id': 'oa:111', 'client_id': '5656789872', 'roles': [1, 2, 3], 'roles_by_name': ['admin', 'pricing'], }, 200, ) def test_role_access(): """Test access role. TODO(mo): test to implement after the method has been implemented. """ request_handler = fixture_request.RequestHandler() assert not access.allow_role()(request_handler) def test_allow_session_access_for_routes_with_token_in_param(monkeypatch): """Test allow_session_access_for_routes fails for token in query param.""" monkeypatch.setattr(session, 'get_token', MagicMock(return_value='Does not matter')) request_handler = fixture_request.RequestHandler(arguments=dict(session='random')) monkeypatch.setattr(endpoint_rules, 'EndpointRulesValidator', MagicMock()) wrapper = access.allow_session_access_for_routes('test.yml') assert not wrapper(request_handler) def test_allow_session_access_for_routes_invalid_token(monkeypatch): """Test allow_session_access_for_routes with invalid token.""" monkeypatch.setattr( session, 'get_token', MagicMock(return_value=response.create_not_found_response()), ) request_handler = fixture_request.RequestHandler( headers=dict(session='other-token') ) monkeypatch.setattr(endpoint_rules, 'EndpointRulesValidator', MagicMock()) wrapper = access.allow_session_access_for_routes('test.yml') assert not wrapper(request_handler) def test_allow_session_access_for_routes_oa_user( monkeypatch, fixture_oa_user_token_response ): """Test allow_session_access_for_routes with OA user.""" monkeypatch.setattr( session, 'get_token', MagicMock(return_value=fixture_oa_user_token_response) ) request_handler = fixture_request.RequestHandler( headers=dict(session='other-token') ) mock_get_user_by_id = MagicMock() monkeypatch.setattr(user, 'get_user_by_id', mock_get_user_by_id) monkeypatch.setattr(endpoint_rules, 'EndpointRulesValidator', MagicMock()) wrapper = access.allow_session_access_for_routes('test.yml') assert wrapper(request_handler) @pytest.mark.parametrize('expected_result', [True, False]) def test_allow_session_access_for_routes_oa_resource( monkeypatch, fixture_oa_user_token_response, expected_result ): """Test allow_session_access_for_routes for resource access by OA user.""" monkeypatch.setattr( session, 'get_token', MagicMock(return_value=fixture_oa_user_token_response) ) monkeypatch.setattr( user, 'is_allowed_for_any_resources', MagicMock(return_value=expected_result) ) request_handler = fixture_request.RequestHandler( headers=dict(session='other-token') ) mock_get_user_by_id = MagicMock() monkeypatch.setattr(user, 'get_user_by_id', mock_get_user_by_id) rules_class = MagicMock() rules_class.has_resource_access.return_value = expected_result monkeypatch.setattr( endpoint_rules, 'EndpointRulesValidator', MagicMock(return_value=rules_class) ) wrapper = access.allow_session_access_for_routes('test.yml', ['test']) assert wrapper(request_handler) == expected_result if expected_result: # test if alw user with valid route, that roles got set mock_user = mock_get_user_by_id() mock_user.message.set_roles.assert_called_with(['admin', 'pricing']) @pytest.mark.parametrize('expected_result', [True, False]) def test_allow_session_access_for_routes( monkeypatch, fixture_user_token_response, expected_result ): """Test allow_session_access_for_routes .""" monkeypatch.setattr( session, 'get_token', MagicMock(return_value=fixture_user_token_response) ) request_handler = fixture_request.RequestHandler( headers=dict(session='other-token') ) mock_get_user_by_id = MagicMock() monkeypatch.setattr(user, 'get_user_by_id', mock_get_user_by_id) rules_class = MagicMock() rules_class.has_access.return_value = expected_result monkeypatch.setattr( endpoint_rules, 'EndpointRulesValidator', MagicMock(return_value=rules_class) ) wrapper = access.allow_session_access_for_routes('test.yml', ['test']) result = wrapper(request_handler) assert result == expected_result if expected_result: # test if alw user with valid route, that roles got set mock_user = mock_get_user_by_id() mock_user.message.set_roles.assert_called_with(['admin', 'catalog']) @pytest.mark.parametrize( 'env,rules_environments,expected_result', [('prod', ['prod'], True), ('qa', ['qa'], True), ('qa', ['prod'], False)], ) def test_allow_session_access_for_routes_with_environment( monkeypatch, fixture_user_token_response, env, rules_environments, expected_result ): """Test allow_session_access_for_routes with environment.""" monkeypatch.setattr( session, 'get_token', MagicMock(return_value=fixture_user_token_response) ) request_handler = fixture_request.RequestHandler( headers=dict(session='other-token') ) mock_get_user_by_id = MagicMock() monkeypatch.setattr(user, 'get_user_by_id', mock_get_user_by_id) rules_class = MagicMock() rules_class.has_access.return_value = True monkeypatch.setattr( endpoint_rules, 'EndpointRulesValidator', MagicMock(return_value=rules_class) ) monkeypatch.setattr(config, 'environment', env) wrapper = access.allow_session_access_for_routes( 'test.yml', environments=rules_environments ) wrapper(request_handler) if expected_result: request_handler.current_user.set_roles.assert_called_with(['admin', 'catalog']) rules_class.has_access.assert_called() def test_allow_session_access_no_rules(monkeypatch, fixture_user_token_response): """Test allow_session_access_for_routes with no rules.""" monkeypatch.setattr( session, 'get_token', MagicMock(return_value=fixture_user_token_response) ) request_handler = fixture_request.RequestHandler( headers=dict(session='other-token') ) mock_get_user_by_id = MagicMock() monkeypatch.setattr(user, 'get_user_by_id', mock_get_user_by_id) wrapper = access.allow_session_access_for_routes(None) result = wrapper(request_handler) request_handler.current_user.set_roles.assert_called_with(['admin', 'catalog']) assert result @pytest.mark.parametrize( ('payload', 'error', 'expected', 'expected_id'), [ ({'dummy'}, 'Unable to parse authentication token', False, False), ({'no': 'identity'}, None, False, False), # valid JWT with only sub so send auth0 id as orchard_identity_id ({'sub': 'auth0|123'}, None, True, '123'), ( {'sub': 'auth0|5df219a7e990080cff401ace'}, None, True, '5df219a7e990080cff401ace', ), ( {'sub': 'auth0|5df219a7e990080cff401ace', JWT_USER_METADATA: {}}, None, True, '5df219a7e990080cff401ace', ), # valid JWT with orchardIdentityId so send that id as orchard_identity_id ( { 'sub': 'auth0|5df219a7e990080cff401ace', JWT_USER_METADATA: {'orchardIdentityId': '36-char-uuid'}, }, None, True, '36-char-uuid', ), ( { 'sub': 'auth0|123', JWT_USER_METADATA: {'orchardIdentityId': '36-char-uuid'}, }, None, True, '36-char-uuid', ), ( { 'sub': 'google-apps|rshield@theorchard.com', JWT_USER_METADATA: {'orchardIdentityId': '36-char-uuid'}, }, None, True, '36-char-uuid', ), ( { 'sub': 'waad|rshield@theorchard.com', JWT_USER_METADATA: {'orchardIdentityId': '36-char-orch-uuid'}, }, None, True, '36-char-orch-uuid', ), ], ) @patch('grass.access.model_session') @patch('grass.access.auth') @patch('grass.access.config') def test_allow_jwt_token( mock_config: MagicMock, mock_auth: MagicMock, mock_model_session: MagicMock, payload, error, expected, expected_id, ): """Test allow_jwt_token with only JWT token.""" mock_config.M2M_API_AUDIENCE = 'no' mock_config.AUTH0_API_AUDIENCE = 'yes' validate_result = (None, 200, payload) if error: validate_result = (error, 401, {}) mock_auth.validate_auth0_token.return_value = validate_result mock_model_session.is_jwt_cached.return_value = False request_headers = {headers.AUTHORIZATION: 'Bearer token1'} request_handler = fixture_request.RequestHandler() request_handler.request.headers = request_headers actual = access.allow_jwt_token()(request_handler) assert actual == expected mock_auth.validate_auth0_token.assert_called_with('token1', ['yes']) if expected: assert isinstance(request_handler.current_user, ProfileUser) assert request_handler.current_user.orchard_identity_id == expected_id else: assert not request_handler.current_user @patch('grass.access.auth') def test_allow_jwt_token_no_token(mock_auth): """Test allow_jwt_token with invalid Auth header.""" jwt_payload = {'sub': 'auth0|123'} mock_auth.validate_auth0_token.return_value = (None, 200, jwt_payload) request_headers = {headers.AUTHORIZATION: 'dummy'} request_handler = fixture_request.RequestHandler() request_handler.request.headers = request_headers actual = access.allow_jwt_token()(request_handler) assert not actual mock_auth.validate_auth0_token.assert_not_called() assert not request_handler.current_user @pytest.mark.parametrize( ('payload', 'request_headers', 'expected', 'active_profile_roles'), [ # no profiles in jwt and no profile headers so => allow. ( {'sub': 'auth0|123', headers.JWT_PROFILES: []}, {headers.AUTHORIZATION: 'Bearer token1'}, True, None, ), # headers don't match profile headers sent => block. ( {'sub': 'auth0|123', headers.JWT_PROFILES: []}, { headers.AUTHORIZATION: 'Bearer token1', headers.ORCHARD_PROFILE_TYPE: 'testprofile', headers.ORCHARD_PROFILE_ID: '1234', }, False, None, ), ( { 'sub': 'auth0|123', headers.JWT_PROFILES: [ { 'profile_type': 'otherprofile', 'profile_id': '1234', 'roles': ['analytics', 'accounting'], }, ], }, { headers.AUTHORIZATION: 'Bearer token1', headers.ORCHARD_PROFILE_TYPE: 'testprofile', headers.ORCHARD_PROFILE_ID: '1234', }, False, None, ), ( { 'sub': 'auth0|123', headers.JWT_PROFILES: [ { 'profile_type': 'otherprofile', 'profile_id': '1234', 'roles': ['analytics', 'accounting'], }, { 'profile_type': 'dummy', 'profile_id': '456', 'roles': ['catalog'], }, ], }, { headers.AUTHORIZATION: 'Bearer token1', headers.ORCHARD_PROFILE_TYPE: 'testprofile', headers.ORCHARD_PROFILE_ID: '1234', }, False, None, ), # headers match one of the JWT profile => allow. ( { 'sub': 'auth0|123', headers.JWT_PROFILES: [ { 'profile_type': 'testprofile', 'profile_id': '1234', 'roles': ['catalog'], }, { 'profile_type': 'dummy', 'profile_id': '456', 'roles': ['analytics'], }, ], }, { headers.AUTHORIZATION: 'Bearer token1', headers.ORCHARD_PROFILE_TYPE: 'testprofile', headers.ORCHARD_PROFILE_ID: '1234', }, True, ['catalog'], ), ( { 'sub': 'auth0|123', headers.JWT_PROFILES: [ { 'profile_type': 'testprofile', 'profile_id': '1234', 'roles': ['analytics', 'accounting'], }, ], }, { headers.AUTHORIZATION: 'Bearer token1', headers.ORCHARD_PROFILE_TYPE: 'testprofile', headers.ORCHARD_PROFILE_ID: '1234', }, True, ['analytics', 'accounting'], ), # Profile UUID with existing profile headers. ( { 'sub': 'auth0|123', headers.JWT_PROFILES: [ { 'uuid': 'f91239bb-53b5-45f5-97a9-b0f4f4edb122', 'profile_type': 'testprofile', 'profile_id': '1234', 'roles': ['analytics', 'accounting'], }, ], }, { headers.AUTHORIZATION: 'Bearer token1', headers.ORCHARD_PROFILE_UUID: 'f91239bb-53b5-45f5-97a9-b0f4f4edb122', # noqa headers.ORCHARD_PROFILE_TYPE: 'testprofile', headers.ORCHARD_PROFILE_ID: '1234', }, True, ['analytics', 'accounting'], ), ( { 'sub': 'auth0|123', headers.JWT_PROFILES: [ { 'uuid': 'f91239bb-53b5-45f5-97a9-b0f4f4edb122', 'profile_type': 'testprofile', 'profile_id': '1234', 'roles': ['analytics', 'accounting'], }, ], }, { headers.AUTHORIZATION: 'Bearer token1', headers.ORCHARD_PROFILE_UUID: 'dummy', headers.ORCHARD_PROFILE_TYPE: 'testprofile', headers.ORCHARD_PROFILE_ID: '1234', }, False, None, ), # Only Profile UUID headers. ( { 'sub': 'auth0|123', headers.JWT_PROFILES: [ { 'uuid': 'f91239bb-53b5-45f5-97a9-b0f4f4edb122', 'profile_type': 'testprofile', 'profile_id': '1234', 'roles': ['analytics', 'accounting'], }, ], }, { headers.AUTHORIZATION: 'Bearer token1', headers.ORCHARD_PROFILE_UUID: 'f91239bb-53b5-45f5-97a9-b0f4f4edb122', # noqa }, True, ['analytics', 'accounting'], ), ( { 'sub': 'auth0|123', headers.JWT_PROFILES: [ { 'uuid': 'f91239bb-53b5-45f5-97a9-b0f4f4edb122', 'profile_type': 'testprofile', 'profile_id': '1234', 'roles': ['analytics', 'accounting'], }, ], }, { headers.AUTHORIZATION: 'Bearer token1', headers.ORCHARD_PROFILE_UUID: 'dummy', }, False, None, ), ], ) @patch('grass.access.auth') @patch('grass.access.config') def test_allow_jwt_token_profile_header( mock_config: MagicMock, mock_auth: MagicMock, payload, request_headers, expected, active_profile_roles, ): """Test allow_jwt_token with JWT token and profile headers.""" mock_config.M2M_API_AUDIENCE = 'no' mock_config.AUTH0_API_AUDIENCE = 'yes' mock_auth.validate_auth0_token.return_value = (None, 200, payload) request_handler = fixture_request.RequestHandler() request_handler.request.headers = request_headers actual = access.allow_jwt_token()(request_handler) assert actual == expected mock_auth.validate_auth0_token.assert_called_with('token1', ['yes']) if expected: assert isinstance(request_handler.current_user, ProfileUser) assert request_handler.current_user.has_active_profile == bool( payload.get(headers.JWT_PROFILES) ) identity = payload.get('sub').replace('auth0|', '') assert request_handler.current_user.orchard_identity_id == identity assert request_handler.current_user.roles == active_profile_roles else: assert not request_handler.current_user def test_allow_auth0_m2m_token_with_no_token(): """Test no token case.""" request_headers = {headers.GRASS_HEADER_USER_ID: 123} request_handler = fixture_request.RequestHandler() request_handler.request.headers = request_headers resp = access.allow_auth0_m2m_token()(request_handler) assert not resp @patch('grass.access.model_session') @patch('grass.access.auth') @patch('grass.access.config') def test_allow_auth0_m2m_token_with_auth0_token( mock_config: MagicMock, mock_auth: MagicMock, mock_model_session: MagicMock, ) -> None: """Test token with vend_contact_id is rejected.""" mock_config.M2M_API_AUDIENCE = 'no' mock_config.AUTH0_API_AUDIENCE = 'yes' jwt_payload = { 'https://grass.theorchard.com/vend_contact_id': 123, 'sub': 'auth0|123', } mock_auth.validate_auth0_token.return_value = (None, 200, jwt_payload) mock_model_session.is_jwt_cached.return_value = False request_headers = { headers.AUTHORIZATION: 'Bearer token', headers.GRASS_HEADER_USER_ID: 123, } request_handler = fixture_request.RequestHandler() request_handler.request.headers = request_headers resp = access.allow_auth0_m2m_token()(request_handler) mock_auth.validate_auth0_token.assert_called_once_with('token', ['yes']) assert not resp @patch('grass.access.auth') @patch('grass.access.config') def test_allow_auth0_m2m_token_with_m2m_token_wrong_profile_type( mock_config: MagicMock, mock_auth: MagicMock, ) -> None: """Test valid token is permitted and sets user.""" mock_config.M2M_API_AUDIENCE = 'no' mock_config.AUTH0_API_AUDIENCE = 'yes' jwt_payload = { 'https://grass.theorchard.com/m2m_profile_type': 'UnknownProfile', 'https://grass.theorchard.com/m2m_profile_id': 321, 'sub': 'auth0|123', } mock_auth.validate_auth0_token.return_value = (None, 200, jwt_payload) request_headers = {headers.AUTHORIZATION: 'Bearer token'} request_handler = fixture_request.RequestHandler() request_handler.request.headers = request_headers resp = access.allow_auth0_m2m_token()(request_handler) mock_auth.validate_auth0_token.assert_called_with('token', ['yes']) assert not resp @patch('grass.access.oa_user') @patch('grass.access.auth') @patch('grass.access.config') def test_allow_auth0_m2m_token_with_m2m_token( mock_config: MagicMock, mock_auth: MagicMock, mock_oa_user: MagicMock, ): """Test valid token is permitted and sets user.""" mock_config.M2M_API_AUDIENCE = 'no' mock_config.AUTH0_API_AUDIENCE = 'yes' jwt_payload = { 'https://grass.theorchard.com/m2m_profile_type': 'OrchAdminProfile', 'https://grass.theorchard.com/m2m_profile_id': 321, 'sub': 'auth0|123', } mock_auth.validate_auth0_token.return_value = (None, 200, jwt_payload) mocked_account = MagicMock() mock_oa_user.get_oa_user_by_id.return_value = response.Response(mocked_account) request_headers = {headers.AUTHORIZATION: 'Bearer token'} request_handler = fixture_request.RequestHandler() request_handler.request.headers = request_headers resp = access.allow_auth0_m2m_token()(request_handler) mock_auth.validate_auth0_token.assert_called_with('token', ['yes']) mock_oa_user.get_oa_user_by_id.assert_called_with('321') assert resp assert request_handler.current_user == mocked_account def test_allow_m2m_token_with_no_token(): """Test no token case.""" request_headers = {headers.GRASS_HEADER_USER_ID: 123} request_handler = fixture_request.RequestHandler() request_handler.request.headers = request_headers resp = access.allow_m2m_token()(request_handler) assert not resp @patch('grass.access.auth') @patch('grass.access.config') def test_allow_m2m_token_with_valid_token( mock_config: MagicMock, mock_auth: MagicMock, ) -> None: """Test token with valid m2m token.""" mock_config.M2M_API_AUDIENCE = 'no' mock_config.AUTH0_API_AUDIENCE = 'yes' mock_auth.validate_auth0_token.return_value = None, 200, 'some token' request_headers = { headers.AUTHORIZATION: 'Bearer token', headers.GRASS_HEADER_USER_ID: 123, } request_handler = fixture_request.RequestHandler() request_handler.request.headers = request_headers resp = access.allow_m2m_token()(request_handler) assert resp mock_auth.validate_auth0_token.assert_called_once_with('token', ['no']) @patch('grass.access.auth') @patch('grass.access.config') def test_allow_m2m_token_with_invalid_token( mock_config: MagicMock, mock_auth: MagicMock, ) -> None: """Test invalid m2m token.""" mock_config.M2M_API_AUDIENCE = 'no' mock_config.AUTH0_API_AUDIENCE = 'yes' mock_auth.validate_auth0_token.return_value = ( 'Unable to parse authentication token, some error', 401, None, ) request_headers = {headers.AUTHORIZATION: 'Bearer token'} request_handler = fixture_request.RequestHandler() request_handler.request.headers = request_headers resp = access.allow_m2m_token()(request_handler) assert not resp mock_auth.validate_auth0_token.assert_called_once_with('token', ['no']) @patch('grass.access.model_session') @patch('grass.access.auth') @patch('grass.access.config') def test_to_validate_allow_auth0_m2m_token(mock_config, mock_auth, mock_model_session): """Test to validate allow auth0 m2m token when ff is on.""" mock_config.M2M_API_AUDIENCE = 'no' mock_config.AUTH0_API_AUDIENCE = 'yes' jwt_payload = {'sub': 'auth0|123'} mock_auth.validate_auth0_token.return_value = None, 200, jwt_payload mock_model_session.is_jwt_cached.return_value = True request_headers = {headers.AUTHORIZATION: 'Bearer token'} request_handler = fixture_request.RequestHandler() request_handler.request.headers = request_headers resp = access.allow_auth0_m2m_token()(request_handler) mock_auth.validate_auth0_token.assert_called_with('token', ['yes']) mock_model_session.is_jwt_cached.assert_called_with('token', 'Unknown') assert not resp @patch('grass.access.model_session') @patch('grass.access.auth') @patch('grass.access.config') def test_to_validate_allow_jwt_token( mock_config: MagicMock, mock_auth: MagicMock, mock_model_session: MagicMock, ) -> None: """Test to validate allow JWT token when ff is on.""" mock_config.M2M_API_AUDIENCE = 'no' mock_config.AUTH0_API_AUDIENCE = 'yes' jwt_payload = {'sub': 'auth0|123'} mock_auth.validate_auth0_token.return_value = (None, 200, jwt_payload) mock_model_session.is_jwt_cached.return_value = True request_headers = {headers.AUTHORIZATION: 'Bearer token'} request_handler = fixture_request.RequestHandler() request_handler.request.headers = request_headers resp = access.allow_jwt_token()(request_handler) mock_auth.validate_auth0_token.assert_called_with('token', ['yes']) mock_model_session.is_jwt_cached.assert_called_with('token', 'Unknown') assert not resp