Authentication

The authenticator module provides the core authentication logic used across all supported frameworks. It handles token extraction from request headers and query parameters, token validation against the authentication service, and permission checking.

Core Components

The AuthenticateBase class is the foundation for all authentication implementations. It provides methods for extracting tokens from requests, calling the authentication service, and performing permission checks. Framework-specific implementations (FastAPI, Django, Django Ninja) extend this base class to integrate with their respective request/response patterns.

Token types supported:

• Public key: key query parameter (starts with woos-)
• Private key: private_key query parameter or X-Api-Key header
• User token: Authorization: Bearer <token> header (JWT)

The EndpointMode enum controls permission enforcement:

• READ_WRITE: Allows both read and write operations (default)
• READ_ONLY: Restricts to GET/POST methods only
• WRITE_ONLY: Requires private keys with write permission

Type System

The types module defines data models for tokens, permissions, and project references. After successful authentication, a ProjectTokenModel is attached to the request containing:

• The validated token instance with organization/project information
• A ProjectReference with organization_id and project_id
• Context flags like should_count (whether to count for metrics)

application_kit.authenticator.base

Framework-agnostic authentication helpers.

See also:

• FastAPI: application_kit.fastapi.security
• Django: application_kit.django.decorators

EndpointMode

Bases: StrEnum

EndpointMode controls how an endpoint should require public or private key based on the request verb.

AuthenticateBase

AuthenticateBase(
    classes=None,
    token_lambda=None,
    product=None,
    write_permission_needed=None,
    endpoint_mode=UNSET,
)

Base class that encompasses token fetching, and perform instance checks

PARAMETER	DESCRIPTION
product	The products required to be activated on the project. TYPE: Products | None DEFAULT: None
write_permission_needed	Deprecated use endpoint_mode instead. TYPE: bool | None DEFAULT: None
endpoint_mode	Controls what kind of permissions are required from the ProjectToken. TYPE: EndpointMode DEFAULT: UNSET

Source code in application_kit/authenticator/base.py

def __init__(
    self,
    classes: list[type[CheckablePermission]] | None = None,
    token_lambda: TokenLambda | None = None,
    product: Annotated[Products | None, Doc("The products required to be activated on the project.")] = None,
    write_permission_needed: Annotated[bool | None, Doc("**Deprecated** use endpoint_mode instead.")] = None,
    endpoint_mode: Annotated[
        EndpointMode, Doc("Controls what kind of permissions are required from the ProjectToken.")
    ] = EndpointMode.UNSET,
) -> None:
    self.classes = classes
    self.token_lambda = token_lambda
    if endpoint_mode == EndpointMode.UNSET:
        match write_permission_needed:
            case None:
                self.endpoint_mode = EndpointMode.READ_WRITE
            case True:
                self.endpoint_mode = EndpointMode.WRITE_ONLY
            case False:
                self.endpoint_mode = EndpointMode.READ_ONLY
    else:
        self.endpoint_mode = endpoint_mode

    if self.classes is None or len(self.classes) == 0:
        if self.endpoint_mode == EndpointMode.WRITE_ONLY:
            self.token_lambda = partial(get_authorization_token, [PRIVATE_KEY])

    self.product = product
    self.write_permission_needed = write_permission_needed

get_project_token

get_project_token(
    current_product,
    headers,
    key_token,
    kind,
    request_method,
    port,
    organization_id=None,
    project_id=None,
)

Calls authentication service to get a token for an incoming request.

PARAMETER	DESCRIPTION
headers	The headers mapping from the request TYPE: Mapping[str, str]
kind	The kind of token detected. TYPE: PublicKeyType | PrivateKeyType | UserTokenType
organization_id	organization_id coming from the request path, used with user tokens. TYPE: int | None DEFAULT: None
project_id	project_id coming from the request path, used with user tokens. TYPE: int | None DEFAULT: None

RETURNS	DESCRIPTION
tuple[ProjectTokenModel, TokenModel]	Returns ProjectTokenModel and a TokenModel, the later will be retired.

Source code in application_kit/authenticator/base.py

def get_project_token(
    self,
    current_product: Products | None,
    headers: Annotated[Mapping[str, str], Doc("The headers mapping from the request")],
    key_token: str,
    kind: Annotated[PublicKeyType | PrivateKeyType | UserTokenType, Doc("The kind of token detected.")],
    request_method: str,
    port: int | None,
    organization_id: Annotated[
        int | None, Doc("organization_id coming from the request path, used with user tokens.")
    ] = None,
    project_id: Annotated[
        int | None, Doc("project_id coming from the request path, used with user tokens.")
    ] = None,
) -> Annotated[
    tuple[ProjectTokenModel, TokenModel],
    Doc("Returns ProjectTokenModel and a TokenModel, the later will be retired."),
]:
    """Calls authentication service to get a token for an incoming request."""
    token, served_stale = authenticate_request(key_token, kind)

    readable_token = decode_jwt(token, allow_expired=served_stale)

    project_reference: ProjectReference

    instance = readable_token.instance
    if isinstance(instance, UserTokenInstanceModel) and user_token_has_tombstone(instance.uuid):
        raise AuthenticationFailed()
    _, project_reference = _get_products_and_project_reference(instance, organization_id, project_id)
    project_token = ProjectTokenModel(instance)
    project_token.reference = project_reference
    project_token.context.should_count = port != 8000

    aicontext = headers.get("ai-context", "false").lower() == "true"
    project_token.context.source = headers.get("x-sdk-source") if aicontext is False else "ai-context"

    project_token.check_product_allowed(current_product)

    project_reference.trace_to_datadog()

    assert request_method is not None

    return project_token, readable_token

perform_instance_checks

perform_instance_checks(
    project_token, headers, request_method
)

Performs permissions, and restrictions checks, returns extra response headers.

Source code in application_kit/authenticator/base.py

def perform_instance_checks(
    self,
    project_token: ProjectTokenModel,
    headers: Mapping[str, str],
    request_method: str,
) -> dict[str, str]:
    """Performs permissions, and restrictions checks, returns extra response headers."""

    instance = project_token.instance

    if instance.kind != USER_TOKEN:
        if not self.classes or len(self.classes) == 0:
            match self.endpoint_mode:
                case EndpointMode.READ_WRITE:
                    instance.check_write_needed(request_method)
                case EndpointMode.WRITE_ONLY:
                    instance.check_permissions("stores", ["write"])
                case EndpointMode.READ_ONLY:
                    if request_method not in ["GET", "POST"]:
                        raise PermissionDenied()

            permissions_extra_headers = instance.check_restriction(headers)
        else:
            warnings.warn(
                "@permission_classes decorator is deprecated for public or private key authentication. see @authenticate_key decorator instead.",
                DeprecationWarning,
                stacklevel=4,
            )
            permissions_extra_headers = authorize_request(project_token, headers, self.classes)
    else:
        permissions_extra_headers = authorize_request(project_token, headers, self.classes or [])
        if instance.kind == USER_TOKEN:
            permissions_extra_headers |= instance.check_origin(headers)
    return permissions_extra_headers

get_public_key

get_public_key(parameters, headers)

Gets a public key from request parameters.

Source code in application_kit/authenticator/base.py

def get_public_key(parameters: dict[str, str], headers: Mapping[str, str]) -> str | None:
    """
    Gets a public key from request parameters.
    """
    public_key = parameters.get("key")

    return public_key

get_private_key

get_private_key(parameters, headers)

Gets a private key from either the request parameters or headers.

Source code in application_kit/authenticator/base.py

def get_private_key(parameters: dict[str, str], headers: Mapping[str, str]) -> str | None:
    """Gets a private key from either the request parameters or headers."""
    private_key = parameters.get("private_key")

    if private_key is None:
        private_key = headers.get("X-Api-Key")

    return private_key

get_user_access_token

get_user_access_token(parameters, headers)

Gets a user access token using the Authorization request header.

Source code in application_kit/authenticator/base.py

def get_user_access_token(parameters: dict[str, str], headers: Mapping[str, str]) -> str | None:
    """Gets a user access token using the Authorization request header."""
    authorization_header = headers.get("Authorization")
    access_token = None

    if authorization_header:
        authorization_components = authorization_header.strip().split(" ")

        if len(authorization_components) == 2 and authorization_components[0].lower() == "bearer":
            access_token = authorization_components[1]

    return access_token

decode_jwt

decode_jwt(jwt_token, *, allow_expired=False)

Decode and validate an access-token JWT.

allow_expired skips only the exp check (signature and every other claim are still verified). It is used when a stale token is served in degraded mode: the auth service issues short-lived access tokens, so a token kept for the grace window has expired by the time it is served, yet remains the last credential we can trust.

Source code in application_kit/authenticator/base.py

def decode_jwt(jwt_token: str, *, allow_expired: bool = False) -> TokenModel:
    """Decode and validate an access-token JWT.

    ``allow_expired`` skips only the ``exp`` check (signature and every other claim are
    still verified). It is used when a stale token is served in degraded mode: the auth
    service issues short-lived access tokens, so a token kept for the grace window has
    expired by the time it is served, yet remains the last credential we can trust.
    """
    try:
        return parse_obj(
            TokenModel,
            jwt.decode(
                jwt_token,
                base64.b64decode(get_configuration("JWT_PUBLIC_KEY")).decode("utf-8"),
                algorithms=["ES256"],
                options={"verify_exp": False} if allow_expired else {},
            ),
        )
    except (jwt.InvalidTokenError, ValueError) as e:
        logger.debug(f"Invalid JWT token. {jwt_token} ")
        raise AuthenticationFailed() from e

user_token_has_tombstone

user_token_has_tombstone(token_uuid)

Return True if the user token has been revoked (tombstoned) by the auth service.

Source code in application_kit/authenticator/base.py

def user_token_has_tombstone(token_uuid: str | None) -> bool:
    """Return True if the user token has been revoked (tombstoned) by the auth service."""
    if token_uuid is None:
        return False
    try:
        return get_redis_instance("authentication_cache").get(_make_tombstone_key(token_uuid)) is not None
    except RedisError as e:
        logger.debug("Unreachable authentication_cache redis server", exc_info=e)
        return False

access_token_for_key

access_token_for_key(key, kind)

Resolve an access token, returning (token, served_stale).

Degraded mode: when the cached token is stale and the authentication service is unavailable (transport error or 5xx, surfaced as a plain HttpException), the stale token is served instead of failing the request and served_stale is True so the caller can accept its expired exp. Authoritative rejections (AuthenticationFailed, OverFreeQuota) are never softened: a stale negative-cache marker also keeps rejecting during an outage.

Source code in application_kit/authenticator/base.py

def access_token_for_key(key: str, kind: PrivateKeyType | PublicKeyType) -> tuple[str, bool]:
    """Resolve an access token, returning ``(token, served_stale)``.

    Degraded mode: when the cached token is stale and the authentication service is
    unavailable (transport error or 5xx, surfaced as a plain ``HttpException``), the stale
    token is served instead of failing the request and ``served_stale`` is ``True`` so the
    caller can accept its expired ``exp``. Authoritative rejections (``AuthenticationFailed``,
    ``OverFreeQuota``) are never softened: a stale negative-cache marker also keeps rejecting
    during an outage.
    """
    cached = _read_cached_token(key, kind)
    if cached is not None and cached.is_fresh:
        return _token_or_raise(cached.token), False
    try:
        return _access_token_for_key_in_auth_service(key, kind), False
    except (AuthenticationFailed, OverFreeQuota):
        raise
    except HttpException:
        if cached is None:
            raise
        set_root_span_tag("auth.stale_token_served", get_token_cache_key(key, kind))
        logger.warning(f"Authentication service unavailable, serving stale cached token for key: {key}")
        return _token_or_raise(cached.token), True

authenticate_request

authenticate_request(token, kind)

Return (token, served_stale); served_stale is only ever True for key tokens.

Source code in application_kit/authenticator/base.py

def authenticate_request(token: str, kind: PublicKeyType | PrivateKeyType | UserTokenType) -> tuple[str, bool]:
    """Return ``(token, served_stale)``; ``served_stale`` is only ever True for key tokens."""
    if token is not None:
        served_stale = False
        if kind != USER_TOKEN:
            token, served_stale = access_token_for_key(token, kind)
        return token, served_stale
    else:
        raise PermissionDenied()

application_kit.authenticator.types

Products

Bases: StrEnum

Billable products. Each member carries its console-facing label.

Labels track the product names shown in the console (console/src/data/Products.js). Defining the label inline (rather than in a side map) makes it intrinsic to the enum: a new member can't be added without a label, so the two never drift. Every member has its own distinct label — billing variants the console groups under a parent product (the INDOOR_* kinds, the deprecated Localities fields) still get their own.

RestrictionModel

Bases: ABC, BaseModel

is_fulfilled abstractmethod

is_fulfilled(headers)

is_fulfilled protocol method.

Source code in application_kit/authenticator/types.py

@abc.abstractmethod
def is_fulfilled(self, headers: Mapping[str, str]) -> tuple[bool, dict[str, str]]:
    """is_fulfilled protocol method."""

KeyTokenInstanceModel

Bases: ABC, BaseModel

Base class for Key token instances.

check_restriction abstractmethod

check_restriction(headers)

Checks that at least one restriction is fulfilled.

Source code in application_kit/authenticator/types.py

@abc.abstractmethod
def check_restriction(self, headers: Mapping[str, str]) -> dict[str, str]:
    """Checks that at least one restriction is fulfilled."""

check_permissions abstractmethod

check_permissions(product, permission_names)

Checks that the instance has the permission for the product.

Source code in application_kit/authenticator/types.py

@abc.abstractmethod
def check_permissions(self, product: str, permission_names: list[str]) -> None:
    """Checks that the instance has the permission for the product."""

check_write_needed

check_write_needed(method)

Checks if the method requires write permission.

Note

The following methods require write permission: POST, PUT, DELETE and PATCH

PARAMETER	DESCRIPTION
method	The http method used by the request. TYPE: str

Source code in application_kit/authenticator/types.py

def check_write_needed(self, method: Annotated[str, Doc("The http method used by the request.")]) -> None:
    """Checks if the method requires write permission.
    !!! note
        The following methods require write permission:
        `POST`, `PUT`, `DELETE` and `PATCH`
    """

    write_methods = ["POST", "PUT", "DELETE", "PATCH"]

    if method in write_methods:
        self.check_permissions("stores", ["write"])

PrivateKeyTokenInstanceModel

Bases: KeyTokenInstanceModel

Token instance for a private key.

Note

Retrived from authentication service /access_tokens/ endpoint.

check_permissions

check_permissions(product, permission_names)

Checks that the instance has the permission for the product.

Warning

This is now defunkt the only permission remaning is write for the stores product.

This is used by every products, and translates to simply "has write permission".

Source code in application_kit/authenticator/types.py

def check_permissions(self, product: str, permission_names: list[str]) -> None:
    """Checks that the instance has the permission for the product.

    !!! warning
        This is now defunkt the only permission remaning is `write` for the `stores` product.

        This is used by every products, and translates to simply "has write permission".
    """
    permissions = self.permissions.get(product, [])
    for name in permission_names:
        if permissions and name in permissions:
            return
    logger.debug(f"Token has not the permission {self}. Requested permissions : {permission_names} on {product}")
    raise PermissionDenied

check_write_needed

check_write_needed(method)

Checks if the method requires write permission.

Note

The following methods require write permission: POST, PUT, DELETE and PATCH

PARAMETER	DESCRIPTION
method	The http method used by the request. TYPE: str

Source code in application_kit/authenticator/types.py

def check_write_needed(self, method: Annotated[str, Doc("The http method used by the request.")]) -> None:
    """Checks if the method requires write permission.
    !!! note
        The following methods require write permission:
        `POST`, `PUT`, `DELETE` and `PATCH`
    """

    write_methods = ["POST", "PUT", "DELETE", "PATCH"]

    if method in write_methods:
        self.check_permissions("stores", ["write"])

PublicKeyTokenInstanceModel

Bases: KeyTokenInstanceModel

Token instance for a public key.

Note

Retrived from authentication service /access_tokens/ endpoint.

check_write_needed

check_write_needed(method)

Checks if the method requires write permission.

Note

The following methods require write permission: POST, PUT, DELETE and PATCH

PARAMETER	DESCRIPTION
method	The http method used by the request. TYPE: str

Source code in application_kit/authenticator/types.py

def check_write_needed(self, method: Annotated[str, Doc("The http method used by the request.")]) -> None:
    """Checks if the method requires write permission.
    !!! note
        The following methods require write permission:
        `POST`, `PUT`, `DELETE` and `PATCH`
    """

    write_methods = ["POST", "PUT", "DELETE", "PATCH"]

    if method in write_methods:
        self.check_permissions("stores", ["write"])

UserTokenInstanceModel

Bases: BaseModel

Token instance for a user.

Note

Retrived from the Authorization: bearer .

check_staff_or_superuser

check_staff_or_superuser()

Checks if user is staff or super user. :raise PermissionDenied :return:

Source code in application_kit/authenticator/types.py

def check_staff_or_superuser(self) -> None:
    """
    Checks if user is staff or super user.
    :raise PermissionDenied
    :return:
    """
    if not self.is_superuser and not self.is_staff:
        raise PermissionDenied

check_superuser

check_superuser()

Checks if user is super user. :raise PermissionDenied

Source code in application_kit/authenticator/types.py

def check_superuser(self) -> None:
    """
    Checks if user is super user.
    :raise PermissionDenied
    """
    if not self.is_superuser:
        raise PermissionDenied

check_membership

check_membership(organization_id)

Checks if user is member of the organization with organization_id. :raise PermissionDenied

Source code in application_kit/authenticator/types.py

def check_membership(self, organization_id: int) -> None:
    """
    Checks if user is member of the organization with `organization_id`.
    :raise PermissionDenied
    """
    if self.organization is not None and self.organization.pk == organization_id:
        return
    raise PermissionDenied

check_ownership

check_ownership(organization_id)

Checks that user has ownership of the organization.

Source code in application_kit/authenticator/types.py

def check_ownership(self, organization_id: int) -> None:
    """Checks that user has ownership of the organization."""
    if self.organization is not None and self.organization.pk == organization_id and self.organization.is_owner:
        return
    raise PermissionDenied

check_origin

check_origin(headers)

Checks if the Origin is allowed for User tokens browser interactions.

Source code in application_kit/authenticator/types.py

def check_origin(self, headers: Mapping[str, str]) -> dict[str, str]:
    """Checks if the Origin is allowed for User tokens browser interactions."""
    from application_kit.settings import get_cors_origin_whitelist

    origin = headers.get(HTTP_ORIGIN)

    if origin is not None:
        parsed_origin = urlparse(origin)
        if parsed_origin.netloc and is_domain_allowed(parsed_origin.netloc, get_cors_origin_whitelist()):
            return {HTTP_ACCESS_CONTROL_ALLOW_ORIGIN: origin}

    return {}

GenericCallable

Bases: Protocol

__call__

__call__(*args, **kwargs)

Generic callable takes Any as parameters and returns Any.

Source code in application_kit/authenticator/types.py

def __call__(self, *args: Any, **kwargs: Any) -> Any:
    """Generic callable takes Any as parameters and returns Any."""

ProjectReference

Bases: BaseModel

Project reference. Encapsulates the organization ID and the project ID coming from authentication.

Tip

Used to identify the targeted project by the request. For Private or Public keys tokens it's direct, but for User tokens it's a bit more involved.

User can target multiple projects or different organizations that's why the endpoints accepting User tokens must have organization_id and project_id path arguments.

organization_id instance-attribute

organization_id

The organization id in the authentication service.

project_id class-attribute instance-attribute

project_id = None

The project id in the authentication service.

trace_to_datadog

trace_to_datadog()

Add project_id, and organization_id tags to the current root span.

Source code in application_kit/authenticator/types.py

def trace_to_datadog(self) -> None:
    """Add project_id, and organization_id tags to the current root span."""
    if self.organization_id:
        set_root_span_tag("organization_id", str(self.organization_id))
    if self.project_id:
        set_root_span_tag("project_id", str(self.project_id))

ProjectTokenContext

Context object that avoids to put too much things on the request objects.

should_count class-attribute instance-attribute

should_count = True

If the current request should be counted. Internal requests (eg. request Host port is 8000) are excluded from metrics counters.

source class-attribute instance-attribute

source = None

The source pulled from either X-SDK-Source header or if ai-context if AI-Context header is set to true

ProjectTokenModel

ProjectTokenModel(token_instance, reference=None)

This is the main objects to interact with when using authentication.

PARAMETER	DESCRIPTION
token_instance	The token instance retrived from authentication service using private key TYPE: AllTokensModel
reference	The project reference ids. TYPE: ProjectReference | None DEFAULT: None

Source code in application_kit/authenticator/types.py

def __init__(
    self,
    token_instance: Annotated[
        AllTokensModel, Doc("The token instance retrived from authentication service using private key")
    ],
    reference: Annotated[ProjectReference | None, Doc("The project reference ids.")] = None,
) -> None:
    self.instance = token_instance
    self.reference = reference
    self.context = ProjectTokenContext()

instance instance-attribute

instance = token_instance

The token instance.

reference instance-attribute

reference = reference

The project reference.

context instance-attribute

context = ProjectTokenContext()

The current context, contains source and should_count.

products property

products

Returns the products found in the token.

Warn

User tokens have no products.

check_product_allowed

check_product_allowed(product)

Raises ProductNotAllowed

Source code in application_kit/authenticator/types.py

def check_product_allowed(self, product: Products | None) -> None:
    """
    Raises ProductNotAllowed
    """
    if product is not None and self.products is not None and product not in self.products:
        raise ProductNotAllowed(product.value)

validate_products

validate_products(value)

Validates a product list, ignores unknown products

Source code in application_kit/authenticator/types.py

def validate_products(value: list[str]) -> list[Products]:
    """Validates a product list, ignores unknown products"""
    products = []
    for product in value:
        try:
            products.append(Products[product])
        except KeyError as e:
            logger.debug(f"Unsupported product {e}")
    return products