<commit_before>#!/usr/bin/python
# -*- coding: utf-8 -*-
#
# Copyright (C) 2005 onwards University of Deusto
# All rights reserved.
#
# This software is licensed as described in the file COPYING, which
# you should have received as part of this distribution.
#
# This software consists of contributions made by many individuals,
# listed below:
#
# Author: <NAME> <<EMAIL>>
#         <NAME> <<EMAIL>>
#

from functools import wraps
import numbers

from sqlalchemy.orm.exc import NoResultFound
from sqlalchemy.sql.expression import desc

from voodoo.log import logged
from voodoo.typechecker import typecheck

import weblab.db.model as model

import weblab.db.gateway as dbGateway

from weblab.data.command import Command
import weblab.data.dto.experiments as ExperimentAllowed
from weblab.data.experiments import ExperimentUsage, CommandSent, FileSent

import weblab.db.exc as DbErrors
import weblab.permissions as permissions

def admin_panel_operation(func):
    """It checks if the requesting user has the admin_panel_access permission with full_privileges (temporal policy)."""

    @wraps(func)
    def wrapper(self, user_login, *args, **kargs):
        session = self.Session()
        try:
            user = self._get_user(session, user_login)
            admin_panel_access_permissions = self._gather_permissions(session, user, "admin_panel_access")
            if len(admin_panel_access_permissions) > 0:
                # The only requirement for now is to have full_privileges, this will be changed in further versions
                if self._get_bool_parameter_from_permission(session, admin_panel_access_permissions[0], "full_privileges"):
                    return func(self, user_login, *args, **kargs)
            return ()
        finally:
            session.close()
    return wrapper

DEFAULT_VALUE = object()

class DatabaseGateway(dbGateway.AbstractDatabaseGateway):

    forbidden_access = 'forbidden_access'

    def __init__(self, cfg_manager):
        super(DatabaseGateway, self).__init__(cfg_manager)

    @typecheck(basestring)
    @logged()
    def get_user_by_name(self, user_login):
        session = self.Session()
        try:
            return self._get_user(session, user_login).to_dto()
        finally:
            session.close()

    @logged()
    def list_clients(self):
        """Lists the ExperimentClients """
        session = self.Session()
        try:
            clients = {}
            for experiment in session.query(model.DbExperiment).all():
                exp = experiment.to_business()
                clients[exp.name, exp.category.name] = exp.client
            return clients
        finally:
            session.close()

    # @typecheck(basestring, (basestring, None), (basestring, None))
    @logged()
    def list_experiments(self, user_login, exp_name = None, cat_name = None):
        session = self.Session()
        try:
            user = self._get_user(session, user_login)
            permissions = self._gather_permissions(session, user, 'experiment_allowed')

            grouped_experiments = {}
            for permission in permissions:
                p_permanent_id                 = self._get_parameter_from_permission(session, permission, 'experiment_permanent_id')
                p_category_id                  = self._get_parameter_from_permission(session, permission, 'experiment_category_id')
                p_time_allowed                 = self._get_float_parameter_from_permission(session, permission, 'time_allowed')
                p_priority                     = self._get_int_parameter_from_permission(session, permission, 'priority', ExperimentAllowed.DEFAULT_PRIORITY)
                p_initialization_in_accounting = self._get_bool_parameter_from_permission(session, permission, 'initialization_in_accounting', ExperimentAllowed.DEFAULT_INITIALIZATION_IN_ACCOUNTING)
                
                # If a filter is passed, ignore those permissions on other experiments
                if cat_name is not None and exp_name is not None:
                    if p_category_id != cat_name or p_permanent_id != exp_name:
                        continue

                experiment = session.query(model.DbExperiment).filter_by(name=p_permanent_id).filter(model.DbExperimentCategory.name==p_category_id).first()
                if experiment is None:
                    continue

                if isinstance(permission, model.DbUserPermission):
                    permission_scope = 'user'
                elif isinstance(permission, model.DbGroupPermission):
                    permission_scope = 'group'
                elif isinstance(permission, model.DbRolePermission):
                    permission_scope = 'role'
                else:
                    permission_scope = 'unknown'<commit_msg>Fix bug: not checking the category<commit_after>#!/usr/bin/python
# -*- coding: utf-8 -*-
#
# Copyright (C) 2005 onwards University of Deusto
# All rights reserved.
#
# This software is licensed as described in the file COPYING, which
# you should have received as part of this distribution.
#
# This software consists of contributions made by many individuals,
# listed below:
#
# Author: <NAME> <<EMAIL>>
#         <NAME> <<EMAIL>>
#

from functools import wraps
import numbers

from sqlalchemy.orm.exc import NoResultFound
from sqlalchemy.sql.expression import desc

from voodoo.log import logged
from voodoo.typechecker import typecheck

import weblab.db.model as model

import weblab.db.gateway as dbGateway

from weblab.data.command import Command
import weblab.data.dto.experiments as ExperimentAllowed
from weblab.data.experiments import ExperimentUsage, CommandSent, FileSent

import weblab.db.exc as DbErrors
import weblab.permissions as permissions

def admin_panel_operation(func):
    """It checks if the requesting user has the admin_panel_access permission with full_privileges (temporal policy)."""

    @wraps(func)
    def wrapper(self, user_login, *args, **kargs):
        session = self.Session()
        try:
            user = self._get_user(session, user_login)
            admin_panel_access_permissions = self._gather_permissions(session, user, "admin_panel_access")
            if len(admin_panel_access_permissions) > 0:
                # The only requirement for now is to have full_privileges, this will be changed in further versions
                if self._get_bool_parameter_from_permission(session, admin_panel_access_permissions[0], "full_privileges"):
                    return func(self, user_login, *args, **kargs)
            return ()
        finally:
            session.close()
    return wrapper

DEFAULT_VALUE = object()

class DatabaseGateway(dbGateway.AbstractDatabaseGateway):

    forbidden_access = 'forbidden_access'

    def __init__(self, cfg_manager):
        super(DatabaseGateway, self).__init__(cfg_manager)

    @typecheck(basestring)
    @logged()
    def get_user_by_name(self, user_login):
        session = self.Session()
        try:
            return self._get_user(session, user_login).to_dto()
        finally:
            session.close()

    @logged()
    def list_clients(self):
        """Lists the ExperimentClients """
        session = self.Session()
        try:
            clients = {}
            for experiment in session.query(model.DbExperiment).all():
                exp = experiment.to_business()
                clients[exp.name, exp.category.name] = exp.client
            return clients
        finally:
            session.close()

    # @typecheck(basestring, (basestring, None), (basestring, None))
    @logged()
    def list_experiments(self, user_login, exp_name = None, cat_name = None):
        session = self.Session()
        try:
            user = self._get_user(session, user_login)
            permissions = self._gather_permissions(session, user, 'experiment_allowed')

            grouped_experiments = {}
            for permission in permissions:
                p_permanent_id                 = self._get_parameter_from_permission(session, permission, 'experiment_permanent_id')
                p_category_id                  = self._get_parameter_from_permission(session, permission, 'experiment_category_id')
                p_time_allowed                 = self._get_float_parameter_from_permission(session, permission, 'time_allowed')
                p_priority                     = self._get_int_parameter_from_permission(session, permission, 'priority', ExperimentAllowed.DEFAULT_PRIORITY)
                p_initialization_in_accounting = self._get_bool_parameter_from_permission(session, permission, 'initialization_in_accounting', ExperimentAllowed.DEFAULT_INITIALIZATION_IN_ACCOUNTING)
                
                # If a filter is passed, ignore those permissions on other experiments
                if cat_name is not None and exp_name is not None:
                    if p_category_id != cat_name or p_permanent_id != exp_name:
                        continue
                experiments = [ exp for exp in session.query(model.DbExperiment).filter_by(name=p_permanent_id).all() if exp.category.name == p_category_id ]
                if len(experiments) == 0:
                    continue

                experiment = experiments[0]

                if isinstance(permission, model.DbUserPermission):
                    permission_scope = 'user'
                elif isinstance(permission, model.DbGroupPermission):
                    permission_scope = 'group'
                elif isinstance(permission, model.DbRolePermission):
                    permission_scope = 'role'
                else:
                    permission_scope = 'unknown'
{'fraction_non_alphanumeric': 0.07903709060853394, 'fraction_numerical': 0.0016530633329889452, 'mean_word_length': 2.5985130111524164, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 17, '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': '4200362', 'n_tokens_mistral': 2630, 'n_tokens_neox': 2558, 'n_words': 644}