bversion/server.py
def push_file(request: Request, context: ConnectionContext) -> Responce:
    """ Push a file to the server """ #NOTE beware that reading post data in flask causes hang until file upload is complete

    session_token = request.headers['session_token'].encode('utf8')
    repository    = request.headers['repository']

    #===
    current_user = have_authenticated_user(request.remote_addr, repository, session_token)
    if current_user is False: return fail(user_auth_fail_msg)

    #===
    repository_path = config['repositories'][repository]['path']

    #===
    if not context.lock:
        return fail(lock_fail_msg)

    #===
    data_store = versioned_storage(repository_path,
                                   path_override = config['repositories'][repository]['transient_db_path_override'])

    if data_store.get_active_commit() is None: return fail(no_active_commit_msg)

    # There is no valid reason for path traversal characters to be in a file path within this system
    file_path = request.headers['path']
    if any(True for item in re.split(r'\\|/', file_path) if item in ['..', '.']): return fail()

    #===
    tmp_path = cpjoin(repository_path, 'tmp_file')
    with open(tmp_path, 'wb') as f:
        while True:
            chunk = request.body.read(1000 * 1000)
            if chunk is None: break
            f.write(chunk)

    #===
    file_info = data_store.fs_put_from_file(tmp_path, {'path' : file_path})

    return success({'file_info_json' : json.dumps(file_info)})
 Push a file to the server 
Push a file to the server
Push a file to the server
def push_file(request: Request, context: ConnectionContext) -> Responce:
    session_token = request.headers['session_token'].encode('utf8')
    repository    = request.headers['repository']
    current_user = have_authenticated_user(request.remote_addr, repository, session_token)
    if current_user is False: return fail(user_auth_fail_msg)
    repository_path = config['repositories'][repository]['path']
    if not context.lock:
        return fail(lock_fail_msg)
    data_store = versioned_storage(repository_path,
                                   path_override = config['repositories'][repository]['transient_db_path_override'])
    if data_store.get_active_commit() is None: return fail(no_active_commit_msg)
    file_path = request.headers['path']
    if any(True for item in re.split(r'\\|/', file_path) if item in ['..', '.']): return fail()
    tmp_path = cpjoin(repository_path, 'tmp_file')
    with open(tmp_path, 'wb') as f:
        while True:
            chunk = request.body.read(1000 * 1000)
            if chunk is None: break
            f.write(chunk)
    file_info = data_store.fs_put_from_file(tmp_path, {'path' : file_path})
    return success({'file_info_json' : json.dumps(file_info)})
def push_file ( request : Request , context : ConnectionContext ) -> Responce : session_token = request . headers [ 'session_token' ] . encode ( 'utf8' ) repository = request . headers [ 'repository' ] current_user = have_authenticated_user ( request . remote_addr , repository , session_token ) if current_user is False : return fail ( user_auth_fail_msg ) repository_path = config [ 'repositories' ] [ repository ] [ 'path' ] if not context . lock : return fail ( lock_fail_msg ) data_store = versioned_storage ( repository_path , path_override = config [ 'repositories' ] [ repository ] [ 'transient_db_path_override' ] ) if data_store . get_active_commit ( ) is None : return fail ( no_active_commit_msg ) file_path = request . headers [ 'path' ] if any ( True for item in re . split ( r'\\|/' , file_path ) if item in [ '..' , '.' ] ) : return fail ( ) tmp_path = cpjoin ( repository_path , 'tmp_file' ) with open ( tmp_path , 'wb' ) as f : while True : chunk = request . body . read ( 1000 * 1000 ) if chunk is None : break f . write ( chunk ) file_info = data_store . fs_put_from_file ( tmp_path , { 'path' : file_path } ) return success ( { 'file_info_json' : json . dumps ( file_info ) } )
Push a file to the server
Push a file to the server
""" Push a file to the server """ #NOTE beware that reading post data in flask causes hang until file upload is complete #=== #=== #=== #=== # There is no valid reason for path traversal characters to be in a file path within this system #=== #===
request
Request
context
ConnectionContext