<commit_before>import csv
import itertools
import os
import urllib.parse
import uuid
from collections import Counter
from datetime import datetime, timedelta
from io import StringIO

from django.conf import settings
from django.contrib import messages
from django.contrib.auth.models import User
from django.core.mail import mail_admins
from django.db.models import Q
from django.http import (
    HttpResponse,
    HttpResponseBadRequest,
    HttpResponseNotFound,
    JsonResponse,
    StreamingHttpResponse,
)
from django.http.response import Http404
from django.shortcuts import redirect, render
from django.template.loader import render_to_string
from django.utils.decorators import method_decorator
from django.utils.functional import cached_property
from django.utils.translation import gettext as _
from django.utils.translation import gettext_lazy
from django.views.generic import FormView, TemplateView, View

from couchdbkit.exceptions import ResourceNotFound
from lxml import etree
from lxml.builder import E
from two_factor.utils import default_device

from casexml.apps.phone.xml import SYNC_XMLNS
from casexml.apps.stock.const import COMMTRACK_REPORT_XMLNS
from corehq.apps.hqadmin.utils import unset_password
from couchexport.models import Format
from couchforms.openrosa_response import RESPONSE_XMLNS
from dimagi.utils.django.email import send_HTML_email

from corehq.apps.app_manager.models import Application
from corehq.apps.domain.auth import basicauth
from corehq.apps.domain.decorators import (
    check_lockout,
    domain_admin_required,
    login_or_basic,
    require_superuser,
)
from corehq.apps.hqadmin.forms import (
    DisableTwoFactorForm,
    DisableUserForm,
    SuperuserManagementForm,
)
from corehq.apps.hqadmin.views.utils import BaseAdminSectionView
from corehq.apps.hqmedia.tasks import create_files_for_ccz
from corehq.apps.ota.views import get_restore_params, get_restore_response
from corehq.apps.users.audit.change_messages import UserChangeMessage
from corehq.apps.users.models import CommCareUser, CouchUser, WebUser
from corehq.apps.users.util import format_username, log_user_change
from corehq.const import USER_CHANGE_VIA_WEB
from corehq.util import reverse
from corehq.util.timer import TimingContext


class UserAdministration(BaseAdminSectionView):
    section_name = gettext_lazy("User Administration")


class SuperuserManagement(UserAdministration):
    urlname = 'superuser_management'
    page_title = _("Grant or revoke superuser access")
    template_name = 'hqadmin/superuser_management.html'

    @method_decorator(require_superuser)
    def dispatch(self, *args, **kwargs):
        return super(SuperuserManagement, self).dispatch(*args, **kwargs)

    @property
    def page_context(self):
        # only staff can toggle is_staff
        can_toggle_is_staff = self.request.user.is_staff
        # render validation errors if rendered after POST
        args = [can_toggle_is_staff, self.request.POST] if self.request.POST else [can_toggle_is_staff]
        return {
            'form': SuperuserManagementForm(*args),
            'users': augmented_superusers(),
        }

    def post(self, request, *args, **kwargs):
        can_toggle_is_staff = request.user.is_staff
        form = SuperuserManagementForm(can_toggle_is_staff, self.request.POST)
        if form.is_valid():
            users = form.cleaned_data['users']
            is_superuser = 'is_superuser' in form.cleaned_data['privileges']
            is_staff = 'is_staff' in form.cleaned_data['privileges']
            fields_changed = {}
            for user in users:
                # save user object only if needed and just once
                if user.is_superuser is not is_superuser:
                    user.is_superuser = is_superuser
                    fields_changed['is_superuser'] = is_superuser

                if can_toggle_is_staff and user.is_staff is not is_staff:
                    user.is_staff = is_staff
                    fields_changed['is_staff'] = is_staff

                if fields_changed:
                    user.save()
                    couch_user = CouchUser.from_django_user(user)
                    log_user_change(by_domain=None, for_domain=None, couch_user=couch_user,
                                    changed_by_user=self.request.couch_user,
                                    changed_via=USER_CHANGE_VIA_WEB, fields_changed=fields_changed,
                                    by_domain_required_for_log=False,
                                    for_domain_required_for_log=False)
            messages.success(request, _("Successfully updated superuser permissions"))

        return self.get(request, *args, **kwargs)


@require_superuser
def superuser_table(request):
    superusers = augmented_superusers()
    f = StringIO()
    csv_writer = csv.writer(f)
    csv_writer.writerow(['Username', 'Developer', 'Superuser', 'Two Factor Enabled'])
    for user in superusers:
        csv_writer.writerow([
            user.username, user.is_staff, user.is_superuser, user.two_factor_enabled])
    response = HttpResponse(content_type=Format.from_format('csv').mimetype)
    response['Content-Disposition'] = 'attachment; filename="superuser_table.csv"'
    response.write(f.getvalue())
    return response


def augmented_superusers():
    return _augment_users_with_two_factor_enabled(User.objects.filter(
        Q(is_superuser=True) | Q(is_staff=True)
    ))


def _augment_users_with_two_factor_enabled(users):
    """Annotate a User queryset with a two_factor_enabled field"""
    for user in users:
        user.two_factor_enabled = bool(default_device(user))
    return users


class AdminRestoreView(TemplateView):
    template_name = 'hqadmin/admin_restore.html'

    @method_decorator(require_superuser)
    def dispatch(self, request, *args, **kwargs):
        return super(AdminRestoreView, self).dispatch(request, *args, **kwargs)

    def _validate_user_access(self, user):
        return True

    def get(self, request, *args, **kwargs):
        self.domain = request.GET.get('domain')
        full_username = request.GET.get('as', '')

        if not full_username or '@' not in full_username:
            msg = 'Please specify a user using ?as=user@domain\nOr specify a web-user using ?as=email&domain=domain'
            return HttpResponseBadRequest(msg)

        self.user = CouchUser.get_by_username(full_username)<commit_msg>fixed line length<commit_after>import csv
import itertools
import os
import urllib.parse
import uuid
from collections import Counter
from datetime import datetime, timedelta
from io import StringIO

from django.conf import settings
from django.contrib import messages
from django.contrib.auth.models import User
from django.core.mail import mail_admins
from django.db.models import Q
from django.http import (
    HttpResponse,
    HttpResponseBadRequest,
    HttpResponseNotFound,
    JsonResponse,
    StreamingHttpResponse,
)
from django.http.response import Http404
from django.shortcuts import redirect, render
from django.template.loader import render_to_string
from django.utils.decorators import method_decorator
from django.utils.functional import cached_property
from django.utils.translation import gettext as _
from django.utils.translation import gettext_lazy
from django.views.generic import FormView, TemplateView, View

from couchdbkit.exceptions import ResourceNotFound
from lxml import etree
from lxml.builder import E
from two_factor.utils import default_device

from casexml.apps.phone.xml import SYNC_XMLNS
from casexml.apps.stock.const import COMMTRACK_REPORT_XMLNS
from corehq.apps.hqadmin.utils import unset_password
from couchexport.models import Format
from couchforms.openrosa_response import RESPONSE_XMLNS
from dimagi.utils.django.email import send_HTML_email

from corehq.apps.app_manager.models import Application
from corehq.apps.domain.auth import basicauth
from corehq.apps.domain.decorators import (
    check_lockout,
    domain_admin_required,
    login_or_basic,
    require_superuser,
)
from corehq.apps.hqadmin.forms import (
    DisableTwoFactorForm,
    DisableUserForm,
    SuperuserManagementForm,
)
from corehq.apps.hqadmin.views.utils import BaseAdminSectionView
from corehq.apps.hqmedia.tasks import create_files_for_ccz
from corehq.apps.ota.views import get_restore_params, get_restore_response
from corehq.apps.users.audit.change_messages import UserChangeMessage
from corehq.apps.users.models import CommCareUser, CouchUser, WebUser
from corehq.apps.users.util import format_username, log_user_change
from corehq.const import USER_CHANGE_VIA_WEB
from corehq.util import reverse
from corehq.util.timer import TimingContext


class UserAdministration(BaseAdminSectionView):
    section_name = gettext_lazy("User Administration")


class SuperuserManagement(UserAdministration):
    urlname = 'superuser_management'
    page_title = _("Grant or revoke superuser access")
    template_name = 'hqadmin/superuser_management.html'

    @method_decorator(require_superuser)
    def dispatch(self, *args, **kwargs):
        return super(SuperuserManagement, self).dispatch(*args, **kwargs)

    @property
    def page_context(self):
        # only staff can toggle is_staff
        can_toggle_is_staff = self.request.user.is_staff
        # render validation errors if rendered after POST
        args = [can_toggle_is_staff, self.request.POST] if self.request.POST else [can_toggle_is_staff]
        return {
            'form': SuperuserManagementForm(*args),
            'users': augmented_superusers(),
        }

    def post(self, request, *args, **kwargs):
        can_toggle_is_staff = request.user.is_staff
        form = SuperuserManagementForm(can_toggle_is_staff, self.request.POST)
        if form.is_valid():
            users = form.cleaned_data['users']
            is_superuser = 'is_superuser' in form.cleaned_data['privileges']
            is_staff = 'is_staff' in form.cleaned_data['privileges']
            fields_changed = {}
            for user in users:
                # save user object only if needed and just once
                if user.is_superuser is not is_superuser:
                    user.is_superuser = is_superuser
                    fields_changed['is_superuser'] = is_superuser

                if can_toggle_is_staff and user.is_staff is not is_staff:
                    user.is_staff = is_staff
                    fields_changed['is_staff'] = is_staff

                if fields_changed:
                    user.save()
                    couch_user = CouchUser.from_django_user(user)
                    log_user_change(by_domain=None, for_domain=None, couch_user=couch_user,
                                    changed_by_user=self.request.couch_user,
                                    changed_via=USER_CHANGE_VIA_WEB, fields_changed=fields_changed,
                                    by_domain_required_for_log=False,
                                    for_domain_required_for_log=False)
            messages.success(request, _("Successfully updated superuser permissions"))

        return self.get(request, *args, **kwargs)


@require_superuser
def superuser_table(request):
    superusers = augmented_superusers()
    f = StringIO()
    csv_writer = csv.writer(f)
    csv_writer.writerow(['Username', 'Developer', 'Superuser', 'Two Factor Enabled'])
    for user in superusers:
        csv_writer.writerow([
            user.username, user.is_staff, user.is_superuser, user.two_factor_enabled])
    response = HttpResponse(content_type=Format.from_format('csv').mimetype)
    response['Content-Disposition'] = 'attachment; filename="superuser_table.csv"'
    response.write(f.getvalue())
    return response


def augmented_superusers():
    return _augment_users_with_two_factor_enabled(User.objects.filter(
        Q(is_superuser=True) | Q(is_staff=True)
    ))


def _augment_users_with_two_factor_enabled(users):
    """Annotate a User queryset with a two_factor_enabled field"""
    for user in users:
        user.two_factor_enabled = bool(default_device(user))
    return users


class AdminRestoreView(TemplateView):
    template_name = 'hqadmin/admin_restore.html'

    @method_decorator(require_superuser)
    def dispatch(self, request, *args, **kwargs):
        return super(AdminRestoreView, self).dispatch(request, *args, **kwargs)

    def _validate_user_access(self, user):
        return True

    def get(self, request, *args, **kwargs):
        self.domain = request.GET.get('domain')
        full_username = request.GET.get('as', '')

        if not full_username or '@' not in full_username:
            msg = 'Please specify a user using ?as=user@domain\nOr a web-user using ?as=email&domain=domain'
            return HttpResponseBadRequest(msg)

        self.user = CouchUser.get_by_username(full_username)
{'fraction_non_alphanumeric': 0.07573872472783826, 'fraction_numerical': 0.00046656298600311044, 'mean_word_length': 3.478064066852368, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 10}, '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': '503587', 'n_tokens_mistral': 3742, 'n_tokens_neox': 3634, 'n_words': 858}