<commit_before># -*- coding: utf-8 -*-
#
#  Copyright (C) 2014 <NAME>
#  License:  AGPLv3
#  contact:  <EMAIL>
#
# 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__ = """This is the resolver to find users in SQL databases.

The file is tested in tests/test_lib_resolver.py
"""

import logging
import yaml

from UserIdResolver import UserIdResolver

from sqlalchemy import and_
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

import traceback
from base64 import (b64decode,
                    b64encode)
import os
import time
import hashlib
import crypt
from privacyidea.lib.crypto import urandom, geturandom
import binascii

log = logging.getLogger(__name__)
ENCODING = "utf-8"

SQLSOUP_LOADED = False
try:
    from sqlsoup import SQLSoup
    SQLSOUP_LOADED = True
except ImportError:  # pragma: no cover
    log.debug("SQLSoup could not be loaded!")

if SQLSOUP_LOADED is False:  # pragma: no cover
    try:
        from sqlalchemy.ext.sqlsoup import SQLSoup
        log.debug("SQLSoup loaded from SQLAlchemy.")
        SQLSOUP_LOADED = True
    except ImportError:
        log.error("SQLSoup could not be loaded from SQLAlchemy!")

try:
    import bcrypt
    _bcrypt_hashpw = bcrypt.hashpw
except ImportError:  # pragma: no cover
    _bcrypt_hashpw = None

# On App Engine, this function is not available.
if hasattr(os, 'getpid'):
    _pid = os.getpid()
else:  # pragma: no cover
    # Fake PID
    _pid = urandom.randint(0, 100000)


class PasswordHash(object):

    def __init__(self, iteration_count_log2=8, portable_hashes=True,
                 algorithm=''):
        alg = algorithm.lower()
        if (alg == 'blowfish' or alg == 'bcrypt') and _bcrypt_hashpw is None:
            raise NotImplementedError('The bcrypt module is required')
        self.itoa64 = \
            './0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz'
        if iteration_count_log2 < 4 or iteration_count_log2 > 31:
            iteration_count_log2 = 8
        self.iteration_count_log2 = iteration_count_log2
        self.portable_hashes = portable_hashes
        self.algorithm = algorithm
        self.random_state = '%r%r' % (time.time(), _pid)

    def get_random_bytes(self, count):
        outp = ''
        try:
            outp = os.urandom(count)
        except Exception as exx:  # pragma: no cover
            log.debug("problem getting os.urandom: %s" % exx)
        if len(outp) < count:  # pragma: no cover
            outp = ''
            rem = count
            while rem > 0:
                self.random_state = hashlib.md5(str(time.time())
                    + self.random_state).hexdigest()
                outp += hashlib.md5(self.random_state).digest()
                rem -= 1
            outp = outp[:count]
        return outp

    def encode64(self, inp, count):
        outp = ''
        cur = 0
        while cur < count:
            value = ord(inp[cur])
            cur += 1
            outp += self.itoa64[value & 0x3f]
            if cur < count:
                value |= (ord(inp[cur]) << 8)
            outp += self.itoa64[(value >> 6) & 0x3f]
            if cur >= count:
                break
            cur += 1
            if cur < count:
                value |= (ord(inp[cur]) << 16)
            outp += self.itoa64[(value >> 12) & 0x3f]
            if cur >= count:
                break
            cur += 1
            outp += self.itoa64[(value >> 18) & 0x3f]
        return outp

    def gensalt_private(self, inp):  # pragma: no cover
        outp = '$P$'
        outp += self.itoa64[min([self.iteration_count_log2 + 5, 30])]
        outp += self.encode64(inp, 6)
        return outp

    def crypt_private(self, pw, setting):  # pragma: no cover
        outp = '*0'
        if setting.startswith(outp):
            outp = '*1'
        if setting[0:3] not in ['$P$', '$H$', '$S$']:
            return outp
        count_log2 = self.itoa64.find(setting[3])
        if count_log2 < 7 or count_log2 > 30:
            return outp
        count = 1 << count_log2
        salt = setting[4:12]
        if len(salt) != 8:
            return outp
        if not isinstance(pw, str):
            pw = pw.encode('utf-8')

        hash_func = hashlib.md5
        encoding_len = 16
        if setting.startswith('$S$'):
            hash_func = hashlib.sha512<commit_msg>Simplify out-of-bounds check<commit_after># -*- coding: utf-8 -*-
#
#  Copyright (C) 2014 <NAME>
#  License:  AGPLv3
#  contact:  <EMAIL>
#
#  2016-04-08 <NAME> <<EMAIL>>
#             Simplifying out of bounds check
#
# 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__ = """This is the resolver to find users in SQL databases.

The file is tested in tests/test_lib_resolver.py
"""

import logging
import yaml

from UserIdResolver import UserIdResolver

from sqlalchemy import and_
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

import traceback
from base64 import (b64decode,
                    b64encode)
import os
import time
import hashlib
import crypt
from privacyidea.lib.crypto import urandom, geturandom
import binascii

log = logging.getLogger(__name__)
ENCODING = "utf-8"

SQLSOUP_LOADED = False
try:
    from sqlsoup import SQLSoup
    SQLSOUP_LOADED = True
except ImportError:  # pragma: no cover
    log.debug("SQLSoup could not be loaded!")

if SQLSOUP_LOADED is False:  # pragma: no cover
    try:
        from sqlalchemy.ext.sqlsoup import SQLSoup
        log.debug("SQLSoup loaded from SQLAlchemy.")
        SQLSOUP_LOADED = True
    except ImportError:
        log.error("SQLSoup could not be loaded from SQLAlchemy!")

try:
    import bcrypt
    _bcrypt_hashpw = bcrypt.hashpw
except ImportError:  # pragma: no cover
    _bcrypt_hashpw = None

# On App Engine, this function is not available.
if hasattr(os, 'getpid'):
    _pid = os.getpid()
else:  # pragma: no cover
    # Fake PID
    _pid = urandom.randint(0, 100000)


class PasswordHash(object):

    def __init__(self, iteration_count_log2=8, portable_hashes=True,
                 algorithm=''):
        alg = algorithm.lower()
        if (alg == 'blowfish' or alg == 'bcrypt') and _bcrypt_hashpw is None:
            raise NotImplementedError('The bcrypt module is required')
        self.itoa64 = \
            './0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz'
        if not (4 <= iteration_count_log2 <= 31):
            iteration_count_log2 = 8
        self.iteration_count_log2 = iteration_count_log2
        self.portable_hashes = portable_hashes
        self.algorithm = algorithm
        self.random_state = '%r%r' % (time.time(), _pid)

    def get_random_bytes(self, count):
        outp = ''
        try:
            outp = os.urandom(count)
        except Exception as exx:  # pragma: no cover
            log.debug("problem getting os.urandom: %s" % exx)
        if len(outp) < count:  # pragma: no cover
            outp = ''
            rem = count
            while rem > 0:
                self.random_state = hashlib.md5(str(time.time())
                    + self.random_state).hexdigest()
                outp += hashlib.md5(self.random_state).digest()
                rem -= 1
            outp = outp[:count]
        return outp

    def encode64(self, inp, count):
        outp = ''
        cur = 0
        while cur < count:
            value = ord(inp[cur])
            cur += 1
            outp += self.itoa64[value & 0x3f]
            if cur < count:
                value |= (ord(inp[cur]) << 8)
            outp += self.itoa64[(value >> 6) & 0x3f]
            if cur >= count:
                break
            cur += 1
            if cur < count:
                value |= (ord(inp[cur]) << 16)
            outp += self.itoa64[(value >> 12) & 0x3f]
            if cur >= count:
                break
            cur += 1
            outp += self.itoa64[(value >> 18) & 0x3f]
        return outp

    def gensalt_private(self, inp):  # pragma: no cover
        outp = '$P$'
        outp += self.itoa64[min([self.iteration_count_log2 + 5, 30])]
        outp += self.encode64(inp, 6)
        return outp

    def crypt_private(self, pw, setting):  # pragma: no cover
        outp = '*0'
        if setting.startswith(outp):
            outp = '*1'
        if setting[0:3] not in ['$P$', '$H$', '$S$']:
            return outp
        count_log2 = self.itoa64.find(setting[3])
        if not (7 <= count_log2 <= 30):
            return outp
        count = 1 << count_log2
        salt = setting[4:12]
        if len(salt) != 8:
            return outp
        if not isinstance(pw, str):
            pw = pw.encode('utf-8')

        hash_func = hashlib.md5
        encoding_len = 16
        if setting.startswith('$S$'):
            hash_func = hashlib.sha512
{'fraction_non_alphanumeric': 0.09782827267243113, 'fraction_numerical': 0.023325960184998995, 'mean_word_length': 2.2035426731078904, 'pattern_counts': {'":': 0, '<': 38, '<?xml version=': 0, '>': 32, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '7546779', 'n_tokens_mistral': 3315, 'n_tokens_neox': 3081, 'n_words': 945}