Feature.authorization rbac

src/cnaas_nms/api/app.py

@@ -3,16 +3,24 @@
 import sys
 from typing import Optional
 
+from authlib.integrations.flask_client import OAuth
 from engineio.payload import Payload
 from flask import Flask, jsonify, request
 from flask_cors import CORS
 from flask_jwt_extended import JWTManager
 from flask_jwt_extended.exceptions import InvalidHeaderError, NoAuthorizationError
 from flask_restx import Api
 from flask_socketio import SocketIO, join_room
-from jwt.exceptions import DecodeError, InvalidSignatureError, InvalidTokenError, ExpiredSignatureError, InvalidKeyError
-from authlib.integrations.flask_client import OAuth
+from jwt.exceptions import (
+    DecodeError,
+    ExpiredSignatureError,
+    InvalidAudienceError,
+    InvalidSignatureError,
+    InvalidTokenError,
+    InvalidKeyError
+)
 
+from cnaas_nms.api.auth import api as auth_api
 from cnaas_nms.api.device import (
     device_api,
     device_cert_api,
@@ -25,7 +33,6 @@
     device_update_interfaces_api,
     devices_api,
 )
-from cnaas_nms.api.auth import api as auth_api
 from cnaas_nms.api.firmware import api as firmware_api
 from cnaas_nms.api.groups import api as groups_api
 from cnaas_nms.api.interface import api as interfaces_api
@@ -37,15 +44,11 @@
 from cnaas_nms.api.repository import api as repository_api
 from cnaas_nms.api.settings import api as settings_api
 from cnaas_nms.api.system import api as system_api
-
-from cnaas_nms.app_settings import auth_settings
-from cnaas_nms.app_settings import api_settings
-
+from cnaas_nms.app_settings import api_settings, auth_settings
 from cnaas_nms.tools.log import get_logger
 from cnaas_nms.tools.security import get_identity, login_required
 from cnaas_nms.version import __api_version__
 
-
 logger = get_logger()
 
 
@@ -64,26 +67,38 @@
 class CnaasApi(Api):
     def handle_error(self, e):
         if isinstance(e, DecodeError):
-            data = {"status": "error", "data": "Could not decode JWT token"}
+            data = {"status": "error", "message": "Could not decode JWT token"}
+            return jsonify(data), 401
+        elif isinstance(e, InvalidAudienceError):
+            data = {"status": "error", "message": "You don't seem to have the rights to execute this call"}
+            return jsonify(data), 403
+        elif isinstance(e, ExpiredSignatureError):
+            data = {"status": "error", "message": "The JWT token is expired", "errorCode": "auth_expired"}
+            return jsonify(data), 401
         elif isinstance(e, InvalidKeyError):
             data = {"status": "error", "data": "Invalid keys {}".format(e)}
+            return jsonify(data), 401
         elif isinstance(e, InvalidTokenError):
-            data = {"status": "error", "data": "Invalid authentication header: {}".format(e)}
+            data = {"status": "error", "message": "Invalid authentication header: {}".format(e)}
+            return jsonify(data), 401
         elif isinstance(e, InvalidSignatureError):
-            data = {"status": "error", "data": "Invalid token signature"}
+            data = {"status": "error", "message": "Invalid token signature"}
+            return jsonify(data), 401
         elif isinstance(e, IndexError):
             # We might catch IndexErrors which are not caused by JWT,
             # but this is better than nothing.
-            data = {"status": "error", "data": "JWT token missing?"}
+            data = {"status": "error", "message": "JWT token missing?"}
+            return jsonify(data), 401
         elif isinstance(e, NoAuthorizationError):
-            data = {"status": "error", "data": "JWT token missing?"}
+            data = {"status": "error", "message": "JWT token missing?"}
+            return jsonify(data), 401
         elif isinstance(e, InvalidHeaderError):
-            data = {"status": "error", "data": "Invalid header, JWT token missing? {}".format(e)}
-        elif isinstance(e, ExpiredSignatureError):
-            data = {"status": "error", "data": "The JWT token is expired"}
+            data = {"status": "error", "message": "Invalid header, JWT token missing? {}".format(e)}
+            return jsonify(data), 401
         else:
             return super(CnaasApi, self).handle_error(e)
-        return jsonify(data), 401
+
+
 
 
 app = Flask(__name__)
@@ -196,18 +211,14 @@ def socketio_on_events(data):
 def log_request(response):
     try:
         url = re.sub(jwt_query_r, "", request.url)
-        if request.headers.get('content-type') == 'application/json':
+        if request.headers.get("content-type") == "application/json":
             logger.info(
                 "Method: {}, Status: {}, URL: {}, JSON: {}".format(
                     request.method, response.status_code, url, request.json
                 )
             )
         else:
-            logger.info(
-                "Method: {}, Status: {}, URL: {}".format(
-                    request.method, response.status_code, url
-                )
-            )
+            logger.info("Method: {}, Status: {}, URL: {}".format(request.method, response.status_code, url))
     except Exception:
         pass
     return response

src/cnaas_nms/api/auth.py

@@ -1,12 +1,15 @@
 from authlib.integrations.base_client.errors import MismatchingStateError
+from authlib.integrations.flask_oauth2 import current_token
+
 from flask import current_app, redirect, url_for
 from flask_restx import Namespace, Resource
 from requests.models import PreparedRequest
 
 from cnaas_nms.api.generic import empty_result
 from cnaas_nms.app_settings import auth_settings
+from cnaas_nms.tools.security import login_required, get_identity, get_oauth_userinfo, login_required_all_permitted
+from cnaas_nms.tools.rbac.rbac import get_permissions_user
 from cnaas_nms.tools.log import get_logger
-from cnaas_nms.tools.security import get_identity, login_required
 from cnaas_nms.version import __api_version__
 
 logger = get_logger()
@@ -80,6 +83,19 @@ def get(self):
         return identity
 
 
+class PermissionsAPI(Resource):
+    @login_required_all_permitted
+    def get(self):
+        permissions_rules = auth_settings.PERMISSIONS
+        if len(permissions_rules) == 0:
+            logger.debug('No permissions defined, so nobody is permitted to do any api calls.')
+            return {}
+        user_info = get_oauth_userinfo(current_token)
+        permissions_of_user = get_permissions_user(permissions_rules, user_info)
+        return permissions_of_user
+
+
 api.add_resource(LoginApi, "/login")
 api.add_resource(AuthApi, "/auth")
 api.add_resource(IdentityApi, "/identity")
+api.add_resource(PermissionsAPI, "/permissions")

src/cnaas_nms/app_settings.py

@@ -69,10 +69,12 @@ class AuthSettings(BaseSettings):
     OIDC_CLIENT_SECRET: str = "xxx"
     OIDC_CLIENT_ID: str = "client-id"
     OIDC_ENABLED: bool = False
+    PERMISSIONS: dict = {}
+    PERMISSIONS_DISABLED: bool = False
 
 
 def construct_api_settings() -> ApiSettings:
-    api_config = Path("/etc/cnaas-nms/api.yml")
+    api_config = Path("etc/cnaas-nms/api.yml")
 
     if api_config.is_file():
         with open(api_config, "r") as api_file:
@@ -109,7 +111,6 @@ def construct_api_settings() -> ApiSettings:
 def construct_app_settings() -> AppSettings:
     db_config = Path("/etc/cnaas-nms/db_config.yml")
     repo_config = Path("/etc/cnaas-nms/repository.yml")
-
     app_settings = AppSettings()
 
     def _create_db_config(settings: AppSettings, config: dict) -> None:
@@ -142,6 +143,19 @@ def _create_repo_config(settings: AppSettings, config: dict) -> None:
 
 def construct_auth_settings() -> AuthSettings:
     auth_settings = AuthSettings()
+    permission_config = Path("/etc/cnaas-nms/permissions.yml")
+
+    def _create_permissions_config(settings: AuthSettings, permissions_rules: dict) -> None:
+        settings.PERMISSIONS = permissions_rules
+
+    if auth_settings.PERMISSIONS_DISABLED:
+        auth_settings.PERMISSIONS = {'config': {'default_permissions': 'default'}, 'roles': {'default': {'allowed_api_methods': ['*'], 'allowed_api_calls': ['*']}}}
+
+    elif permission_config.is_file():
+        '''Load the file with role permission'''
+        with open(permission_config, "r") as permission_file:
+            permissions_rules = yaml.safe_load(permission_file)
+        _create_permissions_config(auth_settings, permissions_rules)
     return auth_settings
 
 

src/cnaas_nms/roles.yml

@@ -0,0 +1,9 @@
+
+-
+  role: read
+  allowed_api_methods: ["GET"]
+  allowed_api_calls: ["/devices", "/jobs"]
+-
+  role: write
+  allowed_api_methods: ["GET", "PUT"]
+  allowed_api_calls: []

src/cnaas_nms/tools/rbac/rbac.py

@@ -0,0 +1,61 @@
+import re
+import fnmatch
+from authlib.oauth2.rfc6749.wrappers import HttpRequest
+
+from cnaas_nms.version import __api_version__
+
+def get_permissions_user(permissions_rules, user_info):
+    '''Get the API permissions of the user'''
+    permissions_of_user = []
+
+    # if no rules, return
+    if len(permissions_rules) == 0:
+        return permissions_of_user
+
+    # first give all the permissions of the fallback role
+    if "default_permissions" in permissions_rules["config"] and permissions_rules["config"]["default_permissions"] in permissions_rules["roles"]:
+        permissions_of_user.extend(permissions_rules["roles"][permissions_rules["config"]["default_permissions"]]["permissions"])
+
+    # if the element is not defined or the user doesn't have the element, return
+    if 'group_claim_key' not in permissions_rules['config'] or permissions_rules['config']['group_claim_key'] not in user_info:
+        return permissions_of_user
+
+    # make the roles of userinfo into the right format, a list of roles
+    if isinstance(user_info[permissions_rules['config']['group_claim_key']], list):
+        user_roles = user_info[permissions_rules['config']['group_claim_key']]
+    else:
+        user_roles = [user_info[permissions_rules['config']['group_claim_key']]]
+
+    # find the relevant roles and add permissions 
+    relevant_roles = list(set(permissions_rules["roles"]) & set(user_roles))
+    for relevant_role in relevant_roles:
+        permissions_of_user.extend(permissions_rules["roles"][relevant_role]["permissions"])
+
+    return permissions_of_user
+
+
+def check_if_api_call_is_permitted(request: HttpRequest, permissions_of_user):
+    '''Checks if the user has permission to execute the API call'''
+    for permission in permissions_of_user:
+        allowed_methods = permission['methods']
+        allowed_endpoints = permission['endpoints']
+
+        # check if allowed based on the method
+        if "*" not in allowed_methods and request.method not in allowed_methods:
+            continue
+
+        # prepare the uri
+        prefix = "/api/{}".format(__api_version__)
+        short_uri = request.uri.strip().removeprefix(prefix).split('?', 1)[0]
+
+        # check if you're permitted to make api call based on uri
+        if "*" in allowed_endpoints or short_uri in allowed_endpoints:
+            return True
+
+         # added the glob patterns so it's easier to add a bunch of api calls (like all /device api calls)
+        for allowed_api_call in allowed_endpoints:
+            matches = fnmatch.filter([short_uri], allowed_api_call)
+            if len(matches) > 0:
+                return True
+
+    return False