Implementation:Neuml Txtai Authorization Init
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:
- Checks if the header is present (not
None) - Computes the SHA-256 hash of the provided token
- Compares the hash against the stored token hash
- 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:
- If the header starts with
"Bearer ", the prefix is stripped - The remaining token string is UTF-8 encoded and hashed with SHA-256
- 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
TOKENis set, authentication is enabled globally for all endpoints - If
TOKENis not set, no authentication is applied - The stored
tokenvalue is the SHA-256 hash -- not the plaintext secret
See Also
- Neuml_Txtai_API_Security - Principle behind token-based API security
- Neuml_Txtai_API_Create - How
Authorizationis registered during API creation - Neuml_Txtai_REST_API_Endpoints - The endpoints protected by this authorization