Added authorisation checks (#1294)

epictrack-api/src/api/__init__.py

@@ -21,6 +21,7 @@
 from http import HTTPStatus
 
 from flask import Flask
+from flask import current_app
 from marshmallow import ValidationError
 
 from api import config
@@ -85,8 +86,17 @@ def handle_error(err):
 
 def setup_jwt_manager(app, jwt_manager):
     """Use flask app to configure the JWTManager to work for a particular Realm."""
+
     def get_roles(a_dict):
-        return a_dict['realm_access']['roles']  # pragma: no cover
+        realm_access = a_dict.get('realm_access', {})
+        realm_roles = realm_access.get('roles', [])
+
+        client_name = current_app.config.get('JWT_OIDC_AUDIENCE')
+
+        resource_access = a_dict.get('resource_access', {})
+        client_roles = resource_access.get(client_name, {}).get('roles', [])
+
+        return realm_roles + client_roles
 
     app.config['JWT_ROLE_CALLBACK'] = get_roles
 

epictrack-api/src/api/resources/work_issues.py

@@ -20,6 +20,7 @@
 from api.schemas import request as req
 from api.schemas import response as res
 from api.services import WorkIssuesService
+from api.utils.roles import Role
 from api.utils import auth, profiletime
 from api.utils.util import cors_preflight
 
@@ -91,7 +92,7 @@ class ApproveIssues(Resource):
 
     @staticmethod
     @cors.crossdomain(origin="*")
-    @auth.require
+    @auth.has_one_of_roles([[Role.CREATE.value]])
     @profiletime
     # pylint: disable=unused-argument
     def patch(work_id, issue_id, update_id):

epictrack-api/src/api/services/authorisation.py

@@ -0,0 +1,51 @@
+"""The Authorization service.
+
+This module is to handle authorization related queries.
+"""
+
+from flask_restx import abort
+
+from api.utils import TokenInfo
+from api.utils.roles import Membership
+from api.models import Staff as StaffModel
+from api.models import StaffWorkRole as StaffWorkRoleModel
+
+
+# pylint: disable=unused-argument,inconsistent-return-statements
+def check_auth(**kwargs):
+    """Check if user is authorized to perform action on the service."""
+    token_roles = set(TokenInfo.get_roles())
+    permitted_roles = set(kwargs.get('one_of_roles', []))
+    has_valid_roles = token_roles & permitted_roles
+    if has_valid_roles:
+        return
+
+    matching_memberships = {membership.name for membership in Membership} & permitted_roles
+
+    if matching_memberships and _has_team_membership(kwargs, matching_memberships):
+        return True
+
+    abort(403)
+
+
+def _has_team_membership(kwargs, team_permitted_roles) -> bool:
+    work_id = kwargs.get('work_id')
+
+    if not work_id:
+        return False
+
+    email = TokenInfo.get_user_data()['email_id']
+    staff_model: StaffModel = StaffModel.find_by_email(email)
+
+    work_roles = StaffWorkRoleModel.find_by_params(
+        {"work_id": work_id, "staff_id": staff_model.id}
+    )
+    if not work_roles:
+        return False
+
+    if Membership.TEAM_MEMBER in team_permitted_roles:
+        return bool(work_roles)
+
+    membership_ids = {membership.value for membership in Membership}
+
+    return any(role.id in membership_ids for role in work_roles)

epictrack-api/src/api/services/work.py

@@ -336,6 +336,7 @@ def update_work(cls, work_id: int, payload: dict):
         work = Work.find_by_id(work_id)
         if not work:
             raise ResourceNotFoundError(f"Work with id '{work_id}' not found")
+
         work = work.update(payload)
         return work
 

epictrack-api/src/api/services/work_status.py

@@ -19,6 +19,9 @@
 from api.exceptions import BusinessError
 from api.models import WorkStatus as WorkStatusModel
 from api.utils import TokenInfo
+from api.utils.roles import Membership
+from api.services import authorisation
+from api.utils.roles import Role as KeycloakRole
 
 
 class WorkStatusService:  # pylint: disable=too-many-public-methods
@@ -69,6 +72,12 @@ def approve_work_status(cls, work_status):
         if work_status.is_approved:
             return work_status
 
+        one_of_roles = (
+            Membership.TEAM_MEMBER.name,
+            KeycloakRole.EDIT.value
+        )
+        authorisation.check_auth(one_of_roles=one_of_roles, work_id=work_status.work_id)
+
         work_status.is_approved = True
         work_status.approved_by = TokenInfo.get_username()
         work_status.approved_date = datetime.utcnow()

epictrack-api/src/api/utils/roles.py

@@ -0,0 +1,34 @@
+# Copyright © 2019 Province of British Columbia
+#
+# Licensed under the Apache License, Version 2.0 (the 'License');
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an 'AS IS' BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Role definitions."""
+from enum import Enum
+
+
+class Role(Enum):
+    """User Role."""
+
+    # Keycloak Based roles
+    CREATE = 'create'
+    EDIT = 'edit'
+
+
+class Membership(Enum):
+    """User Position in EAO"""
+
+    EPD = 1
+    LEAD = 2
+    OTHER = 3
+    FNCAIRT = 4
+    ANALYST = 5
+    TEAM_MEMBER = 'TEAM_MEMBER'

epictrack-api/src/api/utils/token_info.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """Helper for token decoding"""
-from flask import g
+from flask import g, current_app
 
 
 class TokenInfo:
@@ -55,3 +55,15 @@ def is_super_user() -> bool:
         """Return True if the user is staff user."""
         # TODO Implement this method
         return True
+
+    @staticmethod
+    def get_roles():
+        """Return roles of a user from the token."""
+        token_info = g.token_info
+        realm_access = token_info.get('realm_access', {})
+        realm_roles = realm_access.get('roles', [])
+        client_name = current_app.config.get('JWT_OIDC_AUDIENCE')
+        resource_access = token_info.get('resource_access', {})
+        client_roles = resource_access.get(client_name, {}).get('roles', [])
+
+        return realm_roles + client_roles