<commit_before>import logging
import re
import base64

from django.conf import settings
from django.template.loader import render_to_string
from Crypto.PublicKey import RSA
from Crypto.Hash import SHA256
from Crypto.Signature import PKCS1_PSS
from django.templatetags.i18n import language_name
from django.contrib.auth.hashers import get_hasher

from dimagi.utils.decorators.memoized import memoized
from corehq.apps.hqwebapp.forms import BulkUploadForm
from corehq.apps.hqwebapp.models import HashedPasswordLoginAttempt
from corehq.apps.hqwebapp.tasks import send_html_email_async
from corehq.apps.users.models import WebUser
from corehq.util.view_utils import get_request

logger = logging.getLogger(__name__)
HASHED_PASSWORD_EXPIRY = 30  # days
PASSWORD_HASHER = get_hasher()


@memoized
def get_hq_private_key():
    if settings.HQ_PRIVATE_KEY:
        return RSA.importKey(settings.HQ_PRIVATE_KEY)

    raise Exception('No private key found in localsettings.HQ_PRIVATE_KEY')


def sign(message):
    """
    Signs the SHA256 hash of message with HQ's private key, and returns
    the binary signature. The scheme used is RSASSA-PSS.
    """
    private_key = get_hq_private_key()
    sha256_hash = SHA256.new(message)
    signature = PKCS1_PSS.new(private_key).sign(sha256_hash)
    return signature


def send_confirmation_email(invitation):
    invited_user = invitation.email
    subject = '%s accepted your invitation to CommCare HQ' % invited_user
    recipient = WebUser.get_by_user_id(invitation.invited_by).get_email()
    context = {
        'invited_user': invited_user,
    }
    html_content = render_to_string('domain/email/invite_confirmation.html',
                                    context)
    text_content = render_to_string('domain/email/invite_confirmation.txt',
                                    context)
    send_html_email_async.delay(subject, recipient, html_content,
                                text_content=text_content)


def get_bulk_upload_form(context, context_key="bulk_upload"):
    return BulkUploadForm(
        context[context_key]['plural_noun'],
        context[context_key].get('action'),
        context_key + "_form"
    )


def csrf_inline(request):
    """
    Returns "<input type='hidden' name='csrfmiddlewaretoken' value='<csrf-token-value>' />",
    same as csrf_token template tag, but a shortcut without needing a Template or Context explicitly.

    Useful for adding inline forms in messages for e.g. while showing an "'undo' Archive Form" message
    """
    from django.template import Template, RequestContext
    node = "{% csrf_token %}"
    return Template(node).render(RequestContext(request))


def aliased_language_name(lang_code):
    """
    This is needed since we use non-standard language codes as alias, for e.g. 'fra' instead of 'fr' for French
    """
    try:
        return language_name(lang_code)
    except KeyError:
        for code, name in settings.LANGUAGES:
            if code == lang_code:
                return name
        raise KeyError('Unknown language code %s' % lang_code)


def extract_password(password):
    # Passwords set with expected salts length and padding would respect this regex
    reg_exp = r"^sha256\$([a-z|0-9|A-Z]{6})(.*)([a-z|0-9|A-Z]{6})=$"
    match_result = re.match(reg_exp, password)
    # strip out outer level padding of salts/keys and ensure three matches
    if match_result and len(match_result.groups()) == 3:
        match_groups = re.match(reg_exp, password).groups()
        hash_left = match_groups[0]
        hash_right = match_groups[2]
        stripped_password = match_groups[1]
        # decode the stripped password to get internal block
        # decoded(salt1 + encoded_password + salt2)
        try:
            decoded_password = <PASSWORD>64.b64decode(stripped_password)
        except TypeError:
            return ''
        match_result_2 = re.match(reg_exp, decoded_password)
        # strip out hashes from the internal block and ensure 3 matches
        if match_result_2 and len(match_result_2.groups()) == 3:
            match_groups = match_result_2.groups()
            # ensure the same hashes were used in the internal block as the outer
            if match_groups[0] == hash_left and match_groups[2] == hash_right:
                # decode to get the real password
                password_hash = re.match(reg_exp, decoded_password).groups()[1]
                # return password decoded for UTF-8 support
                try:
                    return base64.b64decode(password_hash).decode('utf-8')
                except TypeError:
                    return ''
            else:
                # this sounds like someone tried to hash something but failed so ignore the password submitted
                # completely
                return ''
        else:
            # this sounds like someone tried to hash something but failed so ignore the password submitted
            # completely
            return ''
    else:
        # return the password received AS-IS
        return password


def hash_password(password):
    return PASSWORD_HASHER.encode(password, PASSWORD_HASHER.salt())


def verify_password(password, password_salt):
    return PASSWORD_HASHER.verify(password, password_salt)


def decode_password(password_hash, username=None):
    def replay_attack():
        # Replay attack where the same hash used from previous login attempt
        login_attempts = HashedPasswordLoginAttempt.objects.filter(
            username=username,
        )
        for login_attempt in login_attempts:
            if verify_password(password_hash, login_attempt.password_hash):
                return True

    def record_login_attempt():
        HashedPasswordLoginAttempt.objects.create(
            username=username,
            password_hash=hash_password(password_hash)
        )

    def _decode_password():
        if username:
            if replay_attack():
                return ''
            record_login_attempt()
        return extract_password(password_hash)

    if settings.ENABLE_PASSWORD_HASHING:
        request = get_request()
        if request:
            # 1. an attempt to decode a password should be done just once in a request for the login attempt<commit_msg>skip replay_check for 2-step authentication attempts<commit_after>import logging
import re
import base64

from django.conf import settings
from django.template.loader import render_to_string
from Crypto.PublicKey import RSA
from Crypto.Hash import SHA256
from Crypto.Signature import PKCS1_PSS
from django.templatetags.i18n import language_name
from django.contrib.auth.hashers import get_hasher

from dimagi.utils.decorators.memoized import memoized
from corehq.apps.hqwebapp.forms import BulkUploadForm
from corehq.apps.hqwebapp.models import HashedPasswordLoginAttempt
from corehq.apps.hqwebapp.tasks import send_html_email_async
from corehq.apps.users.models import WebUser
from corehq.util.view_utils import get_request

logger = logging.getLogger(__name__)
HASHED_PASSWORD_EXPIRY = 30  # days
PASSWORD_HASHER = get_hasher()


@memoized
def get_hq_private_key():
    if settings.HQ_PRIVATE_KEY:
        return RSA.importKey(settings.HQ_PRIVATE_KEY)

    raise Exception('No private key found in localsettings.HQ_PRIVATE_KEY')


def sign(message):
    """
    Signs the SHA256 hash of message with HQ's private key, and returns
    the binary signature. The scheme used is RSASSA-PSS.
    """
    private_key = get_hq_private_key()
    sha256_hash = SHA256.new(message)
    signature = PKCS1_PSS.new(private_key).sign(sha256_hash)
    return signature


def send_confirmation_email(invitation):
    invited_user = invitation.email
    subject = '%s accepted your invitation to CommCare HQ' % invited_user
    recipient = WebUser.get_by_user_id(invitation.invited_by).get_email()
    context = {
        'invited_user': invited_user,
    }
    html_content = render_to_string('domain/email/invite_confirmation.html',
                                    context)
    text_content = render_to_string('domain/email/invite_confirmation.txt',
                                    context)
    send_html_email_async.delay(subject, recipient, html_content,
                                text_content=text_content)


def get_bulk_upload_form(context, context_key="bulk_upload"):
    return BulkUploadForm(
        context[context_key]['plural_noun'],
        context[context_key].get('action'),
        context_key + "_form"
    )


def csrf_inline(request):
    """
    Returns "<input type='hidden' name='csrfmiddlewaretoken' value='<csrf-token-value>' />",
    same as csrf_token template tag, but a shortcut without needing a Template or Context explicitly.

    Useful for adding inline forms in messages for e.g. while showing an "'undo' Archive Form" message
    """
    from django.template import Template, RequestContext
    node = "{% csrf_token %}"
    return Template(node).render(RequestContext(request))


def aliased_language_name(lang_code):
    """
    This is needed since we use non-standard language codes as alias, for e.g. 'fra' instead of 'fr' for French
    """
    try:
        return language_name(lang_code)
    except KeyError:
        for code, name in settings.LANGUAGES:
            if code == lang_code:
                return name
        raise KeyError('Unknown language code %s' % lang_code)


def extract_password(password):
    # Passwords set with expected salts length and padding would respect this regex
    reg_exp = r"^sha256\$([a-z|0-9|A-Z]{6})(.*)([a-z|0-9|A-Z]{6})=$"
    match_result = re.match(reg_exp, password)
    # strip out outer level padding of salts/keys and ensure three matches
    if match_result and len(match_result.groups()) == 3:
        match_groups = re.match(reg_exp, password).groups()
        hash_left = match_groups[0]
        hash_right = match_groups[2]
        stripped_password = match_groups[1]
        # decode the stripped password to get internal block
        # decoded(salt1 + encoded_password + salt2)
        try:
            decoded_password = base64.b64decode(stripped_password)
        except TypeError:
            return ''
        match_result_2 = re.match(reg_exp, decoded_password)
        # strip out hashes from the internal block and ensure 3 matches
        if match_result_2 and len(match_result_2.groups()) == 3:
            match_groups = match_result_2.groups()
            # ensure the same hashes were used in the internal block as the outer
            if match_groups[0] == hash_left and match_groups[2] == hash_right:
                # decode to get the real password
                password_hash = re.match(reg_exp, decoded_password).groups()[1]
                # return password decoded for UTF-8 support
                try:
                    return base64.b64decode(password_hash).decode('utf-8')
                except TypeError:
                    return ''
            else:
                # this sounds like someone tried to hash something but failed so ignore the password submitted
                # completely
                return ''
        else:
            # this sounds like someone tried to hash something but failed so ignore the password submitted
            # completely
            return ''
    else:
        # return the password received AS-IS
        return password


def hash_password(password):
    return PASSWORD_HASHER.encode(password, PASSWORD_HASHER.salt())


def verify_password(password, password_salt):
    return PASSWORD_HASHER.verify(password, password_salt)


def decode_password(password_hash, username=None):
    def replay_attack():
        # Replay attack where the same hash used from previous login attempt
        login_attempts = HashedPasswordLoginAttempt.objects.filter(
            username=username,
        )
        for login_attempt in login_attempts:
            if verify_password(password_hash, login_attempt.password_hash):
                return True

    def record_login_attempt():
        HashedPasswordLoginAttempt.objects.create(
            username=username,
            password_hash=hash_password(password_hash)
        )

    def _decode_password():
        # force check for replay attack and recording login attempt only for web sign in by checking for username
        # Also skip those two checks in case of 2-step authentication by checking for auth-username which is not
        # present in consecutive token step's POST params
        if username and request.POST.get('auth-username'):
            if replay_attack():
                return ''
            record_login_attempt()
        return extract_password(password_hash)

    if settings.ENABLE_PASSWORD_HASHING:
        request = get_request()
        if request:
            # 1. an attempt to decode a password should be done just once in a request for the login attempt
{'fraction_non_alphanumeric': 0.07412587412587412, 'fraction_numerical': 0.00916860916860917, 'mean_word_length': 2.8386519534745003, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4984946', 'n_tokens_mistral': 3687, 'n_tokens_neox': 3508, 'n_words': 1008}