tabpy.tabpy_server.handlers.basic_auth_server_middleware_factory

Complexity

Total Complexity

11

Size/Duplication

Total Lines	49
Duplicated Lines	0 %

Test Coverage

Coverage

32.42%

Importance

Changes

0

5 Methods

Rating	Name	Duplication	Size	Complexity
A	BasicAuthServerMiddleware.sending_headers()	0	2	1
A	BasicAuthServerMiddlewareFactory.__init__()	0	3	1
B	BasicAuthServerMiddlewareFactory.start_call()	0	22	6
A	BasicAuthServerMiddleware.__init__()	0	2	1
A	BasicAuthServerMiddlewareFactory.is_valid_user()	0	5	2

import base64
import secrets

from pyarrow.flight import ServerMiddlewareFactory, ServerMiddleware
from pyarrow.flight import FlightUnauthenticatedError

from tabpy.tabpy_server.handlers.util import hash_password

class BasicAuthServerMiddleware(ServerMiddleware):
    def __init__(self, token):
        self.token = token

    def sending_headers(self):
        return {"authorization": f"Bearer {self.token}"}

class BasicAuthServerMiddlewareFactory(ServerMiddlewareFactory):
    def __init__(self, creds):
        self.creds = creds
        self.tokens = {}

    def is_valid_user(self, username, password):
        if username not in self.creds:
            return False
        hashed_pwd = hash_password(username, password)
        return self.creds[username].lower() == hashed_pwd.lower()

    def start_call(self, info, headers):
        auth_header = None
        for header in headers:
            if header.lower() == "authorization":
                auth_header = headers[header][0]
                break

        if not auth_header:
            raise FlightUnauthenticatedError("No credentials supplied")
        
        auth_type, _, value = auth_header.partition(" ")

        if auth_type == "Basic":
            decoded = base64.b64decode(value).decode("utf-8")
            username, _, password = decoded.partition(":")
            if not self.is_valid_user(username, password):
                raise FlightUnauthenticatedError("Invalid credentials")
            token = secrets.token_urlsafe(32)
            self.tokens[token] = username
            return BasicAuthServerMiddleware(token)
        
        raise FlightUnauthenticatedError("No credentials supplied")