casbin_enforcer.py

"""flask-casbin: Flask module for using Casbin with flask apps
"""
import casbin
from flask import request, jsonify
from functools import wraps
from abc import ABC
from abc import abstractmethod
import shlex

from flask_authz.utils import authorization_decoder, UnSupportedAuthType


class CasbinEnforcer:
    """
    Casbin Enforce decorator
    """

    e = None

    def __init__(self, app=None, adapter=None, watcher=None):
        """
        Args:
            app (object): Flask App object to get Casbin Model
            adapter (object): Casbin Adapter
        """
        self.app = app
        self.adapter = adapter
        self.e = None
        self.watcher = watcher
        self._owner_loader = None
        self.user_name_headers = None
        if self.app is not None:
            self.init_app(self.app)

    def init_app(self, app):
        self.app = app
        self.e = casbin.Enforcer(app.config.get("CASBIN_MODEL"), self.adapter)
        if self.watcher:
            self.e.set_watcher(self.watcher)
        self.user_name_headers = app.config.get("CASBIN_USER_NAME_HEADERS", None)

    def set_watcher(self, watcher):
        """
        Set the watcher to use with the underlying casbin enforcer
        Args:
            watcher (object):
        Returns:
            None
        """
        self.e.set_watcher(watcher)

    def owner_loader(self, callback):
        """
        This sets the callback for get owner. The
        function return a owner object, or ``None``

        :param callback: The callback for retrieving a owner object.
        :type callback: callable
        """
        self._owner_loader = callback
        return callback

    def enforcer(self, func, delimiter=","):
        @wraps(func)
        def wrapper(*args, **kwargs):
            if self.e.watcher and self.e.watcher.should_reload():
                self.e.watcher.update_callback()
            # String used to hold the owners user name for audit logging
            owner_audit = ""

            # Check sub, obj act against Casbin polices
            self.app.logger.debug(
                "Enforce Headers Config: %s\nRequest Headers: %s"
                % (self.app.config.get("CASBIN_OWNER_HEADERS"), request.headers)
            )
            # Set resource URI from request
            uri = str(request.path)
            # Get owner from owner_loader
            if self._owner_loader:
                self.app.logger.info("Get owner from owner_loader")
                for owner in self._owner_loader():
                    owner = owner.strip('"') if isinstance(owner, str) else owner
                    if self.e.enforce(owner, uri, request.method):
                        return func(*args, **kwargs)
            for header in map(str.lower, self.app.config.get("CASBIN_OWNER_HEADERS")):
                if header in request.headers:
                    # Make Authorization Header Parser standard
                    if header == "authorization":
                        # Get Auth Value then decode and parse for owner
                        try:
                            owner = authorization_decoder(
                                self.app.config, request.headers.get(header)
                            )
                        except UnSupportedAuthType:
                            # Continue if catch unsupported type in the event of
                            # Other headers needing to be checked
                            self.app.logger.info(
                                "Authorization header type requested for "
                                "decoding is unsupported by flask-casbin at this time"
                            )
                            continue
                        except Exception as e:
                            self.app.logger.info(e)
                            continue

                        if self.user_name_headers and header in map(
                            str.lower, self.user_name_headers
                        ):
                            owner_audit = owner
                        if self.e.enforce(owner, uri, request.method):
                            self.app.logger.info(
                                "access granted: method: %s resource: %s%s"
                                % (
                                    request.method,
                                    uri,
                                    ""
                                    if not self.user_name_headers and owner_audit != ""
                                    else " to user: %s" % owner_audit,
                                )
                            )
                            return func(*args, **kwargs)
                    else:
                        # Split header by ',' in case of groups when groups are
                        # sent "group1,group2,group3,..." in the header
                        for owner in self.sanitize_group_headers(
                            request.headers.get(header), delimiter
                        ):
                            self.app.logger.debug(
                                "Enforce against owner: %s header: %s"
                                % (owner.strip('"'), header)
                            )
                            if self.user_name_headers and header in map(
                                str.lower, self.user_name_headers
                            ):
                                owner_audit = owner
                            if self.e.enforce(owner.strip('"'), uri, request.method):
                                self.app.logger.info(
                                    "access granted: method: %s resource: %s%s"
                                    % (
                                        request.method,
                                        uri,
                                        ""
                                        if not self.user_name_headers
                                        and owner_audit != ""
                                        else " to user: %s" % owner_audit,
                                    )
                                )
                                return func(*args, **kwargs)
            else:
                self.app.logger.error(
                    "Unauthorized attempt: method: %s resource: %s%s"
                    % (
                        request.method,
                        uri,
                        ""
                        if not self.user_name_headers and owner_audit != ""
                        else " by user: %s" % owner_audit,
                    )
                )
                return (jsonify({"message": "Unauthorized"}), 401)

        return wrapper

    @staticmethod
    def sanitize_group_headers(headers_str, delimiter=",") -> list:
        """
        Sanitizes group header string so that it is easily parsable by enforcer
        removes extra spaces, and converts comma delimited or white space
        delimited list into a list.

        Default delimiter: "," (comma)

        Returns:
            list
        """
        if delimiter == " " and (
            (headers_str.startswith("'") and headers_str.endswith("'"))
            or (headers_str.startswith('"') and headers_str.endswith('"'))
        ):
            return [
                string.strip() for string in shlex.split(headers_str) if string != ""
            ]
        return [
            string.strip() for string in headers_str.split(delimiter) if string != ""
        ]

    def manager(self, func):
        """Get the Casbin Enforcer Object to manager Casbin"""

        @wraps(func)
        def wrapper(*args, **kwargs):
            return func(self.e, *args, **kwargs)

        return wrapper


class Watcher(ABC):
    """
    Watcher interface as it should be implemented for flask-casbin
    """

    @abstractmethod
    def update(self):
        """
        Watcher interface as it should be implemented for flask-casbin
        Returns:
            None
        """
        pass

    @abstractmethod
    def set_update_callback(self):
        """
        Set the update callback to be used when an update is detected
        Returns:
            None
        """
        pass

    @abstractmethod
    def should_reload(self):
        """
        Method which checks if there is an update necessary for the casbin
        roles. This is called with each flask request.
        Returns:
            Bool
        """
        pass