Added cross tenant check (#2128)

met-api/src/met_api/resources/staff_user.py

@@ -128,7 +128,7 @@ class UserGroup(Resource):
 
     @staticmethod
     @cross_origin(origins=allowedorigins())
-    @require_role([Role.CREATE_ADMIN_USER.value])
+    @require_role([Role.CREATE_ADMIN_USER.value], skip_tenant_check_for_admin=True)
     def post(user_id):
         """Add user to group."""
         try:

met-api/src/met_api/utils/tenant_validator.py

@@ -24,10 +24,21 @@
 
 from met_api.auth import jwt as _jwt
 from met_api.utils.constants import TENANT_ID_JWT_CLAIM
+from met_api.utils.roles import Role
 
 
-def require_role(role):
-    """Validate a token for roles and against tenant as well."""
+def require_role(role, skip_tenant_check_for_admin=False):
+    """Validate a token for roles and against tenant as well.
+
+    Args:
+        role (str): The role that the user is required to have.
+        skip_tenant_check_for_admin (bool, optional): A flag to indicate whether to skip tenant checks for MET Admins.
+            If set to True, tenant checks are skipped for users with MET administrative privileges.
+            Defaults to False. Set it to True for cross tenant operations like first time adding a super user to tenant.
+
+    Returns:
+        function: A decorator function that can be used to enforce role-based authorization.
+    """
 
     def decorator(func):
         @wraps(func)
@@ -36,8 +47,13 @@ def wrapper(*args, **kwargs):
             # single tenanted env doesn't need tenant id checks..so pass
             if current_app.config.get('IS_SINGLE_TENANT_ENVIRONMENT'):
                 return func(*args, **kwargs)
+
             # Get the tenant information from the JWT payload or any global context
             token_info: Dict = _get_token_info() or {}
+
+            if skip_tenant_check_for_admin and is_met_global_admin(token_info):
+                return func(*args, **kwargs)
+
             tenant_id = token_info.get(TENANT_ID_JWT_CLAIM, None)
             current_app.logger.debug(f'Tenant Id From JWT Claim {tenant_id}')
             current_app.logger.debug(f'Tenant Id From g {g.tenant_id}')
@@ -54,3 +70,11 @@ def wrapper(*args, **kwargs):
 
 def _get_token_info() -> Dict:
     return g.jwt_oidc_token_info if g and 'jwt_oidc_token_info' in g else {}
+
+
+def is_met_global_admin(token_info) -> bool:
+    """Return True if the user is MET Admin ie who can manage all tenants."""
+    roles: list = token_info.get('realm_access', None).get('roles', []) if 'realm_access' in token_info \
+        else []
+
+    return Role.CREATE_TENANT.value in roles

met-api/tests/unit/api/test_user.py

@@ -17,6 +17,7 @@
 
 Test-Suite to ensure that the /user endpoint is working as expected.
 """
+import copy
 from http import HTTPStatus
 from unittest.mock import MagicMock
 
@@ -147,6 +148,44 @@ def test_add_user_to_team_member_group(mocker, client, jwt, session):
     mock_get_user_groups_keycloak.assert_called()
 
 
+def test_add_user_to_team_member_group_across_tenants(mocker, client, jwt, session):
+    """Assert that a user can be added to the team member group."""
+    set_global_tenant(tenant_id=1)
+    user = factory_staff_user_model()
+
+    mock_add_user_to_group_keycloak, mock_get_user_groups_keycloak, mock_add_attribute_to_user = mock_add_user_to_group(
+        mocker,
+        [KeycloakGroupName.EAO_IT_VIEWER.value]
+    )
+
+    claims = copy.deepcopy(TestJwtClaims.staff_admin_role.value)
+    # sets a different tenant id in the request
+    claims['tenant_id'] = 2
+    headers = factory_auth_header(jwt=jwt, claims=claims)
+    rv = client.post(
+        f'/api/user/{user.external_id}/groups?group=TeamMember',
+        headers=headers,
+        content_type=ContentType.JSON.value
+    )
+    # assert staff admin cant do cross tenant operation
+    assert rv.status_code == HTTPStatus.FORBIDDEN
+
+    claims = copy.deepcopy(TestJwtClaims.met_admin_role.value)
+    # sets a different tenant id in the request
+    claims['tenant_id'] = 2
+    headers = factory_auth_header(jwt=jwt, claims=claims)
+    rv = client.post(
+        f'/api/user/{user.external_id}/groups?group=TeamMember',
+        headers=headers,
+        content_type=ContentType.JSON.value
+    )
+    # assert MET admin can do cross tenant operation
+    assert rv.status_code == HTTPStatus.OK
+
+    mock_add_user_to_group_keycloak.assert_called()
+    mock_get_user_groups_keycloak.assert_called()
+
+
 def mock_toggle_user_status(mocker):
     """Mock the KeycloakService.add_user_to_group method."""
     mock_response = MagicMock()

met-api/tests/utilities/factory_scenarios.py

@@ -273,6 +273,32 @@ class TestJwtClaims(dict, Enum):
         }
     }
 
+    met_admin_role = {
+        'iss': CONFIG.JWT_OIDC_TEST_ISSUER,
+        'sub': 'f7a4a1d3-73a8-4cbc-a40f-bb1145302064',
+        'idp_userid': 'f7a4a1d3-73a8-4cbc-a40f-bb1145302064',
+        'preferred_username': f'{fake.user_name()}@idir',
+        'given_name': fake.first_name(),
+        'family_name': fake.last_name(),
+        'tenant_id': 1,
+        'email': '[email protected]',
+        'identity_provider': LoginSource.IDIR.value,
+        'realm_access': {
+            'roles': [
+                'staff',
+                'view_engagement',
+                'create_survey',
+                'view_users',
+                'create_admin_user',
+                'edit_members',
+                'toggle_user_status',
+                'export_to_csv',
+                'update_user_group',
+                'create_tenant'
+            ]
+        }
+    }
+
     staff_admin_role = {
         'iss': CONFIG.JWT_OIDC_TEST_ISSUER,
         'sub': 'f7a4a1d3-73a8-4cbc-a40f-bb1145302064',
@@ -305,7 +331,7 @@ class TestJwtClaims(dict, Enum):
                 'view_all_engagements',
                 'toggle_user_status',
                 'export_to_csv',
-                'update_user_group'
+                'update_user_group',
             ]
         }
     }