"""Tests for getting context from request headers.""" from unittest.mock import MagicMock from unittest.mock import patch from jwtauth import JWTAuth from owsresponse import response from owsresponse import status as response_status import pytest from requests.structures import CaseInsensitiveDict from owsrequest import context from owsrequest.constants import headers from owsrequest.constants.headers import DEFAULT_BRAND @patch('owsrequest.context.auth.validate_and_decode_jwt_token') @pytest.mark.parametrize(( 'request_headers', 'expected_context_type', 'expected_profile_type', 'expected_profile_id', 'expected_identity_id', 'expected_identity_uuid', 'has_orchard_user_id', 'expected_orchard_user_id', 'label_profile', 'expected_roles', 'expected_requestor_service_name', ), [ # OrchAdmin headers, profile takes precedence, add OrchAdminProfile type ( { headers.ORCHARD_USER_ID: 'oa:123', headers.ORCHARD_PROFILE_TYPE: headers.PROFILE_TYPE_ORCH_ADMIN, headers.ORCHARD_PROFILE_ID: 555, headers.ORCHARD_IDENTITY_ID: '5c067bc05f37972e07cb605f', headers.ORCHARD_IDENTITY_UUID: 'f5223b2c-e22b-4b48-a8b4-13e50b87b423', # noqa headers.ORCHARD_ROLES: 'catalog,analytics', headers.ORCHARD_REQUESTOR_SERVICE: 'ows-grass' }, headers.CONTEXT_TYPE_PROFILE, headers.PROFILE_TYPE_ORCH_ADMIN, 555, '5c067bc05f37972e07cb605f', 'f5223b2c-e22b-4b48-a8b4-13e50b87b423', True, 'oa:123', False, ['catalog', 'analytics'], 'ows-grass', ), # OrchAdmin headers - forced profiled ( { headers.ORCHARD_USER_ID: 'oa:777', headers.ORCHARD_ROLES: 'catalog,analytics' }, headers.CONTEXT_TYPE_PROFILE, headers.PROFILE_TYPE_ORCH_ADMIN, 777, None, None, None, None, True, ['catalog', 'analytics'], None, ), # OrchAdmin headers - no forced label ( { headers.ORCHARD_USER_ID: 'oa:777', headers.ORCHARD_ROLES: 'catalog,analytics' }, headers.CONTEXT_TYPE_ACCOUNT, None, None, None, None, True, 'oa:777', False, ['catalog', 'analytics'], None, ), # account based headers, all valid ( { headers.GRASS_ACCOUNT_TYPE: 'vendor', headers.GRASS_ACCOUNT_ID: 7123, headers.ORCHARD_USER_ID: 'alw:777', headers.ORCHARD_ROLES: 'catalog,analytics', headers.ORCHARD_REQUESTOR_SERVICE: 'ows-grass' }, headers.CONTEXT_TYPE_ACCOUNT, 'vendor', 7123, None, None, True, 'alw:777', False, ['catalog', 'analytics'], 'ows-grass', ), # account based headers, all valid, add identity_id ( { headers.GRASS_ACCOUNT_TYPE: 'vendor', headers.GRASS_ACCOUNT_ID: 7123, headers.ORCHARD_USER_ID: 'alw:777', headers.ORCHARD_IDENTITY_ID: '5c067bc05f37972e07cb605f', headers.ORCHARD_ROLES: 'admin', headers.ORCHARD_REQUESTOR_SERVICE: 'graphql-product' }, headers.CONTEXT_TYPE_ACCOUNT, 'vendor', 7123, '5c067bc05f37972e07cb605f', None, True, 'alw:777', False, ['admin'], 'graphql-product', ), # profile based headers, all valid ( { headers.ORCHARD_PROFILE_TYPE: headers.PROFILE_TYPE_ARTIST, headers.ORCHARD_PROFILE_ID: 555 }, headers.CONTEXT_TYPE_PROFILE, headers.PROFILE_TYPE_ARTIST, 555, None, None, False, None, False, None, None, ), # profile based headers, all valid, add identity_id ( { headers.ORCHARD_PROFILE_TYPE: headers.PROFILE_TYPE_ARTIST, headers.ORCHARD_PROFILE_ID: 555, headers.ORCHARD_IDENTITY_ID: '5c067bc05f37972e07cb605f' }, headers.CONTEXT_TYPE_PROFILE, headers.PROFILE_TYPE_ARTIST, 555, '5c067bc05f37972e07cb605f', None, False, None, False, None, None, ), # profile UUID headers, all valid ( { headers.ORCHARD_PROFILE_UUID: 'profile-uuid', headers.ORCHARD_IDENTITY_ID: '5c067bc05f37972e07cb605f' }, headers.CONTEXT_TYPE_PROFILE, None, None, '5c067bc05f37972e07cb605f', None, False, None, False, None, None, ), # both full headers, profile takes precedence ( { headers.GRASS_ACCOUNT_TYPE: 'vendor', headers.GRASS_ACCOUNT_ID: 7123, headers.ORCHARD_USER_ID: 'alw:777', headers.ORCHARD_PROFILE_TYPE: headers.PROFILE_TYPE_ARTIST, headers.ORCHARD_PROFILE_ID: 555, headers.ORCHARD_IDENTITY_ID: '5c067bc05f37972e07cb605f', headers.ORCHARD_IDENTITY_UUID: 'f5223b2c-e22b-4b48-a8b4-13e50b87b423' # noqa }, headers.CONTEXT_TYPE_PROFILE, headers.PROFILE_TYPE_ARTIST, 555, '5c067bc05f37972e07cb605f', 'f5223b2c-e22b-4b48-a8b4-13e50b87b423', None, None, False, None, None, ), # no headers ( {}, headers.CONTEXT_TYPE_NONE, None, None, None, None, False, None, False, None, None, ), # malformed user grass headers, but ok user id ( { headers.GRASS_ACCOUNT_TYPE: 'vendor', headers.ORCHARD_USER_ID: 'alw:777' }, headers.CONTEXT_TYPE_ACCOUNT, None, None, None, None, True, 'alw:777', False, None, None, ), # malformed user grass headers, no user id ( { headers.GRASS_ACCOUNT_TYPE: 'vendor' }, headers.CONTEXT_TYPE_ERROR, None, None, None, None, False, None, False, None, None, ), # malformed profile headers, but ok identity id ( { headers.ORCHARD_PROFILE_TYPE: 'ArtistProfile', headers.ORCHARD_IDENTITY_ID: '5c067bc05f37972e07cb605f' }, headers.CONTEXT_TYPE_PROFILE, None, None, '5c067bc05f37972e07cb605f', None, False, None, False, None, None, ), # malformed profile headers, no identity id ( { headers.ORCHARD_PROFILE_TYPE: 'ArtistProfile' }, headers.CONTEXT_TYPE_ERROR, None, None, None, None, False, None, False, None, None, ), # orchard user header, no profile ( { headers.ORCHARD_USER_ID: 'alw:777', headers.ORCHARD_IDENTITY_ID: '5c067bc05f37972e07cb605f' }, headers.CONTEXT_TYPE_PROFILE, headers.PROFILE_TYPE_LABEL, 777, '5c067bc05f37972e07cb605f', None, False, None, True, None, None, ), # only orchard identity id ( { headers.ORCHARD_IDENTITY_ID: '5c067bc05f37972e07cb605f' }, headers.CONTEXT_TYPE_PROFILE, None, None, '5c067bc05f37972e07cb605f', None, False, None, True, None, None, ), # only orchard identity uuid ( { headers.ORCHARD_IDENTITY_UUID: 'f5223b2c-e22b-4b48-a8b4-13e50b87b423' # noqa }, headers.CONTEXT_TYPE_PROFILE, None, None, None, 'f5223b2c-e22b-4b48-a8b4-13e50b87b423', False, None, True, None, None, ), # only orchard identity uuid, title cased ( { 'Orchard-Identity-Uuid': 'f5223b2c-e22b-4b48-a8b4-13e50b87b423' # noqa }, headers.CONTEXT_TYPE_PROFILE, None, None, None, 'f5223b2c-e22b-4b48-a8b4-13e50b87b423', False, None, True, None, None, ), # only orchard identity uuid, wonky cased ( { 'OrCharD-IdEnTIty-UuiD': 'f5223b2c-e22b-4b48-a8b4-13e50b87b423' # noqa }, headers.CONTEXT_TYPE_PROFILE, None, None, None, 'f5223b2c-e22b-4b48-a8b4-13e50b87b423', False, None, True, None, None, ), # jwt contains brand ( { 'Orchard-Identity-Uuid': 'f5223b2c-e22b-4b48-a8b4-13e50b87b423' # noqa }, headers.CONTEXT_TYPE_PROFILE, None, None, None, 'f5223b2c-e22b-4b48-a8b4-13e50b87b423', False, None, True, None, None, ) ]) def test_get_request_context_from_headers( mock_validate_and_decode_jwt_token, request_headers, expected_context_type, expected_profile_type, expected_profile_id, expected_identity_id, expected_identity_uuid, has_orchard_user_id, expected_orchard_user_id, label_profile, expected_roles, expected_requestor_service_name, ): """Test request context is populated with headers. Headers do NOT include Authorization, thus jwt validate/decode is not used. """ mock_jwt_auth_client = MagicMock(spec=JWTAuth) request_context = context.get_request_context_from_headers( CaseInsensitiveDict(request_headers), label_profile=label_profile, jwt_auth_client=mock_jwt_auth_client, ) assert request_context.context_type == expected_context_type assert hasattr(request_context, 'identity_id') assert hasattr(request_context, 'profile_type') assert hasattr(request_context, 'profile_id') if expected_profile_type: assert request_context.profile_type == expected_profile_type if expected_profile_id: assert request_context.profile_id == expected_profile_id if expected_identity_id: assert request_context.identity_id == expected_identity_id if expected_identity_uuid: assert request_context.identity_uuid == expected_identity_uuid if has_orchard_user_id: assert request_context.orchard_user_id == expected_orchard_user_id if expected_roles: assert request_context.roles == expected_roles else: assert not request_context.roles assert (request_context.requestor_service_name == expected_requestor_service_name) mock_validate_and_decode_jwt_token.assert_not_called() assert request_context.brand == DEFAULT_BRAND,\ 'No Authorization header passed, no JWT decoding, should defer to DEFAULT_BRAND' if request_headers.get(headers.ORCHARD_PROFILE_UUID): assert request_context.profile_uuid == request_headers.get( headers.ORCHARD_PROFILE_UUID) assert request_context.context_type == headers.CONTEXT_TYPE_PROFILE @patch('owsrequest.context.jwt_auth_from_environment') @patch('owsrequest.context.auth.validate_and_decode_jwt_token') @pytest.mark.parametrize('mock_jwt_auth_client', [(None), MagicMock(spec=JWTAuth)]) @pytest.mark.parametrize( ('request_headers', 'decoded_jwt', 'expected_brand', 'expected_identity_id', 'description'), [ ( {}, None, 'orchard', None, 'no authorization header, use default brand and identity id is none', ), ( {'Authorization': 'Bearer jwt-here'}, { headers.JWT_BRAND_FIELD: 'awal', headers.JWT_USER_METADATA_FIELD: { 'orchardIdentityId': 'here i am', } }, 'awal', 'here i am', 'Authorization header has a brand and identity id' ), ]) def test_get_request_context_from_headers_authorization_only( mock_validate_and_decode_jwt_token, mock_jwt_auth_from_environment, mock_jwt_auth_client, request_headers, decoded_jwt, expected_brand, expected_identity_id, description, ): """Test get_request_context_from_headers when only the Authorization header is present.""" mock_validate_and_decode_jwt_token.return_value = response.Response(decoded_jwt) on_the_fly_mock_jwt_auth_client = MagicMock(spec=JWTAuth) mock_jwt_auth_from_environment.return_value = on_the_fly_mock_jwt_auth_client request_context = context.get_request_context_from_headers( CaseInsensitiveDict(request_headers), jwt_auth_client=mock_jwt_auth_client, ) assert request_context.brand == expected_brand, description assert request_context.jwt_identity_id == expected_identity_id, description if decoded_jwt: mock_validate_and_decode_jwt_token.assert_called_once_with( 'jwt-here', jwt_auth_client=mock_jwt_auth_client or on_the_fly_mock_jwt_auth_client, ) if mock_jwt_auth_client: mock_jwt_auth_from_environment.assert_not_called() else: mock_jwt_auth_from_environment.assert_called_once() @patch('owsrequest.context.auth.validate_and_decode_jwt_token') @patch('owsrequest.context.logger') def test_default_brand_is_default(mock_logger, mock_validate_and_decode_jwt_token): """Test brand is DEFAULT_BRAND when there is an error decoding the JWT Token.""" mock_jwt_auth_client = MagicMock(spec=JWTAuth) mock_validate_and_decode_jwt_token.return_value = response.create_error_response( code='unauthorized', message='bad jwt', status=response_status.UNAUTHORIZED ) request_headers = { headers.ORCHARD_IDENTITY_UUID: 'f5223b2c-e22b-4b48-a8b4-13e50b87b423', headers.AUTHORIZATION: 'Bearer lololol' } request_context = context.get_request_context_from_headers( CaseInsensitiveDict(request_headers), label_profile=True, jwt_auth_client=mock_jwt_auth_client) assert request_context.brand is DEFAULT_BRAND mock_logger.info.assert_called() mock_validate_and_decode_jwt_token.assert_called_once_with( 'lololol', jwt_auth_client=mock_jwt_auth_client) def test_get_request_context_from_headers_invalid_param(): """Test that request_headers must be CaseInsensitiveDict.""" type_error = None try: context.get_request_context_from_headers( { headers.ORCHARD_PROFILE_TYPE: 'ArtistProfile', headers.ORCHARD_IDENTITY_ID: '5c067bc05f37972e07cb605f' }, False) except TypeError as type_error: # noqa: F841 assert type_error.args[0] ==\ 'request_headers must be instance of CaseInsensitiveDict' @pytest.mark.parametrize( ('decoded_jwt, expected_identity_id, description'), [ ({}, None, 'Empty jwt means no identity id'), ( {'https://grass.theorchard.com/user_metadata': {}}, None, 'Empty user metadata means no identity id', ), ( {'https://grass.theorchard.com/user_metadata': {'orchardIdentityId': 'my id'}}, 'my id', 'Expect identity id from user_metadata', ), ] ) def test_get_jwt_identity_id(decoded_jwt, expected_identity_id, description): """Test _get_jwt_identity_id.""" actual = context._get_jwt_identity_id(decoded_jwt) assert actual == expected_identity_id, description