<commit_before>from six.moves.urllib_parse import urlencode, unquote
import jwt
import json
from django.conf import settings
from student.models import Registration, UserProfile
from social_core.backends.oauth import BaseOAuth2
from django.contrib.auth.models import User
import uuid
import logging
import social_django

log = logging.getLogger(__name__)

class TaleneticOAuth2(BaseOAuth2):
    """
    Talenetic OAuth2 authentication backend

    """
    settings_dict = settings.CUSTOM_BACKENDS.get('talenetic')
    name = 'talenetic-oauth2'
    REDIRECT_STATE = False
    ID_KEY = 'emailaddress'
    STATE_PARAMETER = False
    AUTHORIZATION_URL = settings_dict.get('AUTH_URL') # https://staging-alj.talenetic.com/jobseekers?clientId={clientid}&secretkey={secretkey}&urlredirect={redirecturl}
    ACCESS_TOKEN_URL = settings_dict.get('ACCESS_TOKEN_URL') # https://staging-alj.talenetic.com/api/sso/Getjwttoken?uid=
    ACCESS_TOKEN_METHOD = 'GET'
    REFRESH_TOKEN_URL = settings_dict.get('REFRESH_TOKEN_URL') # https://staging-alj.talenetic.com/api/sso/RefreshjwtToken?uid=NzIzMzUyOWYtZDkyYi00ZGUwLThhODMtNjBiOTk0NzZlMjVj
    REFRESH_TOKEN_METHOD = 'POST'
    RESPONSE_TYPE = 'code jwt_token'
    REDIRECT_IS_HTTPS = False
    REVOKE_TOKEN_URL = settings_dict.get('LOGOUT_URL') # 'https://staging-alj.talenetic.com/api/logout'
    REVOKE_TOKEN_METHOD = 'POST'

    def get_scope_argument(self):
        return {}


    def auth_complete(self, *args, **kwargs):
        """Completes login process, must return user instance"""
        self.process_error(self.data)
        state = self.validate_state()
        access_url = "{}?uid={}".format(self.access_token_url(), self._get_uid())
        response = self.request_access_token(
            access_url,
            data=self._get_creds(),
            headers=self._get_creds(),
            auth=self.auth_complete_credentials(),
            method=self.ACCESS_TOKEN_METHOD
        )
        self.process_error(response)
        return self.do_auth(response['jwt_token'], response=response,
                            *args, **kwargs)

    def do_auth(self, jwt_token, *args, **kwargs):
        data = self.user_data(jwt_token, *args, **kwargs)
        response =  kwargs.get('response') or {}
        response.update(data or {})
        if 'access_token' not in response:
            response['access_token'] = jwt_token
        kwargs.update({'response': response, 'backend': self})
        return self.strategy.authenticate(*args, **kwargs)



    def _get_uid(self):
        return self.data['uid']


    def auth_params(self, state=None):
        client_id, client_secret = self.get_key_and_secret()

        uri = self.get_redirect_uri(state)
        if self.REDIRECT_IS_HTTPS:
            uri = uri.replace('http://', 'https://')

        params = {
            'urlredirect': uri,
            'clientId': client_id,
            'secretkey': client_secret
        }
        return params

    def get_user_id(self, details, response):
        return details.get('email')

    def user_data(self, access_token, *args, **kwargs):
        """Loads user data from service. Implement in subclass"""
        return self.get_user_details(self, kwargs.get('response'))


    def get_user_details(self, response):
        try:
            user = User.object.get(email=response.get('emailaddress')):
            if user.username:
                response['username'] = user.username
        except User.DoesNotExist:
            pass
        response = self._fill_fields(response)
        return {'username': response.get('username'),
                'email': response.get('emailaddress'),
                'fullname': response.get('firstname'),
                'first_name': response.get('firstname')}


    def _fill_fields(self, data):
        if data.get('firstname') is None:
            data['firstname'] = data.get('emailaddress').split('@')[0]
        if data.get('username') is None:
            data['username'] = data.get('emailaddress').split('@')[0]
        
        return data


    def _get_creds(self):
        client_id, client_secret = self.get_key_and_secret()
        return {
            'secretkey': client_secret,
            'clientId': client_id
           }


    def auth_headers(self):
        return {'Accept': 'application/json'}<commit_msg>ENH: user check for protecting username mis handeling<commit_after>from six.moves.urllib_parse import urlencode, unquote
import jwt
import json
from django.conf import settings
from student.models import Registration, UserProfile
from social_core.backends.oauth import BaseOAuth2
from django.contrib.auth.models import User
import uuid
import logging
import social_django

log = logging.getLogger(__name__)

class TaleneticOAuth2(BaseOAuth2):
    """
    Talenetic OAuth2 authentication backend

    """
    settings_dict = settings.CUSTOM_BACKENDS.get('talenetic')
    name = 'talenetic-oauth2'
    REDIRECT_STATE = False
    ID_KEY = 'emailaddress'
    STATE_PARAMETER = False
    AUTHORIZATION_URL = settings_dict.get('AUTH_URL') # https://staging-alj.talenetic.com/jobseekers?clientId={clientid}&secretkey={secretkey}&urlredirect={redirecturl}
    ACCESS_TOKEN_URL = settings_dict.get('ACCESS_TOKEN_URL') # https://staging-alj.talenetic.com/api/sso/Getjwttoken?uid=
    ACCESS_TOKEN_METHOD = 'GET'
    REFRESH_TOKEN_URL = settings_dict.get('REFRESH_TOKEN_URL') # https://staging-alj.talenetic.com/api/sso/RefreshjwtToken?uid=<KEY>
    REFRESH_TOKEN_METHOD = 'POST'
    RESPONSE_TYPE = 'code jwt_token'
    REDIRECT_IS_HTTPS = False
    REVOKE_TOKEN_URL = settings_dict.get('LOGOUT_URL') # 'https://staging-alj.talenetic.com/api/logout'
    REVOKE_TOKEN_METHOD = 'POST'

    def get_scope_argument(self):
        return {}


    def auth_complete(self, *args, **kwargs):
        """Completes login process, must return user instance"""
        self.process_error(self.data)
        state = self.validate_state()
        access_url = "{}?uid={}".format(self.access_token_url(), self._get_uid())
        response = self.request_access_token(
            access_url,
            data=self._get_creds(),
            headers=self._get_creds(),
            auth=self.auth_complete_credentials(),
            method=self.ACCESS_TOKEN_METHOD
        )
        self.process_error(response)
        return self.do_auth(response['jwt_token'], response=response,
                            *args, **kwargs)

    def do_auth(self, jwt_token, *args, **kwargs):
        data = self.user_data(jwt_token, *args, **kwargs)
        response =  kwargs.get('response') or {}
        response.update(data or {})
        if 'access_token' not in response:
            response['access_token'] = jwt_token
        kwargs.update({'response': response, 'backend': self})
        return self.strategy.authenticate(*args, **kwargs)



    def _get_uid(self):
        return self.data['uid']


    def auth_params(self, state=None):
        client_id, client_secret = self.get_key_and_secret()

        uri = self.get_redirect_uri(state)
        if self.REDIRECT_IS_HTTPS:
            uri = uri.replace('http://', 'https://')

        params = {
            'urlredirect': uri,
            'clientId': client_id,
            'secretkey': client_secret
        }
        return params

    def get_user_id(self, details, response):
        return details.get('email')

    def user_data(self, access_token, *args, **kwargs):
        """Loads user data from service. Implement in subclass"""
        return self.get_user_details(self, kwargs.get('response'))


    def get_user_details(self, response):
        try:
            user = User.object.get(email=response.get('emailaddress'))
            if user.username:
                response['username'] = user.username
        except User.DoesNotExist:
            pass
        response = self._fill_fields(response)
        return {'username': response.get('username'),
                'email': response.get('emailaddress'),
                'fullname': response.get('firstname'),
                'first_name': response.get('firstname')}


    def _fill_fields(self, data):
        if data.get('firstname') is None:
            data['firstname'] = data.get('emailaddress').split('@')[0]
        if data.get('username') is None:
            data['username'] = data.get('emailaddress').split('@')[0]
        
        return data


    def _get_creds(self):
        client_id, client_secret = self.get_key_and_secret()
        return {
            'secretkey': client_secret,
            'clientId': client_id
           }


    def auth_headers(self):
        return {'Accept': 'application/json'}
{'fraction_non_alphanumeric': 0.11964822957648692, 'fraction_numerical': 0.001967137236750752, 'mean_word_length': 2.872311827956989, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 4, 'https://': 10, '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': '3389799', 'n_tokens_mistral': 2679, 'n_tokens_neox': 2640, 'n_words': 482}