<commit_before># -*- coding: utf-8 -*-
#
#  2016-04-08 <NAME> <<EMAIL>>
#             Avoid consecutive if statements
#
#  privacyIDEA
#  May 11, 2014 <NAME>, <EMAIL>
#  http://www.privacyidea.org
#
# This code is free software; you can redistribute it and/or
# modify it under the terms of the GNU AFFERO GENERAL PUBLIC LICENSE
# License as published by the Free Software Foundation; either
# version 3 of the License, or any later version.
#
# This code is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU AFFERO GENERAL PUBLIC LICENSE for more details.
#
# You should have received a copy of the GNU Affero General Public
# License along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
#
__doc__ = """The SQL Audit Module is used to write audit entries to an SQL
database.
The SQL Audit Module is configured like this:

    PI_AUDIT_MODULE = "privacyidea.lib.auditmodules.sqlaudit"
    PI_AUDIT_KEY_PRIVATE = "tests/testdata/private.pem"
    PI_AUDIT_KEY_PUBLIC = "tests/testdata/public.pem"
    PI_AUDIT_SERVERNAME = "your choice"

    Optional:
    PI_AUDIT_SQL_URI = "sqlite://"
    PI_AUDIT_SQL_TRUNCATE = True | False
    PI_AUDIT_SQL_COLUMN_LENGTH = {"user": 60, "info": 10 ...}

If the PI_AUDIT_SQL_URI is omitted the Audit data is written to the
token database.
"""

import logging
from collections import OrderedDict
from privacyidea.lib.auditmodules.base import (Audit as AuditBase, Paginate)
from privacyidea.lib.crypto import Sign
from privacyidea.lib.pooling import get_engine
from privacyidea.lib.utils import censor_connect_string
from privacyidea.lib.lifecycle import register_finalizer
from privacyidea.lib.utils import truncate_comma_list, is_true
from sqlalchemy import MetaData, cast, String
from sqlalchemy import asc, desc, and_, or_
import datetime
import traceback
from six import string_types
from privacyidea.models import audit_column_length as column_length
from privacyidea.models import Audit as LogEntry
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session

log = logging.getLogger(__name__)

metadata = MetaData()


class Audit(AuditBase):
    """
    This is the SQLAudit module, which writes the audit entries
    to an SQL database table.
    It requires the configuration parameters in pi.cfg:
    * PI_AUDIT_KEY_PUBLIC
    * PI_AUDIT_KEY_PRIVATE

    If you want to host the SQL Audit database in another DB than the
    token DB, you can use:
    * PI_AUDIT_SQL_URI

    It also takes the optional parameters:
    * PI_AUDIT_POOL_SIZE
    * PI_AUDIT_POOL_RECYCLE
    * PI_AUDIT_SQL_TRUNCATE
    * PI_AUDIT_NO_SIGN

    You can use PI_AUDIT_NO_SIGN = True to avoid signing of the audit log.

    If PI_CHECK_OLD_SIGNATURES = True old style signatures (text-book RSA) will
    be checked as well, otherwise they will be marked as 'FAIL'.
    """

    is_readable = True

    def __init__(self, config=None, startdate=None):
        super(Audit, self).__init__(config, startdate)
        self.name = "sqlaudit"
        self.sign_data = not self.config.get("PI_AUDIT_NO_SIGN")
        self.sign_object = None
        self.verify_old_sig = self.config.get('PI_CHECK_OLD_SIGNATURES')
        if self.sign_data:
            self.read_keys(self.config.get("PI_AUDIT_KEY_PUBLIC"),
                           self.config.get("PI_AUDIT_KEY_PRIVATE"))
            self.sign_object = Sign(self.private, self.public)
        # Read column_length from the config file
        config_column_length = self.config.get("PI_AUDIT_SQL_COLUMN_LENGTH", {})
        # fill the missing parts with the default from the models
        self.custom_column_length = {k: (v if k not in config_column_length else config_column_length[k])
                                     for k, v in column_length.items()}
        # We can use "sqlaudit" as the key because the SQLAudit connection
        # string is fixed for a running privacyIDEA instance.
        # In other words, we will not run into any problems with changing connect strings.
        self.engine = get_engine(self.name, self._create_engine)
        # create a configured "Session" class. ``scoped_session`` is not
        # necessary because we do not share session objects among threads.
        # We use it anyway as a safety measure.
        Session = scoped_session(sessionmaker(bind=self.engine))
        self.session = Session()
        # Ensure that the connection gets returned to the pool when the request has
        # been handled. This may close an already-closed session, but this is not a problem.
        register_finalizer(self._finalize_session)
        self.session._model_changes = {}

    def _create_engine(self):
        """
        :return: a new SQLAlchemy engine connecting to the database specified in PI_AUDIT_SQL_URI.
        """
        # an Engine, which the Session will use for connection
        # resources
        connect_string = self.config.get("PI_AUDIT_SQL_URI", self.config.get(
            "SQLALCHEMY_DATABASE_URI"))
        log.debug("using the connect string {0!s}".format(censor_connect_string(connect_string)))
        try:
            pool_size = self.config.get("PI_AUDIT_POOL_SIZE", 20)
            engine = create_engine(
                connect_string,
                pool_size=pool_size,
                pool_recycle=self.config.get("PI_AUDIT_POOL_RECYCLE", 600))
            log.debug("Using SQL pool size of {}".format(pool_size))
        except TypeError:
            # SQLite does not support pool_size
            engine = create_engine(connect_string)
            log.debug("Using no SQL pool_size.")
        return engine

    def _finalize_session(self):
        """ Close current session and dispose connections of db engine"""
        self.session.close()
        self.engine.dispose()

    def _truncate_data(self):
        """
        Truncate self.audit_data according to the self.custom_column_length.
        :return: None
        """
        for column, l in self.custom_column_length.items():
            if column in self.audit_data:
                data = self.audit_data[column]
                if isinstance(data, string_types):
                    if column == "policies":
                        # The policies column is shortend per comma entry
                        data = truncate_comma_list(data, l)
                    else:
                        data = data[:l]
                self.audit_data[column] = data

    @staticmethod
    def _create_filter(param, timelimit=None):
        """
        create a filter condition for the logentry
        """
        conditions = []
        param = param or {}
        for search_key in param.keys():
            search_value = param.get(search_key)
            if search_key == "allowed_audit_realm":
                # Add each realm in the allowed_audit_realm list to the
                # search condition
                realm_conditions = []
                for realm in search_value:
                    realm_conditions.append(LogEntry.realm == realm)
                filter_realm = or_(*realm_conditions)
                conditions.append(filter_realm)
            # We do not search if the search value only consists of '*'
            elif search_value.strip() != '' and search_value.strip('*') != '':
                try:
                    if search_key == "success":
                        # "success" is the only integer.
                        search_value = search_value.strip("*")
                        conditions.append(getattr(LogEntry, search_key) ==
                                          int(is_true(search_value)))
                    else:
                        # All other keys are compared as strings
                        column = getattr(LogEntry, search_key)
                        if search_key == "date":
                            # but we cast "date" to a string first (required on postgresql)
                            column = cast(column, String)
                        search_value = search_value.replace('*', '%')
                        if '%' in search_value:
                            conditions.append(column.like(search_value))
                        else:
                            conditions.append(column == search_value)
                except Exception as exx:
                    # The search_key was no search key but some
                    # bullshit stuff in the param
                    log.debug("Not a valid searchkey: {0!s}".format(exx))

        if timelimit:
            conditions.append(LogEntry.date >= datetime.datetime.now() -
                              timelimit)
        # Combine them with or to a BooleanClauseList
        filter_condition = and_(*conditions)
        return filter_condition

    def get_total(self, param, AND=True, display_error=True, timelimit=None):
        """
        This method returns the total number of audit entries
        in the audit store
        """
        count = 0
        # if param contains search filters, we build the search filter
        # to only return the number of those entries
        filter_condition = self._create_filter(param, timelimit=timelimit)

        try:
            count = self.session.query(LogEntry.id) \<commit_msg>Fix search in audit date strings<commit_after># -*- coding: utf-8 -*-
#
#  2016-04-08 <NAME> <<EMAIL>>
#             Avoid consecutive if statements
#
#  privacyIDEA
#  May 11, 2014 <NAME>, <EMAIL>
#  http://www.privacyidea.org
#
# This code is free software; you can redistribute it and/or
# modify it under the terms of the GNU AFFERO GENERAL PUBLIC LICENSE
# License as published by the Free Software Foundation; either
# version 3 of the License, or any later version.
#
# This code is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU AFFERO GENERAL PUBLIC LICENSE for more details.
#
# You should have received a copy of the GNU Affero General Public
# License along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
#
__doc__ = """The SQL Audit Module is used to write audit entries to an SQL
database.
The SQL Audit Module is configured like this:

    PI_AUDIT_MODULE = "privacyidea.lib.auditmodules.sqlaudit"
    PI_AUDIT_KEY_PRIVATE = "tests/testdata/private.pem"
    PI_AUDIT_KEY_PUBLIC = "tests/testdata/public.pem"
    PI_AUDIT_SERVERNAME = "your choice"

    Optional:
    PI_AUDIT_SQL_URI = "sqlite://"
    PI_AUDIT_SQL_TRUNCATE = True | False
    PI_AUDIT_SQL_COLUMN_LENGTH = {"user": 60, "info": 10 ...}

If the PI_AUDIT_SQL_URI is omitted the Audit data is written to the
token database.
"""

import logging
from collections import OrderedDict
from privacyidea.lib.auditmodules.base import (Audit as AuditBase, Paginate)
from privacyidea.lib.crypto import Sign
from privacyidea.lib.pooling import get_engine
from privacyidea.lib.utils import censor_connect_string
from privacyidea.lib.lifecycle import register_finalizer
from privacyidea.lib.utils import truncate_comma_list, is_true
from sqlalchemy import MetaData, cast, String
from sqlalchemy import asc, desc, and_, or_
from sqlalchemy.sql import expression
from sqlalchemy.ext.compiler import compiles
import datetime
import traceback
from six import string_types
from privacyidea.models import audit_column_length as column_length
from privacyidea.models import Audit as LogEntry
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session

log = logging.getLogger(__name__)

metadata = MetaData()


# Define function to convert SQL DateTime objects to an ISO-format string
# By using <https://docs.sqlalchemy.org/en/13/core/compiler.html> we can
# differentiate between different dialects.
class to_isodate(expression.FunctionElement):
    name = 'to_isodate'


@compiles(to_isodate, 'oracle')
@compiles(to_isodate, 'postgresql')
def fn_to_isodate(element, compiler, **kw):
    return "to_char(%s, 'IYYY-MM-DD HH24:MI:SS')" % compiler.process(element.clauses, **kw)


@compiles(to_isodate)
def fn_to_isodate(element, compiler, **kw):
    # The four percent signs are necessary for two format substitutions
    return "date_format(%s, '%%%%Y-%%%%m-%%%%d %%%%H:%%%%i:%%%%s')" % compiler.process(
        element.clauses, **kw)


class Audit(AuditBase):
    """
    This is the SQLAudit module, which writes the audit entries
    to an SQL database table.
    It requires the configuration parameters in pi.cfg:
    * PI_AUDIT_KEY_PUBLIC
    * PI_AUDIT_KEY_PRIVATE

    If you want to host the SQL Audit database in another DB than the
    token DB, you can use:
    * PI_AUDIT_SQL_URI

    It also takes the optional parameters:
    * PI_AUDIT_POOL_SIZE
    * PI_AUDIT_POOL_RECYCLE
    * PI_AUDIT_SQL_TRUNCATE
    * PI_AUDIT_NO_SIGN

    You can use PI_AUDIT_NO_SIGN = True to avoid signing of the audit log.

    If PI_CHECK_OLD_SIGNATURES = True old style signatures (text-book RSA) will
    be checked as well, otherwise they will be marked as 'FAIL'.
    """

    is_readable = True

    def __init__(self, config=None, startdate=None):
        super(Audit, self).__init__(config, startdate)
        self.name = "sqlaudit"
        self.sign_data = not self.config.get("PI_AUDIT_NO_SIGN")
        self.sign_object = None
        self.verify_old_sig = self.config.get('PI_CHECK_OLD_SIGNATURES')
        if self.sign_data:
            self.read_keys(self.config.get("PI_AUDIT_KEY_PUBLIC"),
                           self.config.get("PI_AUDIT_KEY_PRIVATE"))
            self.sign_object = Sign(self.private, self.public)
        # Read column_length from the config file
        config_column_length = self.config.get("PI_AUDIT_SQL_COLUMN_LENGTH", {})
        # fill the missing parts with the default from the models
        self.custom_column_length = {k: (v if k not in config_column_length else config_column_length[k])
                                     for k, v in column_length.items()}
        # We can use "sqlaudit" as the key because the SQLAudit connection
        # string is fixed for a running privacyIDEA instance.
        # In other words, we will not run into any problems with changing connect strings.
        self.engine = get_engine(self.name, self._create_engine)
        # create a configured "Session" class. ``scoped_session`` is not
        # necessary because we do not share session objects among threads.
        # We use it anyway as a safety measure.
        Session = scoped_session(sessionmaker(bind=self.engine))
        self.session = Session()
        # Ensure that the connection gets returned to the pool when the request has
        # been handled. This may close an already-closed session, but this is not a problem.
        register_finalizer(self._finalize_session)
        self.session._model_changes = {}

    def _create_engine(self):
        """
        :return: a new SQLAlchemy engine connecting to the database specified in PI_AUDIT_SQL_URI.
        """
        # an Engine, which the Session will use for connection
        # resources
        connect_string = self.config.get("PI_AUDIT_SQL_URI", self.config.get(
            "SQLALCHEMY_DATABASE_URI"))
        log.debug("using the connect string {0!s}".format(censor_connect_string(connect_string)))
        try:
            pool_size = self.config.get("PI_AUDIT_POOL_SIZE", 20)
            engine = create_engine(
                connect_string,
                pool_size=pool_size,
                pool_recycle=self.config.get("PI_AUDIT_POOL_RECYCLE", 600))
            log.debug("Using SQL pool size of {}".format(pool_size))
        except TypeError:
            # SQLite does not support pool_size
            engine = create_engine(connect_string)
            log.debug("Using no SQL pool_size.")
        return engine

    def _finalize_session(self):
        """ Close current session and dispose connections of db engine"""
        self.session.close()
        self.engine.dispose()

    def _truncate_data(self):
        """
        Truncate self.audit_data according to the self.custom_column_length.
        :return: None
        """
        for column, l in self.custom_column_length.items():
            if column in self.audit_data:
                data = self.audit_data[column]
                if isinstance(data, string_types):
                    if column == "policies":
                        # The policies column is shortend per comma entry
                        data = truncate_comma_list(data, l)
                    else:
                        data = data[:l]
                self.audit_data[column] = data

    @staticmethod
    def _create_filter(param, timelimit=None):
        """
        create a filter condition for the logentry
        """
        conditions = []
        param = param or {}
        for search_key in param.keys():
            search_value = param.get(search_key)
            if search_key == "allowed_audit_realm":
                # Add each realm in the allowed_audit_realm list to the
                # search condition
                realm_conditions = []
                for realm in search_value:
                    realm_conditions.append(LogEntry.realm == realm)
                filter_realm = or_(*realm_conditions)
                conditions.append(filter_realm)
            # We do not search if the search value only consists of '*'
            elif search_value.strip() != '' and search_value.strip('*') != '':
                try:
                    if search_key == "success":
                        # "success" is the only integer.
                        search_value = search_value.strip("*")
                        conditions.append(getattr(LogEntry, search_key) ==
                                          int(is_true(search_value)))
                    else:
                        # All other keys are compared as strings
                        column = getattr(LogEntry, search_key)
                        if search_key.endswith("date"):
                            # but we cast "date" to a string first (required on postgresql)
                            column = to_isodate(column)
                        search_value = search_value.replace('*', '%')
                        if '%' in search_value:
                            conditions.append(column.like(search_value))
                        else:
                            conditions.append(column == search_value)
                except Exception as exx:
                    # The search_key was no search key but some
                    # bullshit stuff in the param
                    log.debug("Not a valid searchkey: {0!s}".format(exx))

        if timelimit:
            conditions.append(LogEntry.date >= datetime.datetime.now() -
                              timelimit)
        # Combine them with or to a BooleanClauseList
        filter_condition = and_(*conditions)
        return filter_condition

    def get_total(self, param, AND=True, display_error=True, timelimit=None):
        """
        This method returns the total number of audit entries
        in the audit store
        """
        count = 0
        # if param contains search filters, we build the search filter
        # to only return the number of those entries
        filter_condition = self._create_filter(param, timelimit=timelimit)

        try:
            count = self.session.query(LogEntry.id) \
{'fraction_non_alphanumeric': 0.0775364170713008, 'fraction_numerical': 0.003066700741119346, 'mean_word_length': 2.3389078498293516, 'pattern_counts': {'":': 11, '<': 16, '<?xml version=': 0, '>': 18, 'https://': 1, 'lorem ipsum': 0, 'www.': 4, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2876215', 'n_tokens_mistral': 5537, 'n_tokens_neox': 5312, 'n_words': 1689}