Class HeaderAuthServerMiddlewareFactory

python/pyarrow/tests/test_flight.py:679–704

Validates incoming username and password.

class HeaderAuthServerMiddlewareFactory(ServerMiddlewareFactory):
    """Validates incoming username and password."""

    def start_call(self, info, headers):
        auth_header = case_insensitive_header_lookup(
            headers,
            'Authorization'
        )
        values = auth_header[0].split(' ')
        token = ''
        error_message = 'Invalid credentials'

        if values[0] == 'Basic':
            decoded = base64.b64decode(values[1])
            pair = decoded.decode("utf-8").split(':')
            if not (pair[0] == 'test' and pair[1] == 'password'):
                raise flight.FlightUnauthenticatedError(error_message)
            token = 'token1234'
        elif values[0] == 'Bearer':
            token = values[1]
            if not token == 'token1234':
                raise flight.FlightUnauthenticatedError(error_message)
        else:
            raise flight.FlightUnauthenticatedError(error_message)

        return HeaderAuthServerMiddleware(token)


class HeaderAuthServerMiddleware(ServerMiddleware):