bversion/server.py
def authenticate(request: Request, context: ConnectionContext) -> Responce: # pylint: disable=W0613
    """ This does two things, either validate a pre-existing session token
    or create a new one from a signed authentication token. """

    if int(request.headers['client_version']) < version_numbers.minimum_client_version:
        return fail(unsuported_client_msg)

    # ==
    client_ip     = request.remote_addr
    repository    = request.headers['repository']
    if repository not in config['repositories']:
        return fail(no_such_repo_msg)

    # ==
    repository_path = config['repositories'][repository]['path']
    sdb = get_server_db_instance_for_thread(repository_path,
                                            path_override = config['repositories'][repository]['transient_db_path_override'])
    sdb.gc_tokens()

    # Allow resume of an existing session
    if 'session_token' in request.headers:
        session_token = request.headers['session_token']

        if sdb.validate_session(session_token, client_ip) != []:
            return success({
                'server_version' : str(version_numbers.server_version).encode('utf8'),
                'session_token'  : session_token
            })
        else:
            return fail(user_auth_fail_msg)

    # Create a new session
    else:
        user       = request.headers['user']
        auth_token = request.headers['auth_token']
        signiture  = request.headers['signature']

        try:
            public_key = config['users'][user]['public_key']

            # signature
            pysodium.crypto_sign_verify_detached(base64.b64decode(signiture), auth_token, base64.b64decode(public_key))

            # Validate token matches one we sent
            res = sdb.get_token(auth_token, client_ip)
            if res == [] or len(res) > 1:
                return fail(user_auth_fail_msg)

            # Does the user have permission to use this repository?
            if repository not in config['users'][user]['uses_repositories']:
                return fail(user_auth_fail_msg)


            # Everything OK, clean up transient token and
            # generate a session token for the client
            sdb.delete_token(auth_token)

            session_token = sdb.create_session(extend_session_duration, client_ip, user)

            return success({
                'server_version' : str(version_numbers.server_version).encode('utf8'),
                'session_token'  : session_token
            })

        except Exception: # pylint: disable=broad-except
            return fail(user_auth_fail_msg)
 This does two things, either validate a pre-existing session token
    or create a new one from a signed authentication token. 
This does two things, either validate a pre-existing session token
or create a new one from a signed authentication token.
This does two things either validate a pre - existing session token or create a new one from a signed authentication token .
def authenticate(request: Request, context: ConnectionContext) -> Responce: 
    if int(request.headers['client_version']) < version_numbers.minimum_client_version:
        return fail(unsuported_client_msg)
    client_ip     = request.remote_addr
    repository    = request.headers['repository']
    if repository not in config['repositories']:
        return fail(no_such_repo_msg)
    repository_path = config['repositories'][repository]['path']
    sdb = get_server_db_instance_for_thread(repository_path,
                                            path_override = config['repositories'][repository]['transient_db_path_override'])
    sdb.gc_tokens()
    if 'session_token' in request.headers:
        session_token = request.headers['session_token']
        if sdb.validate_session(session_token, client_ip) != []:
            return success({
                'server_version' : str(version_numbers.server_version).encode('utf8'),
                'session_token'  : session_token
            })
        else:
            return fail(user_auth_fail_msg)
    else:
        user       = request.headers['user']
        auth_token = request.headers['auth_token']
        signiture  = request.headers['signature']
        try:
            public_key = config['users'][user]['public_key']
            pysodium.crypto_sign_verify_detached(base64.b64decode(signiture), auth_token, base64.b64decode(public_key))
            res = sdb.get_token(auth_token, client_ip)
            if res == [] or len(res) > 1:
                return fail(user_auth_fail_msg)
            if repository not in config['users'][user]['uses_repositories']:
                return fail(user_auth_fail_msg)
            sdb.delete_token(auth_token)
            session_token = sdb.create_session(extend_session_duration, client_ip, user)
            return success({
                'server_version' : str(version_numbers.server_version).encode('utf8'),
                'session_token'  : session_token
            })
        except Exception: 
            return fail(user_auth_fail_msg)
def authenticate ( request : Request , context : ConnectionContext ) -> Responce : if int ( request . headers [ 'client_version' ] ) < version_numbers . minimum_client_version : return fail ( unsuported_client_msg ) client_ip = request . remote_addr repository = request . headers [ 'repository' ] if repository not in config [ 'repositories' ] : return fail ( no_such_repo_msg ) repository_path = config [ 'repositories' ] [ repository ] [ 'path' ] sdb = get_server_db_instance_for_thread ( repository_path , path_override = config [ 'repositories' ] [ repository ] [ 'transient_db_path_override' ] ) sdb . gc_tokens ( ) if 'session_token' in request . headers : session_token = request . headers [ 'session_token' ] if sdb . validate_session ( session_token , client_ip ) != [ ] : return success ( { 'server_version' : str ( version_numbers . server_version ) . encode ( 'utf8' ) , 'session_token' : session_token } ) else : return fail ( user_auth_fail_msg ) else : user = request . headers [ 'user' ] auth_token = request . headers [ 'auth_token' ] signiture = request . headers [ 'signature' ] try : public_key = config [ 'users' ] [ user ] [ 'public_key' ] pysodium . crypto_sign_verify_detached ( base64 . b64decode ( signiture ) , auth_token , base64 . b64decode ( public_key ) ) res = sdb . get_token ( auth_token , client_ip ) if res == [ ] or len ( res ) > 1 : return fail ( user_auth_fail_msg ) if repository not in config [ 'users' ] [ user ] [ 'uses_repositories' ] : return fail ( user_auth_fail_msg ) sdb . delete_token ( auth_token ) session_token = sdb . create_session ( extend_session_duration , client_ip , user ) return success ( { 'server_version' : str ( version_numbers . server_version ) . encode ( 'utf8' ) , 'session_token' : session_token } ) except Exception : return fail ( user_auth_fail_msg )
This does two things, either validate a pre-existing session token
or create a new one from a signed authentication token.
This does two things either validate a pre - existing session token or create a new one from a signed authentication token .
# pylint: disable=W0613 """ This does two things, either validate a pre-existing session token
    or create a new one from a signed authentication token. """ # == # == # Allow resume of an existing session # Create a new session # signature # Validate token matches one we sent # Does the user have permission to use this repository? # Everything OK, clean up transient token and # generate a session token for the client # pylint: disable=broad-except
request
Request
context
ConnectionContext