Jump to content

Connect SuperML | Leeroopedia MCP: Equip your AI agents with best practices, code verification, and debugging knowledge. Powered by Leeroo — building Organizational Superintelligence. Contact us at founders@leeroo.com.

Implementation:Neuml Txtai Authorization Init

From Leeroopedia


Overview

The Authorization class provides basic token-based authentication for txtai API endpoints. It implements SHA-256 token hashing and integrates with FastAPI's dependency injection system to enforce authorization on every incoming request. When a valid token is not provided, the class raises an HTTP 401 Unauthorized exception.

API Signature

Authorization.__init__

from txtai.api import Authorization

auth = Authorization(token=None)
Parameter Type Default Description
token str or None None SHA-256 hash of the expected authorization token. If None, falls back to the TOKEN environment variable.

Returns: An Authorization callable instance suitable for use as a FastAPI dependency.

Authorization.__call__

auth(authorization=Header(default=None))
Parameter Type Default Description
authorization str or None None The value of the HTTP Authorization header, injected by FastAPI.

Returns: None (succeeds silently on valid authorization).

Raises: HTTPException(status_code=401) if the token is missing or invalid.

Authorization.digest

hash_value = auth.digest(authorization)
Parameter Type Default Description
authorization str required The raw authorization header value (e.g., "Bearer my-token").

Returns: str -- the SHA-256 hex digest of the token portion of the header.

Source Reference

File: src/python/txtai/api/authorization.py (Lines 11-53)

Complete Implementation

import hashlib
import os

from fastapi import Header, HTTPException


class Authorization:
    """
    Basic token authorization.
    """

    def __init__(self, token=None):
        self.token = token if token else os.environ.get("TOKEN")

    def __call__(self, authorization: str = Header(default=None)):
        if not authorization or self.token != self.digest(authorization):
            raise HTTPException(status_code=401, detail="Invalid Authorization Token")

    def digest(self, authorization):
        # Replace Bearer prefix
        prefix = "Bearer "
        token = authorization[len(prefix):] if authorization.startswith(prefix) else authorization

        # Compute SHA-256 hash
        return hashlib.sha256(token.encode("utf-8")).hexdigest()

Detailed Behavior

Constructor (__init__)

The constructor accepts an optional token parameter. If not provided, it reads from the TOKEN environment variable. This token must be the SHA-256 hex digest of the actual secret token -- not the plaintext token itself.

# The stored token is a SHA-256 hash, not plaintext
self.token = token if token else os.environ.get("TOKEN")

Callable Validation (__call__)

When FastAPI invokes this dependency, it automatically extracts the Authorization HTTP header and passes it as the authorization parameter. The method:

  1. Checks if the header is present (not None)
  2. Computes the SHA-256 hash of the provided token
  3. Compares the hash against the stored token hash
  4. Raises HTTP 401 if validation fails
if not authorization or self.token != self.digest(authorization):
    raise HTTPException(status_code=401, detail="Invalid Authorization Token")

Token Hashing (digest)

The digest method handles both raw tokens and Bearer-prefixed tokens:

  1. If the header starts with "Bearer ", the prefix is stripped
  2. The remaining token string is UTF-8 encoded and hashed with SHA-256
  3. The hex digest string is returned for comparison

Usage Examples

Setting Up Token Auth via Environment Variable

# Generate SHA-256 hash of your token
export TOKEN=$(echo -n "my-secret-token" | sha256sum | awk '{print $1}')

# Start the API server
CONFIG=config.yml uvicorn "txtai.api:app"

Making Authenticated Requests

# Include Bearer token in requests
curl -H "Authorization: Bearer my-secret-token" \
     "http://localhost:8000/search?query=test"

Using in Python Client Code

import requests

headers = {"Authorization": "Bearer my-secret-token"}
response = requests.get(
    "http://localhost:8000/search",
    params={"query": "machine learning"},
    headers=headers
)

Programmatic Registration

from fastapi import Depends, FastAPI
from txtai.api import Authorization

# Create with explicit token hash
token_hash = "a1b2c3..."  # SHA-256 hash of secret
app = FastAPI(dependencies=[Depends(Authorization(token_hash))])

Error Responses

HTTP Status Detail Condition
401 "Invalid Authorization Token" Missing Authorization header
401 "Invalid Authorization Token" Token hash does not match stored hash

Integration with API Startup

The Authorization class is registered during API creation in src/python/txtai/api/application.py:

token = os.environ.get("TOKEN")
if token:
    dependencies.append(Depends(Authorization(token)))

This means:

  • If TOKEN is set, authentication is enabled globally for all endpoints
  • If TOKEN is not set, no authentication is applied
  • The stored token value is the SHA-256 hash -- not the plaintext secret

See Also

Requires Environment

Page Connections

Double-click a node to navigate. Hold to expand connections.
Principle
Implementation
Heuristic
Environment