add permission layer for admins for backend API

areas/auth/auth.py

@@ -37,7 +37,7 @@ def hydra_callback():
             identity = i
 
     access_token = create_access_token(
-        identity=token, expires_delta=timedelta(days=365)
+        identity=token, expires_delta=timedelta(days=365), additional_claims={"user_id": identity["id"]}
     )
 
     apps = App.query.all()

areas/roles/role_service.py

@@ -1,3 +1,4 @@
+from areas.apps.models import AppRole
 from .models import Role
 
 
@@ -10,3 +11,7 @@ class RoleService:
     @staticmethod
     def get_role_by_id(role_id):
         return Role.query.filter_by(id=role_id).first()
+
+    def is_user_admin(userId):
+        dashboard_role_id = AppRole.query.filter_by(user_id=userId, app_id=1).first().role_id
+        return dashboard_role_id == 1

areas/users/users.py

@@ -5,6 +5,7 @@ from flask_expects_json import expects_json
 
 from areas import api_v1
 from helpers import KratosApi
+from helpers.auth_guard import admin_required
 
 from .validation import schema
 from .user_service import UserService
@@ -13,6 +14,7 @@ from .user_service import UserService
 @api_v1.route("/users", methods=["GET"])
 @jwt_required()
 @cross_origin()
+@admin_required()
 def get_users():
     res = UserService.get_users()
     return jsonify(res)
@@ -49,6 +51,7 @@ def put_user(id):
 @api_v1.route("/users/<string:id>", methods=["DELETE"])
 @jwt_required()
 @cross_origin()
+@admin_required()
 def delete_user(id):
     res = KratosApi.delete("/identities/{}".format(id))
     if res.status_code == 204:

helpers/auth_guard.py

@@ -0,0 +1,24 @@
+from functools import wraps
+
+from flask import jsonify
+from areas.roles.role_service import RoleService
+
+from flask_jwt_extended import verify_jwt_in_request
+from flask_jwt_extended import get_jwt
+
+def admin_required():
+    def wrapper(fn):
+        @wraps(fn)
+        def decorator(*args, **kwargs):
+            verify_jwt_in_request()
+            claims = get_jwt()
+            userId = claims["user_id"]
+            isAdmin = RoleService.is_user_admin(userId)
+            if isAdmin:
+                return fn(*args, **kwargs)
+            else:
+                return jsonify(msg="Admins only!"), 403
+
+        return decorator
+
+    return wrapper