throttle - backend

BaseBackend

Bases: ABC

Abstract base class for authentication backends. All authentication backends must inherit from this class.

authenticate abstractmethod async

authenticate(scope)

Performs authentication using the request scope.

PARAMETER	DESCRIPTION
scope	ASGI request scope TYPE: dict

RETURNS	DESCRIPTION
Payload	Authentication data or None TYPE: AuthData

Source code in webtool/auth/backend.py

@abstractmethod
async def authenticate(self, scope: dict) -> AuthData:
    """
    Performs authentication using the request scope.

    Parameters:
        scope: ASGI request scope

    Returns:
        Payload: Authentication data or None
    """

    raise NotImplementedError

BaseAnnoBackend

Bases: BaseBackend

Base backend class for handling anonymous users The implementation of BaseAnnoBackend should include a function to identify unauthenticated users

authenticate abstractmethod async

authenticate(scope)

Performs authentication using the request scope.

PARAMETER	DESCRIPTION
scope	ASGI request scope TYPE: dict

RETURNS	DESCRIPTION
Payload	Authentication data or None TYPE: AuthData

Source code in webtool/auth/backend.py

@abstractmethod
async def authenticate(self, scope: dict) -> AuthData:
    """
    Performs authentication using the request scope.

    Parameters:
        scope: ASGI request scope

    Returns:
        Payload: Authentication data or None
    """

    raise NotImplementedError

verify_identity abstractmethod

verify_identity(*args, **kwargs)

Method to verify the identity of anonymous users

Source code in webtool/auth/backend.py

@abstractmethod
def verify_identity(self, *args, **kwargs) -> Any:
    """
    Method to verify the identity of anonymous users
    """

    raise NotImplementedError

IPBackend

Bases: BaseBackend

Authentication backend based on IP address

authenticate async

authenticate(scope)

Performs authentication using the client's IP address.

PARAMETER	DESCRIPTION
scope	ASGI request scope TYPE: dict

RETURNS	DESCRIPTION
AuthData	IP address or None TYPE: AuthData

Source code in webtool/auth/backend.py

async def authenticate(self, scope: dict) -> AuthData:
    """
    Performs authentication using the client's IP address.

    Parameters:
        scope: ASGI request scope

    Returns:
        AuthData: IP address or None
    """

    client = scope.get("client")
    if client is None:
        raise ValueError("Authentication Failed")

    return AuthData(identifier=client[0])

SessionBackend

SessionBackend(session_name)

Bases: BaseBackend

Session-based authentication backend

PARAMETER	DESCRIPTION
session_name	Name of the session cookie TYPE: str

Source code in webtool/auth/backend.py

def __init__(self, session_name: str):
    """
    Parameters:
        session_name: Name of the session cookie
    """

    self.session_name = session_name

session_name instance-attribute

session_name = session_name

get_session

get_session(scope)

Extracts session information from request scope.

PARAMETER	DESCRIPTION
scope	ASGI request scope TYPE: dict

RETURNS	DESCRIPTION
str | None	Session string or None

Source code in webtool/auth/backend.py

def get_session(self, scope: dict) -> str | None:
    """
    Extracts session information from request scope.

    Parameters:
        scope: ASGI request scope

    Returns:
        Session string or None
    """

    headers = scope.get("headers")
    if headers is None:
        return None

    cookie = _get_header_value(headers, b"cookie")
    if cookie is None:
        return None

    session = _get_cookie_value(cookie, self.session_name.encode())
    if session is None:
        return None

    return session

authenticate async

authenticate(scope)

Performs authentication using session information.

PARAMETER	DESCRIPTION
scope	ASGI request scope TYPE: dict

RETURNS	DESCRIPTION
AuthData	session TYPE: AuthData

Source code in webtool/auth/backend.py

async def authenticate(self, scope: dict) -> AuthData:
    """
    Performs authentication using session information.

    Parameters:
        scope: ASGI request scope

    Returns:
        AuthData: session
    """

    session = self.get_session(scope)
    if not session:
        raise ValueError("Authentication Failed")

    return AuthData(identifier=session)

AnnoSessionBackend

AnnoSessionBackend(
    session_name,
    max_age=1209600,
    secure=True,
    same_site="lax",
    session_factory=uuid4,
)

Bases: SessionBackend, BaseAnnoBackend

Session backend for anonymous users. Automatically creates and assigns new sessions.

PARAMETER	DESCRIPTION
session_name	Name of the session cookie TYPE: str
max_age	Session expiration time (seconds) TYPE: int DEFAULT: 1209600
secure	HTTPS only flag TYPE: bool DEFAULT: True
same_site	SameSite cookie policy: lax, strict, none TYPE: Literal['lax', 'strict', 'none'] | None DEFAULT: 'lax'
session_factory	Session ID generation function TYPE: Optional[Callable] DEFAULT: uuid4

Source code in webtool/auth/backend.py

def __init__(
    self,
    session_name: str,
    max_age: int = 1209600,
    secure: bool = True,
    same_site: Literal["lax", "strict", "none"] | None = "lax",
    session_factory: Optional[Callable] = uuid4,
):
    """
    Parameters:
        session_name: Name of the session cookie
        max_age: Session expiration time (seconds)
        secure: HTTPS only flag
        same_site: SameSite cookie policy: lax, strict, none
        session_factory: Session ID generation function
    """

    super().__init__(session_name)

    self.session_factory = session_factory
    self.security_flags = f"httponly; samesite={same_site}; Max-Age={max_age};"
    if secure:
        self.security_flags += " secure;"

session_name instance-attribute

session_name = session_name

session_factory instance-attribute

session_factory = session_factory

security_flags instance-attribute

security_flags = (
    f"httponly; samesite={same_site}; Max-Age={max_age};"
)

authenticate async

authenticate(scope)

Performs authentication using session information.

PARAMETER	DESCRIPTION
scope	ASGI request scope TYPE: dict

RETURNS	DESCRIPTION
AuthData	session TYPE: AuthData

Source code in webtool/auth/backend.py

async def authenticate(self, scope: dict) -> AuthData:
    """
    Performs authentication using session information.

    Parameters:
        scope: ASGI request scope

    Returns:
        AuthData: session
    """

    session = self.get_session(scope)
    if not session:
        raise ValueError("Authentication Failed")

    return AuthData(identifier=session)

get_session

get_session(scope)

Extracts session information from request scope.

PARAMETER	DESCRIPTION
scope	ASGI request scope TYPE: dict

RETURNS	DESCRIPTION
str | None	Session string or None

Source code in webtool/auth/backend.py

def get_session(self, scope: dict) -> str | None:
    """
    Extracts session information from request scope.

    Parameters:
        scope: ASGI request scope

    Returns:
        Session string or None
    """

    headers = scope.get("headers")
    if headers is None:
        return None

    cookie = _get_header_value(headers, b"cookie")
    if cookie is None:
        return None

    session = _get_cookie_value(cookie, self.session_name.encode())
    if session is None:
        return None

    return session

verify_identity async

verify_identity(scope, send)

Assigns new session to anonymous users and redirects.

PARAMETER	DESCRIPTION
scope	ASGI request scope TYPE: dict
send	ASGI send function TYPE: Callable

Source code in webtool/auth/backend.py

async def verify_identity(self, scope: dict, send: Callable):
    """
    Assigns new session to anonymous users and redirects.

    Parameters:
        scope: ASGI request scope
        send: ASGI send function
    """

    async def send_wrapper(message):
        if message["type"] == "http.response.start":
            headers = message.get("headers", [])

            cookie = _get_header_value(headers, b"cookie")
            if cookie is not None:
                session = _get_cookie_value(cookie, self.session_name.encode())
                if session is not None:
                    return await send(message)

            headers.append(
                (
                    b"set-cookie",
                    f"{self.session_name}={self.session_factory().hex}; path=/; {self.security_flags}".encode(),
                )
            )

            message["headers"] = headers

        await send(message)

    return send_wrapper

JWTBackend

JWTBackend(jwt_service)

Bases: BaseBackend

JWT (JSON Web Token) based authentication backend

PARAMETER	DESCRIPTION
jwt_service	Service object for JWT processing TYPE: BaseJWTService

Source code in webtool/auth/backend.py

def __init__(self, jwt_service: "BaseJWTService"):
    """
    Parameters:
        jwt_service: Service object for JWT processing
    """

    self.jwt_service = jwt_service

jwt_service instance-attribute

jwt_service = jwt_service

validate_token async

validate_token(token)

Validates JWT.

PARAMETER	DESCRIPTION
token	JWT string TYPE: str

RETURNS	DESCRIPTION
Payload	Validated token data or None

Source code in webtool/auth/backend.py

async def validate_token(self, token: str) -> Payload:
    """
    Validates JWT.

    Parameters:
        token: JWT string

    Returns:
        Validated token data or None
    """

    validated_token = await self.jwt_service.validate_access_token(token)

    if validated_token is None or validated_token.get("sub") is None:
        raise ValueError("Authentication Failed")

    return validated_token

authenticate async

authenticate(scope)

Performs authentication using JWT.

PARAMETER	DESCRIPTION
scope	ASGI request scope TYPE: dict

RETURNS	DESCRIPTION
AuthData	Validated token data or None

Source code in webtool/auth/backend.py

async def authenticate(self, scope: dict) -> AuthData:
    """
    Performs authentication using JWT.

    Parameters:
        scope: ASGI request scope

    Returns:
        Validated token data or None
    """

    scheme, param = _get_access_token(scope)
    validated_token = await self.validate_token(param.decode())

    return AuthData(identifier=validated_token.pop("sub"), data=validated_token)

KeycloakBackend

KeycloakBackend(keycloak_connection)

Bases: BaseBackend

Source code in webtool/auth/backend.py

def __init__(self, keycloak_connection: KeycloakOpenID):
    self.keycloak_connection = keycloak_connection

keycloak_connection instance-attribute

keycloak_connection = keycloak_connection

authenticate async

authenticate(scope)

Source code in webtool/auth/backend.py

async def authenticate(self, scope: dict) -> AuthData:
    scheme, param = _get_access_token(scope)

    try:
        tokeninfo = await self.keycloak_connection.a_introspect(param.decode())
    except (KeycloakAuthenticationError, KeycloakGetError):
        raise ValueError("Authentication Failed")

    tokeninfo.setdefault("access_token", param.decode())

    return AuthData(identifier=tokeninfo.get("sub"), data=tokeninfo)