Authentication & Authorization

Overview

The Authentication and Authorization module provides a flexible, policy-based system for securing API endpoints. It is designed to clearly separate the concepts of Authentication (verifying who a user or service is) and Authorization (determining what they are allowed to do).

Key Features

Policy-Based: Access control is defined by simple, named policies (e.g., "public", "private", "internal").
Multiple Mechanisms: Out-of-the-box support for common authentication mechanisms like JWT Bearer tokens and API Keys (both shared and per-client).
Decoupled Architecture: A clean separation between Policies (Executors), which define what access is required, and Authentication Providers, which handle how credentials are validated.
Extensible: New authentication providers or authorization policies can be added by implementing simple protocols and registering them.

Core Concepts

Policies (Executors)

An Authorization Policy is a named rule that defines the security requirements for an endpoint. For example, a "private" policy might require a valid JWT, while an "internal" policy might require a specific IP address.

In Athomic, policies are implemented by classes that follow the AuthExecutorProtocol. The AuthService uses a central registry to look up the correct executor for a given policy name.

Authentication Providers

An Authentication Provider is a component that performs the actual credential validation. For example, the JWTAuth provider knows how to decode a JWT, verify its signature, and check its expiration. The APIKeyDBProvider knows how to look up an API key's hash in a database.

All providers implement the AuthProviderProtocol, allowing the executors to be decoupled from the specific validation logic.

`AuthService`

The AuthService is the central orchestrator that ties everything together. When an endpoint needs to be secured, you ask the AuthService to authorize the request against a specific policy. The service finds the correct executor for that policy and runs it, which in turn uses the appropriate authentication provider.

Available Policies (Executors)

Athomic comes with a set of pre-configured policies:

public: Allows anonymous, unrestricted access.
private: Requires a valid JWT Bearer token in the Authorization header.
shared_api_key: Requires a valid, globally shared API key in the configured header.
db_api_key: Requires a valid, per-client API key that is looked up in the database.
internal: Requires the request's IP address to be on a configured allowlist.

How to Secure an Endpoint

Securing an endpoint is typically done at the web framework layer, for example, using a FastAPI dependency. The dependency would be responsible for extracting credentials from the request, building a SecurityRequestContext, and calling the AuthService.

Conceptual FastAPI Example

from fastapi import Depends, HTTPException, Request
from nala.athomic.security.auth import AuthService, SecurityRequestContext
from nala.athomic.security.exceptions import AthomicAuthError, AthomicForbiddenError

auth_service = AuthService() # Singleton instance

async def require_private_access(request: Request):
    """A FastAPI dependency that enforces the 'private' policy."""

    context = SecurityRequestContext(
        client_host=request.client.host,
        headers=dict(request.headers)
    )

    try:
        # Ask the AuthService to authorize the request against the "private" policy
        authenticated_user = await auth_service.authorize(context, "private")
        return authenticated_user
    except AthomicAuthError as e:
        raise HTTPException(status_code=401, detail=str(e))
    except AthomicForbiddenError as e:
        raise HTTPException(status_code=403, detail=str(e))

@router.get("/profile")
async def get_user_profile(user: AuthenticatedUser = Depends(require_private_access)):
    # This endpoint is now protected. The 'user' object contains the
    # validated claims from the JWT.
    return {"user_id": user.sub, "scopes": user.scopes}

Configuration

Authentication is configured under the [security.auth] section in your settings.toml.

[default.security.auth]
# A master switch to enable or disable authentication.
enable = true

# The default backend to use. Not always necessary as policies are explicit.
backend = "jwt"

  # Configuration for JWT-based authentication.
  [default.security.auth.jwt]
  # The secret key for signing tokens. Best practice is to use a secret reference.
  secret_key = { path = "auth/jwt", key = "secret" }
  algorithm = "HS256"
  expiration_minutes = 60

  # Configuration for API Key-based authentication.
  [default.security.auth.api_key]
  enabled = true
  header_name = "X-API-KEY"

  # The secret for the "shared_api_key" policy.
  secret_key = { path = "auth/api-key", key = "shared-secret" }

  # The database connection for the "db_api_key" policy.
  document_db_connection_name = "default_mongo"

Authorization providers — RBAC and ReBAC (switchable by one setting)

Authorization (deciding what a principal may do) is served by a pluggable AuthorizationProvider port. Two engines are built in and interchangeable via a single setting — authorization.provider — with zero call-site changes:

rbac (default): the role→permission engine (AuthorizationService) wrapped behind the port. A permission resource:action is granted when the principal's scopes/roles (roles expand via the resolver) match it under the wildcard rule.
rebac: a Zanzibar-style relation-tuple engine. Access is a relationship fact — RelationTuple(subject, relation, resource) — and the model computes a relation via direct grants, union/computed usersets, and tuple-to-userset hierarchy (e.g. an admin of the parent org is an editor of the workspace). Default-deny. Backed by an infra-free embedded store (in-memory / KV) or by OpenFGA (store = "openfga", recommended for prod).

The port exposes three questions, both engines answer the first:

check(subject, relation, resource, context=None) -> Decision — allow/deny (+ reason).
list_objects(subject, relation, resource_type) -> list[str] — "what can X see?".
expand(relation, resource) -> ExpandTree — "who has access / why?".

list_objects/expand are relationship questions RBAC cannot answer; the RBAC provider raises NotSupportedError there rather than returning a misleading result.

mode (disabled | dry_run | enforce) is orthogonal to provider and applies identically to both engines — a consumer can roll a new provider out in dry_run (metering authz_decisions_total{result="would_deny", provider=…}) before enforcing.

The requires(...) route guard and get_authorization_provider() are provider-agnostic: flipping rbac ↔ rebac is a pure settings change.

[default.security.auth.authorization]
provider = "rbac"          # rbac | rebac | abac | allow_all
mode = "dry_run"           # disabled | dry_run | enforce (orthogonal to provider)

  # RBAC: role → permission strings (wildcards allowed)
  [default.security.auth.authorization.roles]
  admin = ["*:*"]

  # ReBAC (read only when provider = "rebac")
  [default.security.auth.authorization.rebac]
  store = "embedded"       # embedded (no infra) | openfga (prod)
    [default.security.auth.authorization.rebac.openfga]
    api_url = "http://localhost:8080"
    store_id = "..."
    model_id = "..."

A ReBAC deployment injects its AuthorizationModel + TupleStore at the composition root (like the RBAC kv_store); the provider surfaces write_tuples/delete_tuples/ read_tuples so a consumer syncs tuples from its own domain events — athomic consumes none. A local OpenFGA server for dev/CI is available via make openfga-up (behind the rebac provider only; not required for RBAC deployments).

API Reference

`nala.athomic.security.auth.service.AuthService`

The central authorization service.

This class acts as an Orchestrator, delegating the specific policy execution logic to the appropriate AuthExecutorProtocol instance, which is retrieved through a standardized registry. It integrates metrics for all policy check attempts.

`init(registry=None)`

Initializes the service with a specific executor registry.

Parameters:

Name	Type	Description	Default
`registry`	`Optional[AuthExecutorRegistry]`	An optional registry instance for dependency injection (DI). Defaults to the global singleton.	`None`

`authorize(context, policy_name)` `async`

Finds the correct executor in the registry and executes the authorization check.

This method encapsulates the authorization flow: it records the attempt, retrieves the executor, executes the check, and logs the final success or failure.

Parameters:

Name	Type	Description	Default
`context`	`SecurityRequestContext`	The request context containing headers and client information.	required
`policy_name`	`str`	The name of the authorization policy to enforce (e.g., 'private', 'db_api_key').	required

Returns:

Name	Type	Description
`AuthenticatedUser`	`AuthenticatedUser`	A user object representing the authenticated identity.

Raises:

Type	Description
`ValueError`	If no executor is registered for the specified `policy_name`.
`Exception`	Propagates the exception raised by the executor upon authorization failure.

`nala.athomic.security.auth.executors.protocols.AuthExecutorProtocol`

Bases: Protocol

Defines the contract for an authorization execution strategy.

Each executor is responsible for enforcing a single, specific form of validation (e.g., JWT token check, shared API key validation, IP allowlist check), adhering to the Single Responsibility Principle (SRP).

`execute(context)` `async`

Executes the authorization logic against the provided request context.

Parameters:

Name	Type	Description	Default
`context`	`SecurityRequestContext`	The request context containing client host and headers necessary for validation.	required

Returns:

Name	Type	Description
`AuthenticatedUser`	`AuthenticatedUser`	An object representing the authenticated identity and its claims.

Raises:

Type	Description
`AthomicAuthError`	If authentication fails (e.g., invalid token, expired, revoked).
`AthomicForbiddenError`	If access is denied (e.g., IP not allowed, insufficient scope).

`nala.athomic.security.auth.protocol.AuthProviderProtocol`

Bases: Protocol

`check(token)` `async`

Verify a token and return its decoded payload if valid.

:param token: The token to verify. :return: A dictionary with the decoded payload. :raises: AuthenticationError or other relevant exception if invalid.

`create_token(subject, scope=None)`

Create a signed token for the given subject.

:param subject: Unique identifier for the token owner (e.g., user ID). :param scope: Optional scope string for authorization. :return: A JWT or compatible token as a string.

`nala.athomic.security.auth.authz.provider.AuthorizationProvider`

Bases: Protocol

Provider-agnostic authorization contract (RBAC, ReBAC, and later ABAC/allow_all).

`check(subject, relation, resource, context=None)` `async`

Decide whether subject holds relation on resource (default-deny).

Parameters:

Name	Type	Description	Default
`subject`	`str`	The subject URN (e.g. `nala:user:alice`).	required
`relation`	`str`	The relation/permission token (e.g. `editor`, `features:promote`).	required
`resource`	`str`	The resource URN (e.g. `nala:workspace:w1`).	required
`context`	`Optional[Mapping[str, object]]`	Optional extra inputs a provider may need (e.g. the RBAC provider reads the `AuthenticatedUser` principal here to expand its roles).	`None`

Returns:

Name	Type	Description
`A`	`Decision`	class:`Decision` (never a bare bool), carrying the allow/deny + reason.

`expand(relation, resource)` `async`

Explain who holds relation on resource (the userset tree).

The explainability primitive ("why / who has access?"). May raise :class:NotSupportedError where an engine cannot expand.

`list_objects(subject, relation, resource_type)` `async`

List the resource URNs of resource_type on which subject holds relation.

The list-filtering primitive ("what can X see?"). May raise :class:NotSupportedError for engines that cannot enumerate instances.

`nala.athomic.security.auth.authz.provider_factory.build_authorization_provider(settings, *, registry=None, kv_store=None, tuple_store=None, rebac_model=None, openfga_client=None)`

Build the provider selected by settings.provider (default-deny by construction).

Parameters:

Name	Type	Description	Default
`settings`	`AuthorizationSettings`	Authorization settings (`provider` selects the engine).	required
`registry`	`Optional[AuthorizationProviderRegistry]`	Provider-builder registry; defaults to the module-level one.	`None`
`kv_store`	`Optional[KVStoreProtocol]`	KV client for RBAC's `kv` resolver / KV-backed tuple store.	`None`
`tuple_store`	`Optional[TupleStore]`	ReBAC tuple store (embedded); defaults to in-memory.	`None`
`rebac_model`	`Optional[AuthorizationModel]`	ReBAC authorization model (injected at the composition root).	`None`
`openfga_client`	`Optional[object]`	Pre-built OpenFGA client (for the `openfga` store / testing).	`None`

Raises:

Type	Description
`ValueError`	If the selected provider is unknown or missing a required dependency.

`nala.athomic.security.auth.authz.rebac.tuples.RelationTuple`

Bases: BaseModel

An atomic ReBAC fact: subject has relation on resource.

Attributes:

Name	Type	Description
`subject`	`str`	Subject URN (`nala:user:alice`) or an object URN when it names a userset together with `subject_relation` (`nala:org:o1` + `member`).
`relation`	`str`	The relation token this tuple grants (`editor`, `parent`, `member`).
`resource`	`str`	The resource URN the relation is on (`nala:workspace:w1`).
`subject_relation`	`Optional[str]`	Optional userset relation on `subject` — when set, this tuple grants the relation to the members of `subject`'s `subject_relation` set, enabling group-to-group / nested-group rewrites.

`key()`

A hashable identity for de-duplication.

`nala.athomic.security.auth.authz.rebac.model.AuthorizationModel`

Bases: BaseModel

The full model: object type → its relation definitions.

`rewrite_for(object_type, relation)`

Return the rewrite for object_type/relation.

Raises:

Type	Description
`KeyError`	If the type or relation is undefined — the engine treats this as a default-deny (an undefined relation grants nothing).

`nala.athomic.security.auth.models.auth_user.AuthenticatedUser`

Bases: BaseModel

Represents an authenticated identity extracted from a validated token payload (e.g., JWT).

This model provides a normalized view of the authenticated user's claims, used throughout the application for authorization decisions.

Attributes:

Name	Type	Description
`sub`	`str`	The unique identifier for the user or entity (Subject) that the token represents.
`scopes`	`List[str]`	A list of normalized (lowercase, split) user permission scopes.
`roles`	`List[str]`	A list of roles assigned to the user.
`jti`	`str`	The unique ID of the token (JWT ID), crucial for token denylist/revocation checks.
`exp`	`int`	The timestamp (UNIX epoch) when the token expires.
`raw_scope`	`str`	The original, unmodified scope string from the token payload.
`tenant_id`	`Optional[str]`	The tenant the principal belongs to, when the token/API-key payload carries a `tenant_id` claim. Drives tenant-scoped execution context (multi-tenancy) downstream.

`from_payload(payload)` `classmethod`

Creates an AuthenticatedUser instance from a raw decoded token payload dictionary.

This factory method handles normalization, such as splitting the raw scope string into a list of lowercase scopes.

Parameters:

Name	Type	Description	Default
`payload`	`Dict[str, Any]`	The decoded token payload (dictionary).	required

Returns:

Name	Type	Description
`AuthenticatedUser`	`AuthenticatedUser`	A validated instance of the user model.

Authentication & Authorization

Overview

Key Features

Core Concepts

Policies (Executors)

Authentication Providers

AuthService

Available Policies (Executors)

How to Secure an Endpoint

Conceptual FastAPI Example

Configuration

Authorization providers — RBAC and ReBAC (switchable by one setting)

API Reference

nala.athomic.security.auth.service.AuthService

__init__(registry=None)

authorize(context, policy_name) async

nala.athomic.security.auth.executors.protocols.AuthExecutorProtocol

execute(context) async

nala.athomic.security.auth.protocol.AuthProviderProtocol

check(token) async

create_token(subject, scope=None)

nala.athomic.security.auth.authz.provider.AuthorizationProvider

check(subject, relation, resource, context=None) async

expand(relation, resource) async

list_objects(subject, relation, resource_type) async

nala.athomic.security.auth.authz.provider_factory.build_authorization_provider(settings, *, registry=None, kv_store=None, tuple_store=None, rebac_model=None, openfga_client=None)

nala.athomic.security.auth.authz.rebac.tuples.RelationTuple

key()

nala.athomic.security.auth.authz.rebac.model.AuthorizationModel

rewrite_for(object_type, relation)

nala.athomic.security.auth.models.auth_user.AuthenticatedUser

from_payload(payload) classmethod

`AuthService`

`nala.athomic.security.auth.service.AuthService`

`init(registry=None)`

`authorize(context, policy_name)` `async`

`nala.athomic.security.auth.executors.protocols.AuthExecutorProtocol`

`execute(context)` `async`

`nala.athomic.security.auth.protocol.AuthProviderProtocol`

`check(token)` `async`

`create_token(subject, scope=None)`

`nala.athomic.security.auth.authz.provider.AuthorizationProvider`

`check(subject, relation, resource, context=None)` `async`

`expand(relation, resource)` `async`

`list_objects(subject, relation, resource_type)` `async`

`nala.athomic.security.auth.authz.provider_factory.build_authorization_provider(settings, *, registry=None, kv_store=None, tuple_store=None, rebac_model=None, openfga_client=None)`

`nala.athomic.security.auth.authz.rebac.tuples.RelationTuple`

`key()`

`nala.athomic.security.auth.authz.rebac.model.AuthorizationModel`

`rewrite_for(object_type, relation)`

`nala.athomic.security.auth.models.auth_user.AuthenticatedUser`

`from_payload(payload)` `classmethod`