Features

• JWT Authentication -- Access tokens (HS256) with short-lived expiry and opaque refresh tokens with rotation, family tracking, and reuse detection
• TOTP MFA -- Time-based one-time passwords with Fernet-encrypted secrets and single-use recovery codes
• WebAuthn / Passkeys -- Registration and authentication via the WebAuthn standard
• RBAC -- Role-based access control with granular permissions
• API Keys -- Prefix-based lookup, scoped keys with optional expiry
• Password Security -- Configurable complexity rules, password history enforcement, and HIBP breach checking via k-anonymity
• Pluggable Storage -- Sync and async SQLite backends included; implement StorageInterface or AsyncStorageInterface for any database
• Event Bus -- Typed lifecycle events (login, MFA, token refresh, RBAC changes, etc.) with sync and async handler support
• Audit Logging -- Append-only audit trail with optional chain hashing for tamper evidence
• Framework-Agnostic -- Works with FastAPI, Flask, Django, or standalone scripts
• Full Async Support -- AsyncAuthManager with AsyncSQLiteStorage for async-first applications

Requirements

• Python 3.10+
• SQLite (included in Python standard library)

Installation

pip install authority-auth

For async SQLite storage support:

pip install "authority-auth[async]"

For development:

pip install "authority-auth[dev]"

Quick Start

from authority import AuthConfig, AuthManager
from authority.storage import SQLiteStorage

# Configure
config = AuthConfig(
    jwt_secret_key="your-secret-key-min-32-chars",
    fernet_key="your-fernet-key",  # Generate with: python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
)

# Initialize storage (creates/opens the database)
storage = SQLiteStorage("auth.db")

# Create the auth manager
auth = AuthManager(config, storage)

# Register a user
user = auth.register(
    name="Alice",
    email="alice@example.com",
    password="SecureP@ssw0rd!",
    auto_verify=True,
)
print(f"Registered user: {user['id']}")

# Login
result = auth.login(email="alice@example.com", password="SecureP@ssw0rd!")
print(f"Access token: {result['access_token'][:20]}...")
print(f"Refresh token: {result['refresh_token'][:20]}...")

# Verify an access token
payload = auth.verify_access_token(result["access_token"])
print(f"User ID from token: {payload['user_id']}")

# Refresh tokens
new_tokens = auth.refresh_access_token(result["refresh_token"])

# Logout
auth.logout(
    user_id=user["id"],
    refresh_token=result["refresh_token"],
    access_token_jti=payload["jti"],
)

# Clean up
auth.close()

Async Usage

import asyncio
from authority import AuthConfig, AsyncAuthManager
from authority.storage import AsyncSQLiteStorage

async def main():
    config = AuthConfig(
        jwt_secret_key="your-secret-key-min-32-chars",
        fernet_key="your-fernet-key",
    )

    storage = AsyncSQLiteStorage("auth.db")
    await storage.connect()

    auth = AsyncAuthManager(config, storage)

    # Register
    user = await auth.register(
        name="Bob",
        email="bob@example.com",
        password="SecureP@ssw0rd!",
        auto_verify=True,
    )

    # Login
    result = await auth.login(email="bob@example.com", password="SecureP@ssw0rd!")
    print(f"Access token: {result['access_token'][:20]}...")

    # Verify
    payload = await auth.verify_access_token(result["access_token"])
    print(f"User ID: {payload['user_id']}")

    # Refresh
    new_tokens = await auth.refresh_access_token(result["refresh_token"])

    # Logout all devices
    count = await auth.logout_all(user_id=user["id"])
    print(f"Revoked {count} tokens")

    await auth.close()

asyncio.run(main())

FastAPI Integration

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from authority import AuthConfig, AsyncAuthManager
from authority.storage import AsyncSQLiteStorage
from authority.exceptions import (
    InvalidCredentialsError,
    TokenExpiredError,
    InvalidTokenError,
    MFAFailedError,
    UserExistsError,
    ValidationError,
)

app = FastAPI()
security = HTTPBearer()

config = AuthConfig(
    jwt_secret_key="your-secret-key-min-32-chars",
    fernet_key="your-fernet-key",
)
storage = AsyncSQLiteStorage("auth.db")
auth = AsyncAuthManager(config, storage)


@app.on_event("startup")
async def startup():
    await storage.connect()


@app.on_event("shutdown")
async def shutdown():
    await auth.close()


async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
) -> dict:
    try:
        payload = await auth.verify_access_token(credentials.credentials)
        user = await auth.get_user(payload["user_id"])
        return user
    except (InvalidTokenError, TokenExpiredError):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or expired token",
        )


@app.post("/register")
async def register(name: str, email: str, password: str):
    try:
        user = await auth.register(name, email, password, auto_verify=True)
        return {"user_id": user["id"], "email": user["email"]}
    except UserExistsError:
        raise HTTPException(status_code=409, detail="Email already registered")
    except ValidationError as e:
        raise HTTPException(status_code=422, detail=str(e))


@app.post("/login")
async def login(email: str, password: str):
    try:
        result = await auth.login(email=email, password=password)
        if result.get("mfa_required"):
            return {"mfa_required": True, "user_id": result["user_id"]}
        return {
            "access_token": result["access_token"],
            "refresh_token": result["refresh_token"],
            "token_type": "Bearer",
        }
    except InvalidCredentialsError:
        raise HTTPException(status_code=401, detail="Invalid credentials")


@app.get("/me")
async def get_me(user: dict = Depends(get_current_user)):
    return {"id": user["id"], "name": user["name"], "email": user["email"]}

Configuration

AuthConfig accepts values via constructor arguments, environment variables (prefixed with AUTHORITY_), or defaults. Constructor kwargs take highest priority.

MFA Setup

TOTP-based multi-factor authentication with encrypted secret storage and recovery codes:

# Initiate MFA setup -- returns secret and provisioning URI
setup = auth.setup_mfa(user_id=user["id"])
print(f"Secret: {setup['secret']}")
print(f"URI: {setup['provisioning_uri']}")
# Show this URI as a QR code in your frontend

# User scans QR code, enters 6-digit code to confirm
result = auth.verify_and_enable_mfa(user_id=user["id"], code="123456")
print(f"Recovery codes (store these!): {result['recovery_codes']}")

# During login, after password succeeds:
login_result = auth.login(email="alice@example.com", password="...")
if login_result.get("mfa_required"):
    # Ask user for TOTP code or recovery code
    tokens = auth.verify_mfa_login(
        user_id=login_result["user_id"],
        code="654321",  # TOTP code or recovery code
    )

# Check MFA status
status = auth.get_mfa_status(user_id=user["id"])
print(f"MFA enabled: {status['mfa_enabled']}")
print(f"Recovery codes remaining: {status['recovery_codes_remaining']}")

# Disable MFA (requires password re-authentication)
auth.disable_mfa(user_id=user["id"], password="...")

RBAC

Role-based access control with permissions:

# Create roles
admin_role = auth.create_role("admin", "Full system access")
editor_role = auth.create_role("editor", "Can edit content")

# Create permissions
read_perm = auth.create_permission("posts:read")
write_perm = auth.create_permission("posts:write")
delete_perm = auth.create_permission("posts:delete")

# Assign permissions to roles
auth.assign_permission_to_role(admin_role["id"], read_perm["id"])
auth.assign_permission_to_role(admin_role["id"], write_perm["id"])
auth.assign_permission_to_role(admin_role["id"], delete_perm["id"])
auth.assign_permission_to_role(editor_role["id"], read_perm["id"])
auth.assign_permission_to_role(editor_role["id"], write_perm["id"])

# Assign roles to users
auth.assign_role_to_user(user_id=1, role_id=admin_role["id"])

# Check permissions
if auth.has_permission(user_id=1, permission_code="posts:delete"):
    print("User can delete posts")

# Get all effective permissions
perms = auth.get_user_permissions(user_id=1)
# ["posts:read", "posts:write", "posts:delete"]

API Keys

Scoped API keys with prefix-based lookup:

# Create an API key
key_result = auth.create_api_key(
    user_id=user["id"],
    description="Production API access",
    scopes=["read", "write"],
    expires_in_days=90,
)
print(f"Key (save this, shown only once): {key_result['key']}")
print(f"Prefix: {key_result['prefix']}")

# Verify an API key (e.g., from a request header)
key_info = auth.verify_api_key(key_result["key"])
print(f"User: {key_info['user_id']}, Scopes: {key_info['scopes']}")

# List all API keys for a user
keys = auth.list_api_keys(user_id=user["id"])
for k in keys:
    print(f"{k['key_prefix']}... - {k['description']}")

# Revoke an API key
auth.revoke_api_key(user_id=user["id"], key_prefix=key_result["prefix"])

Event System

Subscribe to lifecycle events for logging, notifications, analytics, or custom workflows:

from authority import EventBus, Event

bus = EventBus()

# Sync handler
def on_login(data):
    print(f"Login: user {data['user_id']} from {data.get('ip_address')}")

# Async handler
async def on_register(data):
    await send_welcome_email(data["email"])

bus.on(Event.USER_LOGIN_SUCCESS, on_login)
bus.on(Event.USER_REGISTERED, on_register)

# Pass the event bus when creating the manager
auth = AuthManager(config, storage, event_bus=bus)

# Supported events:
#   USER_REGISTERED, USER_LOGIN_SUCCESS, USER_LOGIN_FAILED, USER_LOGOUT
#   USER_PASSWORD_CHANGED, USER_PASSWORD_RESET, USER_EMAIL_CHANGED, USER_DELETED
#   MFA_SETUP_INITIATED, MFA_ENABLED, MFA_DISABLED, MFA_FAILED
#   TOKEN_REFRESHED, TOKEN_REUSE_DETECTED
#   WEBAUTHN_CREDENTIAL_ADDED, WEBAUTHN_CREDENTIAL_REMOVED
#   RBAC_ROLE_ASSIGNED, RBAC_ROLE_REVOKED
#   API_KEY_CREATED, API_KEY_REVOKED

Storage

Authority ships with SQLite backends. For other databases, implement StorageInterface (sync) or AsyncStorageInterface (async):

from authority.storage.base import StorageInterface

class PostgresStorage(StorageInterface):
    def get_user_by_id(self, user_id: int) -> dict | None:
        # Your implementation here
        ...

    # Implement all abstract methods ...

Security

See SECURITY.md for the full security policy, vulnerability reporting process, and security considerations.

Key security properties:

• Algorithm pinned in every jwt.decode() call to prevent algorithm confusion
• Refresh tokens stored as SHA-256 hashes, never plaintext
• Token rotation with family tracking and automatic revocation on reuse
• MFA secrets encrypted at rest with Fernet (AES-128-CBC)
• Passwords hashed with bcrypt (cost factor 12+)
• HIBP integration uses k-anonymity (only SHA-1 prefix sent)
• Audit log is append-only with chain hashing for tamper evidence

Contributing

Contributions are welcome. See CONTRIBUTING.md for guidelines.

License

MIT -- authority-auth (c) 2026 rkriad585.