<commit_before>import logging, json

from flask import current_app, redirect, Response

from itsdangerous import (
    TimedJSONWebSignatureSerializer as Serializer,
    SignatureExpired,
    BadSignature,
)

import sqlalchemy as sa
from sqlalchemy.sql.expression import func


from pypnusershub.db.tools import (
    InsufficientRightsError,
    AccessRightsExpiredError,
    UnreadableAccessRightsError,
)

from geonature.core.gn_permissions.models import VUsersPermissions, TFilters
from geonature.utils.env import DB

log = logging.getLogger(__name__)


def user_from_token(token, secret_key=None):
    secret_key = secret_key or current_app.config["SECRET_KEY"]

    try:
        s = Serializer(current_app.config["SECRET_KEY"])
        user = s.loads(token)
        return user

    except SignatureExpired:
        raise AccessRightsExpiredError("Token expired")

    except BadSignature:
        raise UnreadableAccessRightsError("Token BadSignature", 403)


def get_user_from_token_and_raise(
    request, secret_key=None, redirect_on_expiration=None, redirect_on_invalid_token=None,
):
    """
    Deserialize the token
    catch excetpion and return appropriate Response(403, 302 ...)
    """
    try:
        token = request.cookies["token"]
        return user_from_token(token, secret_key)

    except AccessRightsExpiredError:
        if redirect_on_expiration:
            res = redirect(redirect_on_expiration, code=302)
        else:
            res = Response("Token Expired", 403)
        res.set_cookie("token", expires=0)
        return res
    except InsufficientRightsError as e:
        log.info(e)
        if redirect_on_expiration:
            res = redirect(redirect_on_expiration, code=302)
        else:
            res = Response("Forbidden", 403)
        return res
    except KeyError as e:
        if redirect_on_expiration:
            return redirect(redirect_on_expiration, code=302)
        return Response("No token", 403)

    except UnreadableAccessRightsError:
        log.info("Invalid Token : BadSignature")
        # invalid token
        if redirect_on_invalid_token:
            res = redirect(redirect_on_invalid_token, code=302)
        else:
            res = Response("Token BadSignature", 403)
        res.set_cookie("token", expires=0)
        return res

    except Exception as e:
        trap_all_exceptions = current_app.config.get("TRAP_ALL_EXCEPTIONS", True)
        if not trap_all_exceptions:
            raise
        log.critical(e)
        msg = json.dumps({"type": "Exception", "msg": repr(e)})
        return Response(msg, 403)


class UserCruved:
    def __init__(self,):
        self.is_herited = False

    def build_herited_user_cruved(self, user_permissions, module_code, object_code):
        """
        Parameters:
            - user_permissions(list<VUsersPermissions>)
            - module_code(str)
            - object_code(str)
        Return:
            VUsersPermissions

        """
        # loop on user permissions
        # return the module permission if exist
        # otherwise return GEONATURE permission
        object_permissions = []
        module_permissions = []
        geonature_permission = []
        # filter the GeoNature perm and the module perm in two
        # arrays to make heritage
        for user_permission in user_permissions:
            if user_permission.code_object == object_code:
                object_permissions.append(user_permission)
            elif user_permission.module_code == module_code:
                module_permissions.append(user_permission)
            else:
                geonature_permission.append(user_permission)

        # take the max of the different permissions
        if len(object_permissions) > 0:
            return get_max_perm(object_permissions)
        elif len(object_permissions) == 0 and len(module_permissions) > 0:
            if object_code:
                self.is_herited = True
            return get_max_perm(module_permissions)
        # if no module permission take the max of GN perm
        elif len(module_permissions) == 0:
            if module_code != "GEONATURE":
                self.is_herited = True
            return get_max_perm(geonature_permission)


def query_user_perm(
    id_role, code_filter_type, code_action=None, module_code=None, object_code=None
):

    ors = [VUsersPermissions.module_code.ilike("GEONATURE")]

    q = VUsersPermissions.query.filter(VUsersPermissions.id_role == id_role).filter(
        VUsersPermissions.code_filter_type == code_filter_type
    )
    if code_action:
        q = q.filter(VUsersPermissions.code_action == code_action)
    if module_code:
        ors.append(VUsersPermissions.module_code.ilike(module_code))
    if object_code:
        ors.append(VUsersPermissions.code_object == object_code)
    # if object code is None, only take ALL
    else:
        q = q.filter(VUsersPermissions.code_object == "ALL")
    return q.filter(sa.or_(*ors)).all()


def get_user_permissions(
    user, code_filter_type, code_action=None, module_code=None, code_object=None
):
    """
        Get all the permissions of a user for an action, a module (or an object) and a filter_type
        Users permissions could be multiples because of user's group. The view mapped by VUsersPermissions does not take the
        max because some filter type could be not quantitative
        
        Parameters:
            user(dict)
            code_filter_type(str): <SCOPE, GEOGRAPHIC ...>
            code_action(str): <C,R,U,V,E,D> or None if all actions wanted
            module_code(str): 'GEONATURE', 'OCCTAX'
            code_object(str): 'PERMISSIONS', 'DATASET' (table gn_permissions.t_oject)
        Return:
            Array<VUsersPermissions>
    """
    user_cruved = query_user_perm(
        id_role=user["id_role"],
        code_filter_type=code_filter_type,
        code_action=code_action,
        object_code=code_object,<commit_msg>fix bug on cruved - fix #1146<commit_after>import logging, json

from flask import current_app, redirect, Response

from itsdangerous import (
    TimedJSONWebSignatureSerializer as Serializer,
    SignatureExpired,
    BadSignature,
)

import sqlalchemy as sa
from sqlalchemy.sql.expression import func


from pypnusershub.db.tools import (
    InsufficientRightsError,
    AccessRightsExpiredError,
    UnreadableAccessRightsError,
)

from geonature.core.gn_permissions.models import VUsersPermissions, TFilters
from geonature.utils.env import DB

log = logging.getLogger(__name__)


def user_from_token(token, secret_key=None):
    secret_key = secret_key or current_app.config["SECRET_KEY"]

    try:
        s = Serializer(current_app.config["SECRET_KEY"])
        user = s.loads(token)
        return user

    except SignatureExpired:
        raise AccessRightsExpiredError("Token expired")

    except BadSignature:
        raise UnreadableAccessRightsError("Token BadSignature", 403)


def get_user_from_token_and_raise(
    request, secret_key=None, redirect_on_expiration=None, redirect_on_invalid_token=None,
):
    """
    Deserialize the token
    catch excetpion and return appropriate Response(403, 302 ...)
    """
    try:
        token = request.cookies["token"]
        return user_from_token(token, secret_key)

    except AccessRightsExpiredError:
        if redirect_on_expiration:
            res = redirect(redirect_on_expiration, code=302)
        else:
            res = Response("Token Expired", 403)
        res.set_cookie("token", expires=0)
        return res
    except InsufficientRightsError as e:
        log.info(e)
        if redirect_on_expiration:
            res = redirect(redirect_on_expiration, code=302)
        else:
            res = Response("Forbidden", 403)
        return res
    except KeyError as e:
        if redirect_on_expiration:
            return redirect(redirect_on_expiration, code=302)
        return Response("No token", 403)

    except UnreadableAccessRightsError:
        log.info("Invalid Token : BadSignature")
        # invalid token
        if redirect_on_invalid_token:
            res = redirect(redirect_on_invalid_token, code=302)
        else:
            res = Response("Token BadSignature", 403)
        res.set_cookie("token", expires=0)
        return res

    except Exception as e:
        trap_all_exceptions = current_app.config.get("TRAP_ALL_EXCEPTIONS", True)
        if not trap_all_exceptions:
            raise
        log.critical(e)
        msg = json.dumps({"type": "Exception", "msg": repr(e)})
        return Response(msg, 403)


class UserCruved:
    def __init__(self,):
        self.is_herited = False

    def build_herited_user_cruved(self, user_permissions, module_code, object_code):
        """
        Parameters:
            - user_permissions(list<VUsersPermissions>)
            - module_code(str)
            - object_code(str)
        Return:
            VUsersPermissions

        """
        # loop on user permissions
        # return the module permission if exist
        # otherwise return GEONATURE permission
        object_permissions = []
        module_permissions = []
        geonature_permission = []
        # filter the GeoNature perm and the module perm in two
        # arrays to make heritage
        for user_permission in user_permissions:
            if user_permission.code_object == object_code:
                object_permissions.append(user_permission)
            elif user_permission.module_code == module_code:
                module_permissions.append(user_permission)
            else:
                geonature_permission.append(user_permission)

        # take the max of the different permissions
        if len(object_permissions) > 0:
            return get_max_perm(object_permissions)
        elif len(object_permissions) == 0 and len(module_permissions) > 0:
            if object_code:
                self.is_herited = True
            return get_max_perm(module_permissions)
        # if no module permission take the max of GN perm
        elif len(module_permissions) == 0:
            if module_code != "GEONATURE":
                self.is_herited = True
            return get_max_perm(geonature_permission)


def query_user_perm(
    id_role, code_filter_type, code_action=None, module_code=None, object_code=None
):

    ors = [VUsersPermissions.module_code.ilike("GEONATURE")]

    q = VUsersPermissions.query.filter(VUsersPermissions.id_role == id_role).filter(
        VUsersPermissions.code_filter_type == code_filter_type
    )
    if code_action:
        q = q.filter(VUsersPermissions.code_action == code_action)
    if module_code:
        ors.append(VUsersPermissions.module_code.ilike(module_code))
    if object_code:
        ors.append(VUsersPermissions.code_object == object_code)
    # if object code is None, only take ALL
    else:
        q = q.filter(VUsersPermissions.code_object == "ALL")
    return q.filter(sa.or_(*ors)).all()


def get_user_permissions(
    user, code_filter_type, code_action=None, module_code=None, code_object=None
):
    """
        Get all the permissions of a user for an action, a module (or an object) and a filter_type
        Users permissions could be multiples because of user's group. The view mapped by VUsersPermissions does not take the
        max because some filter type could be not quantitative
        
        Parameters:
            user(dict)
            code_filter_type(str): <SCOPE, GEOGRAPHIC ...>
            code_action(str): <C,R,U,V,E,D> or None if all actions wanted
            module_code(str): 'GEONATURE', 'OCCTAX'
            code_object(str): 'PERMISSIONS', 'DATASET' (table gn_permissions.t_oject)
        Return:
            Array<VUsersPermissions>
    """
    user_cruved = query_user_perm(
        id_role=user["id_role"],
        code_filter_type=code_filter_type,
        code_action=code_action,
        module_code=module_code,
        object_code=code_object,
{'fraction_non_alphanumeric': 0.07791990636234429, 'fraction_numerical': 0.007357244377560405, 'mean_word_length': 2.7322932917316693, 'pattern_counts': {'":': 6, '<': 11, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5775426', 'n_tokens_mistral': 3558, 'n_tokens_neox': 3407, 'n_words': 809}