anph-auth 1.0.1

Internal auth package — PASETO v4.public + RBAC + Redis revocation

Installation & Usage Guide — anph-auth

---

Table of Contents

1. System Requirements
2. Install the Package
3. Generate PASETO Keys
4. Environment Configuration
5. Initialize the Database
6. Auth Service — Login & Sign Token
7. Gateway / Microservice — Verify Token
8. FastAPI Integration
9. Token Revocation
10. RBAC Management via AuthRepository
11. Manual Permission Checking
12. Running Tests

---

1. System Requirements

Component	Minimum Version	Role
Python	3.11+	Runtime
PostgreSQL	14+	RBAC storage (Auth Service only)
Redis	6.0+	JTI Blacklist — immediate token revocation

---

2. Install the Package

pip install anph-auth

Verify the installation:

python -c "import auth_core; print(auth_core.__version__)"
# → 1.0.1

For development: Clone the repo and install in editable mode:

git clone <repo-url>
cd auth
pip install -e ".[dev]"

---

3. Generate PASETO Keys

Required before running in any environment.

The package ships with the auth-keygen CLI:

# Generate keys and print them ready to paste into .env
auth-keygen --env

Sample output:

PASETO_PRIVATE_KEY=LS0tLS1CRUdJTiBQUklWQVRFIEtFWS0tLS0t...
PASETO_PUBLIC_KEY=LS0tLS1CRUdJTiBQVUJMSUMgS0VZLS0tLS0t...

Other options:

auth-keygen               # Print both keys as base64
auth-keygen private       # Print only the private key (base64)
auth-keygen public        # Print only the public key (base64)
auth-keygen --pem         # Print raw PEM (for K8s Secrets / Vault)

Key distribution by service:

Service	Required Keys
Auth Service (signing)	PASETO_PRIVATE_KEY + PASETO_PUBLIC_KEY
Gateway / Microservice (verify)	PASETO_PUBLIC_KEY only

---

4. Environment Configuration

Create a .env.local file at the project root (or use .enviroment — the package loads both automatically):

# ── PostgreSQL (Auth Service only) ──────────────────────────────────────────
PG_HOST=192.168.1.10
PG_PORT=5432
PG_USER=root
PG_PASSWORD=your_password
PG_DATABASE=auth

# ── Redis (any service that needs revocation checks) ────────────────────────
REDIS_HOST=192.168.1.10
REDIS_PORT=6379
REDIS_PASSWORD=your_redis_password
REDIS_DB=0

# ── PASETO Keys (paste output from step 3) ──────────────────────────────────
PASETO_PRIVATE_KEY=LS0tLS1CRUdJTi...
PASETO_PUBLIC_KEY=LS0tLS1CRUdJTi...

Config load priority (high → low):

System environment variables  >  .env.local  >  .enviroment  >  .env  >  Field defaults

Verify the config loaded correctly:

from auth_core import settings

print(settings.pg_host)       # 192.168.1.10
print(settings.redis_host)    # 192.168.1.10
print(settings.postgres_url)  # postgresql+asyncpg://root:...@192.168.1.10:5432/auth

---

5. Initialize the Database

Run the following script once to create all 7 tables in PostgreSQL:

# scripts/init_db.py
import asyncio
from auth_core import settings, Base

async def main():
    engine = settings.build_async_engine()
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    await engine.dispose()
    print("Database initialized.")

asyncio.run(main())

python scripts/init_db.py

Schema diagram:

users ──< user_groups >── groups ──< group_roles >── roles ──< role_permissions >── permissions

Table	Description
users	Users — id, username, email, hashed_password, is_active
groups	Groups / Departments — id, name, description
roles	Roles — id, name, description
permissions	Atomic permissions — id, slug (e.g. user.edit, file.upload)
user_groups	User ↔ Group relationship
group_roles	Group ↔ Role relationship
role_permissions	Role ↔ Permission relationship

---

6. Auth Service — Login & Sign Token

The core flow: query DB → flatten permissions → sign PASETO token.

import asyncio
from auth_core import settings, AuthRepository, TokenPayload, db_session, sign_token

async def login(username: str) -> str:
    session_factory = settings.build_db_session_factory()
    keypair = settings.load_keypair()

    async with db_session(session_factory) as session:
        repo = AuthRepository(session)

        # 1. Fetch the user
        user = await repo.get_user_by_username(username)
        if user is None:
            raise ValueError("User not found")

        # 2. JOIN across 7 tables → collect all permissions (flattening)
        perms = await repo.get_flat_permissions(user.id)
        # → ["user.read", "user.edit", "report.export", ...]

        # 3. Build payload and sign PASETO token
        payload = TokenPayload(sub=str(user.id), perms=perms)
        token = sign_token(payload.to_dict(), keypair.private_key)

    return token

token = asyncio.run(login("an"))
print(token)
# → v4.public.eyJleHAiOiIyMDI2...

---

7. Gateway / Microservice — Verify Token

Microservices do not need a DB connection — only the Public Key is required.

from auth_core import settings, verify_token, TokenPayload, PermissionChecker

public_key = settings.load_public_key_obj()

def handle_request(token: str, required_permission: str):
    # 1. Verify signature and expiry
    raw = verify_token(token, public_key)
    payload = TokenPayload.from_dict(raw)

    # 2. Check permission directly from the payload (O(1) — no DB needed)
    checker = PermissionChecker(list(payload.perms))
    if not checker.has(required_permission):
        raise PermissionError(f"Missing permission: {required_permission}")

    return payload

---

8. FastAPI Integration

Setup at startup

from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
from auth_core import settings, AuthDependency, TokenPayload

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.auth = AuthDependency(
        public_key_pem=settings.load_public_key_bytes(),
        revocation_store=settings.build_revocation_store(),
        leeway_seconds=5,
    )
    yield
    await app.state.auth._store.close()

app = FastAPI(lifespan=lifespan)

Protecting endpoints

@app.get("/me")
async def get_me(payload: TokenPayload = Depends(app.state.auth)):
    return {"sub": payload.sub, "perms": payload.perms}

@app.get("/users")
async def list_users(payload: TokenPayload = Depends(app.state.auth.require("user.read"))):
    return {"data": [...]}

@app.post("/users")
async def create_user(payload: TokenPayload = Depends(app.state.auth.require("user.create"))):
    return {"status": "created"}

Multiple permissions (AND logic)

# Requires BOTH permissions
@app.delete("/users/{user_id}")
async def delete_user(payload: TokenPayload = Depends(app.state.auth.require("user.delete", "admin.access"))):
    ...

Wildcard permissions

A token containing "config.*" matches config.read, config.write, config.delete, etc.

# Token contains: ["user.read", "config.*"]
@app.put("/config/smtp")
async def update_smtp(payload: TokenPayload = Depends(app.state.auth.require("config.write"))):
    # ✅ Allowed — config.* covers config.write
    ...

HTTP responses on auth failure:

// 401 Unauthorized — token expired
{"detail": "Token has expired"}

// 401 Unauthorized — token invalid
{"detail": "Invalid or malformed token"}

// 403 Forbidden — missing permission
{"detail": "Missing required permissions: user.delete"}

---

9. Token Revocation

Revoke a token immediately without waiting for it to expire:

import asyncio
from auth_core import settings, TokenPayload

async def revoke(token_payload: TokenPayload):
    store = settings.build_revocation_store()
    try:
        # Blacklist the JTI with a TTL equal to the token's remaining lifetime
        await store.revoke(token_payload.jti, token_payload.ttl_seconds)
        print(f"Token {token_payload.jti} has been revoked.")
    finally:
        await store.close()

Manual check:

async def check(jti: str):
    store = settings.build_revocation_store()
    is_revoked = await store.is_revoked(jti)
    await store.close()
    return is_revoked

The FastAPI middleware (AuthDependency) automatically checks the blacklist on every request when a revocation_store is provided at initialization.

---

10. RBAC Management via AuthRepository

Seeding initial data

import asyncio
from auth_core import settings, AuthRepository, db_session

async def seed():
    factory = settings.build_db_session_factory()
    async with db_session(factory) as session:
        repo = AuthRepository(session)

        # Create permissions
        p_read  = await repo.create_permission("user.read",   "View user list")
        p_edit  = await repo.create_permission("user.edit",   "Edit user info")
        p_del   = await repo.create_permission("user.delete", "Delete users")
        p_cfg   = await repo.create_permission("config.*",    "Full config access")

        # Create roles
        admin  = await repo.create_role("admin",  "Administrator")
        editor = await repo.create_role("editor", "Editor")

        # Assign permissions to roles
        await repo.assign_permission_to_role(admin.id, p_read.id)
        await repo.assign_permission_to_role(admin.id, p_edit.id)
        await repo.assign_permission_to_role(admin.id, p_del.id)
        await repo.assign_permission_to_role(admin.id, p_cfg.id)
        await repo.assign_permission_to_role(editor.id, p_read.id)
        await repo.assign_permission_to_role(editor.id, p_edit.id)

        # Create a group
        dev_team = await repo.create_group("dev-team", "Development team")

        # Assign role to group
        await repo.assign_role_to_group(dev_team.id, admin.id)

        # Create a user and add them to the group
        user = await repo.create_user(
            username="an",
            hashed_password="$argon2id$...",  # hashed by your own library
            email="[e-mail]",
        )
        await repo.assign_user_to_group(user.id, dev_team.id)

asyncio.run(seed())

Fetching a user's permissions (used during login)

async def get_perms(username: str) -> list[str]:
    factory = settings.build_db_session_factory()
    async with db_session(factory) as session:
        repo = AuthRepository(session)
        user = await repo.get_user_by_username(username)
        return await repo.get_flat_permissions(user.id)

perms = asyncio.run(get_perms("an"))
# → ["config.*", "user.delete", "user.edit", "user.read"]

---

11. Manual Permission Checking

from auth_core import PermissionChecker

checker = PermissionChecker(["user.read", "config.*"])

checker.has("user.read")       # True  — exact match
checker.has("config.write")    # True  — wildcard match (config.*)
checker.has("user.delete")     # False

checker.has_all(["user.read", "config.smtp"])   # True
checker.has_any(["user.delete", "config.smtp"]) # True

checker.missing(["user.read", "user.delete"])
# → ["user.delete"]

---

12. Running Tests

linux

PYTHONPATH=src pytest tests/ -v

cmd window

set PYTHONPATH=src & python -m pytest tests/ -v

---

Documentation last updated by An — Monday, April 27th, 2026