Security

Authentication, TLS, user management, and the FastAPI server framework.

Authentication

SCRAM-SHA-256 and PLAINTEXT client authentication handshake for the Haystack HTTP protocol.

Haystack authentication handshake.

Implements the Project Haystack authentication protocol including SCRAM-SHA-256 and PLAINTEXT mechanisms.

Uses the cryptography library for all key derivation, HMAC, and hashing.

See: https://project-haystack.org/doc/docHaystack/Auth

async hs_py.auth.authenticate(session, base_url, username, password)

Run the Haystack auth handshake and return a bearer auth token.

Attempts SCRAM-SHA-256 first, falling back to PLAINTEXT if offered.

Parameters:

• session (ClientSession) – An open aiohttp client session.

• base_url (str) – Haystack server base URL (e.g. http://host/api/).

• username (str) – Username for authentication.

• password (str) – Password for authentication.

Return type:

str

Returns:

Bearer auth token string for subsequent requests.

Raises:

AuthError – On authentication failure.

hs_py.auth.scram_client_final(password, first, server_first_msg, hash_name='SHA-256')

Build the SCRAM client-final message from the server-first response.

Parameters:

• password (str) – The plaintext password.

• first (ScramClientFirst) – The ScramClientFirst from scram_client_first().

• server_first_msg (str) – Raw server-first message (r=...,s=...,i=...).

• hash_name (str (default: 'SHA-256')) – Hash algorithm name (default SHA-256).

Return type:

ScramClientFinal

Returns:

ScramClientFinal with the proof message.

Raises:

AuthError – On nonce mismatch or unsupported hash.

hs_py.auth.scram_client_first(username)

Build the SCRAM client-first message.

Parameters:

username (str) – The username to authenticate as.

Return type:

ScramClientFirst

Returns:

ScramClientFirst with the message and nonce.

Auth Types

Authentication protocol interfaces and credential backends.

Server-side authentication types.

Provides credential storage and authenticator protocols for Haystack server implementations. Framework-independent — used by both aiohttp and FastAPI servers.

class hs_py.auth_types.Authenticator(*args, **kwargs)

Bases: Protocol

Protocol for server-side authentication.

async scram_credentials(username)

Return SCRAM credentials for a user, or None if unknown.

Return type:

ScramCredentials | None

Parameters:

username (str)

class hs_py.auth_types.CertAuthenticator(allowed_cns)

Bases: object

Authenticator that authorizes clients by TLS client certificate CN.

When mutual TLS is enabled, the server can extract the Common Name (CN) from the client certificate and check it against an allowed set. This bypasses SCRAM entirely for certificate-authenticated connections.

Parameters:

allowed_cns (set[str]) – Set of Common Name strings that are authorized.

authorize(peercert)

Check a peer certificate and return the username if authorized.

Parameters:

peercert (dict[str, object] | None) – Certificate dict from ssl.SSLSocket.getpeercert().

Return type:

str | None

Returns:

The CN string if authorized, or None if not.

class hs_py.auth_types.ScramCredentials(salt, iterations, stored_key, server_key)

Bases: object

Pre-computed SCRAM-SHA-256 credentials for a user.

Parameters:

• salt (bytes)

• iterations (int)

• stored_key (bytes)

• server_key (bytes)

salt: bytes

Random salt used during key derivation.

iterations: int

PBKDF2 iteration count.

stored_key: bytes

H(ClientKey) — used to verify client proof.

server_key: bytes

HMAC(SaltedPassword, “Server Key”) — used to compute server signature.

class hs_py.auth_types.SimpleAuthenticator(users, *, iterations=600000)

Bases: object

Authenticator that derives SCRAM keys from a username→password dict.

Parameters:

• users (dict[str, str])

• iterations (int)

async scram_credentials(username)

Return SCRAM credentials for a user, or None if unknown.

Parameters:

username (str) – The username to look up.

Return type:

ScramCredentials | None

Returns:

ScramCredentials or None.

class hs_py.auth_types.StorageAuthenticator(user_store)

Bases: object

Authenticator that reads SCRAM credentials from a UserStore.

Returns None for disabled or missing users, preventing login.

Parameters:

user_store (UserStore) – The user store to read credentials from.

async scram_credentials(username)

Return SCRAM credentials for a user, or None if unknown/disabled.

Parameters:

username (str) – The username to look up.

Return type:

ScramCredentials | None

Returns:

ScramCredentials or None.

Users & Roles

User model, role-based permissions, and user creation helpers.

User model and helpers for Haystack server user management.

Provides a frozen User dataclass, a Role enum for role-based access control, and a create_user() factory that derives SCRAM-SHA-256 credentials from a plaintext password. Passwords are never stored — only the derived ScramCredentials.

class hs_py.user.Role(*values)

Bases: Enum

User role for permission enforcement.

Roles form a strict hierarchy: ADMIN > OPERATOR > VIEWER.

• VIEWER — read-only Haystack ops (read, nav, hisRead, defs, …).

• OPERATOR — read + write ops (hisWrite, pointWrite, invokeAction, watches).

• ADMIN — full access including user management.

VIEWER = 'viewer'

OPERATOR = 'operator'

ADMIN = 'admin'

property level: int

Return numeric privilege level (higher = more access).

class hs_py.user.User(username, credentials, first_name='', last_name='', email='', role=Role.VIEWER, enabled=True, created_at=<factory>, updated_at=<factory>)

Bases: object

Immutable user record.

Passwords are stored as pre-computed ScramCredentials — the plaintext is never retained.

Parameters:

• username (str) – Unique login identifier (required).

• credentials (ScramCredentials) – SCRAM-SHA-256 credentials derived from the password.

• first_name (str (default: '')) – User’s first name.

• last_name (str (default: '')) – User’s last name.

• email (str (default: '')) – User’s email address.

• role (Role (default: <Role.VIEWER: 'viewer'>)) – Permission role (ADMIN, OPERATOR, or VIEWER).

• enabled (bool (default: True)) – Whether the user can log in.

• created_at (float (default: <factory>)) – Monotonic creation timestamp.

• updated_at (float (default: <factory>)) – Monotonic last-update timestamp.

username: str

credentials: ScramCredentials

first_name: str

last_name: str

email: str

role: Role

enabled: bool

created_at: float

updated_at: float

hs_py.user.create_user(username, password, *, first_name='', last_name='', email='', role=Role.VIEWER, enabled=True, iterations=600000)

Create a User with derived SCRAM credentials.

Parameters:

• username (str) – Unique login identifier.

• password (str) – Plaintext password (used to derive credentials, then discarded).

• first_name (str (default: '')) – User’s first name.

• last_name (str (default: '')) – User’s last name.

• email (str (default: '')) – User’s email address.

• role (Role (default: <Role.VIEWER: 'viewer'>)) – Permission role (default Role.VIEWER).

• enabled (bool (default: True)) – Whether the user can log in.

• iterations (int (default: 600000)) – PBKDF2 iteration count.

Return type:

User

Returns:

Frozen User instance.

hs_py.user.user_from_dict(d)

Deserialize a User from a plain dict.

Inverse of user_to_dict().

Return type:

User

Parameters:

d (dict[str, Any])

hs_py.user.user_to_dict(user)

Serialize a User to a plain dict for storage.

Credential bytes are base64-encoded for JSON compatibility.

Return type:

dict[str, Any]

Parameters:

user (User)

Bootstrap

Admin user bootstrap on server startup.

Admin user bootstrap for Haystack servers.

On startup, verifies that at least one enabled admin user exists in the UserStore. If none is found, attempts to seed one from environment variables HS_SUPERUSER_USERNAME and HS_SUPERUSER_PASSWORD. Exits with an error if neither source provides an admin.

async hs_py.bootstrap.ensure_superuser(store)

Ensure at least one enabled admin user exists.

Check order:

1. Query the store for any enabled admin — if found, return.

2. Read HS_SUPERUSER_USERNAME and HS_SUPERUSER_PASSWORD from environment variables.

3. If both are set, create a new admin user and persist it.

4. If either is missing, exit the process with an error message.

Parameters:

store (UserStore) – User store backend to check and seed.

Return type:

None

TLS

TLS configuration helpers for client and server SSL contexts, including mutual TLS (mTLS) and test certificate generation.

TLS 1.3 configuration and SSL context builders.

Provides TLSConfig for declaring certificate paths and helper functions for constructing ssl.SSLContext instances that enforce TLS 1.3 minimum.

System CA trust is deliberately excluded — only explicitly configured CA certificates are loaded.

Uses the cryptography library for test certificate generation.

class hs_py.tls.TLSConfig(certificate_path=None, private_key_path=None, ca_certificates_path=None, key_password=None)

Bases: object

TLS 1.3 configuration for Haystack client and server.

Parameters:

• certificate_path (str | None (default: None)) – Path to PEM certificate file.

• private_key_path (str | None (default: None)) – Path to PEM private key file.

• ca_certificates_path (str | None (default: None)) – Colon-separated paths to CA PEM files or directories.

• key_password (bytes | str | None (default: None)) – Passphrase for the private key (bytes or str).

certificate_path: str | None

Path to PEM certificate file.

private_key_path: str | None

Path to PEM private key file.

ca_certificates_path: str | None

Colon-separated paths to CA PEM files or directories.

key_password: bytes | str | None

Passphrase for the private key (bytes or str).

hs_py.tls.build_client_ssl_context(config)

Build a TLS 1.3 client SSL context.

PROTOCOL_TLS_CLIENT enables hostname verification and requires server certificates by default.

Parameters:

config (TLSConfig) – TLS configuration with certificate paths.

Return type:

SSLContext

Returns:

Configured ssl.SSLContext.

hs_py.tls.build_server_ssl_context(config)

Build a TLS 1.3 server SSL context with mutual authentication.

verify_mode = CERT_REQUIRED enforces client certificate verification.

Parameters:

config (TLSConfig) – TLS configuration with certificate paths.

Return type:

SSLContext

Returns:

Configured ssl.SSLContext.

hs_py.tls.extract_peer_cn(peercert)

Extract the Common Name (CN) from a peer certificate dict.

Parameters:

peercert (dict[str, object] | None) – Certificate dict as returned by ssl.SSLSocket.getpeercert().

Return type:

str | None

Returns:

The CN string, or None if not present.

hs_py.tls.extract_peer_sans(peercert)

Extract Subject Alternative Names from a peer certificate dict.

Parameters:

peercert (dict[str, object] | None) – Certificate dict as returned by ssl.SSLSocket.getpeercert().

Return type:

list[str]

Returns:

List of SAN values (DNS names and IP addresses).

hs_py.tls.generate_test_certificates(directory)

Generate a self-signed CA and server/client certificates for testing.

Uses EC P-256 keys with SHA-256 signatures. Certificates are valid for one year from generation time. Writes PEM files to directory:

• ca.pem — CA certificate

• server.pem / server.key — Server certificate and private key

• client.pem / client.key — Client certificate and private key

Parameters:

directory (str | Path) – Directory to write certificate files into.

Return type:

TLSConfig

Returns:

A TLSConfig pointing to the server certificate and CA.

FastAPI Server

FastAPI-based Haystack HTTP server with SCRAM authentication middleware, content negotiation, and WebSocket endpoint.

FastAPI-based Haystack HTTP server.

Provides a FastAPI application factory with content-negotiated Haystack routes, SCRAM-SHA-256 authentication middleware, and WebSocket support.

See: https://project-haystack.org/doc/docHaystack/HttpApi

class hs_py.fastapi_server.ScramAuthMiddleware(app, authenticator, auth_tokens=None)

Bases: object

SCRAM-SHA-256 authentication middleware for FastAPI.

Pure ASGI middleware — avoids BaseHTTPMiddleware overhead and streaming issues. Implements the Haystack SCRAM handshake:

1. Client sends Authorization: HELLO username=<b64>

2. Server returns 401 with WWW-Authenticate: SCRAM handshakeToken=...

3. Client sends Authorization: SCRAM handshakeToken=..., data=<client-first>

4. Server returns 401 with server-first message

5. Client sends Authorization: SCRAM handshakeToken=..., data=<client-final>

6. Server returns 200 with Authentication-Info: authToken=..., data=<server-final>

7. Subsequent requests use Authorization: BEARER authToken=...

Parameters:

• app (Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]]) – The ASGI application to wrap.

• authenticator (Authenticator) – Server-side credential store.

• auth_tokens (dict[str, TokenEntry] | None (default: None)) – Shared token dict (also used by the WS endpoint).

property tokens: dict[str, TokenEntry]

Expose the token store (read-only access for tests).

hs_py.fastapi_server.create_fastapi_app(ops=None, *, storage=None, authenticator=None, namespace=None, user_store=None, prefix='/api', cors_origins=None, default_format='json')

Create a FastAPI application with Haystack HTTP routes.

The returned app supports content-negotiated responses (JSON, Zinc, Trio, CSV, Turtle, JSON-LD), SCRAM-SHA-256 authentication (when an authenticator is provided), and WebSocket connections at {prefix}/ws.

Parameters:

• ops (HaystackOps | None (default: None)) – HaystackOps implementation to dispatch to. When None and storage is provided, a default HaystackOps is constructed automatically.

• storage (StorageAdapter | None (default: None)) – Optional StorageAdapter backend. Its start() / close() lifecycle methods are called automatically via the FastAPI lifespan.

• authenticator (Authenticator | None (default: None)) – Optional Authenticator for SCRAM-SHA-256 auth. When None, all requests are accepted without auth.

• namespace (Namespace | None (default: None)) – Optional Namespace for defs and libs operations.

• user_store (UserStore | None (default: None)) – Optional UserStore for user management. When provided, user CRUD endpoints are mounted under {prefix}/users/ and superuser bootstrapping runs at startup.

• prefix (str (default: '/api')) – URL path prefix for all Haystack routes (default "/api").

• cors_origins (list[str] | None (default: None)) – Optional list of allowed CORS origins. When provided, CORSMiddleware is added with credentials support. Example: ["http://localhost:3000", "https://app.example.com"].

• default_format (str (default: 'json')) – Default response format when the client sends */* or no Accept header. The Haystack spec defines "zinc" as the default; this library defaults to "json" for ecosystem compatibility. Supported values: "json", "zinc".

Return type:

FastAPI

Returns:

Configured fastapi.FastAPI application.

Example:

from hs_py.fastapi_server import create_fastapi_app
from hs_py.storage.memory import InMemoryAdapter
from hs_py.auth_types import StorageAuthenticator

storage = InMemoryAdapter()
auth = StorageAuthenticator(storage)
app = create_fastapi_app(storage=storage, authenticator=auth, user_store=storage)
# uvicorn.run(app, host="0.0.0.0", port=8080)