UnkeyStrategy¶
API keys that Unkey keeps: people sign up and get a key, their scripts send it, and Unkey checks it on every request, with its permissions, its credits and its rate limit. The strategy: UnkeyStrategy.
Each key reads documents (documents.read), has 100 credits a month and 10 requests a minute.
Writing a document takes 5 credits and needs documents.write, which signing up doesn't give.
Run it¶
Create an API in Unkey's dashboard, and two root keys: one that may only verify its keys, one that may create and delete them. Then, from a clone of the repository:
make deps # .venv, with every extra
source .venv/bin/activate
export UNKEY_VERIFY_KEY=... UNKEY_ADMIN_KEY=... UNKEY_API_ID=api_...
uvicorn examples.auth_strategies.unkey_keys:app
Try it¶
Sign up: the key is shown this once, Unkey keeps only its hash.
$ curl -X POST http://localhost:8000/signup -H "Content-Type: application/json" -d '{"username": "ana"}'
{"key":"docs_...","key_id":"key_..."}
The key reads documents; each request takes a credit and one of the minute's 10:
$ curl http://localhost:8000/documents -H "Authorization: Bearer docs_..."
[{"title":"Style guide","owner":"unkey"}]
Writing needs documents.write, which Ana's key doesn't have: a 403. Give it the permission in
Unkey's dashboard, and the same key writes, for 5 credits:
$ curl -i -X POST http://localhost:8000/documents -H "Authorization: Bearer docs_..." -H "Content-Type: application/json" -d '{"title": "Draft"}'
HTTP/1.1 403 Forbidden
content-type: application/json
{"detail":"Insufficient permissions"}
The eleventh request of a minute gets a 429 with Retry-After, and a key out of credits a 429
too ({"detail":"No credits left"}) until they're refilled on the 1st. Ana deletes her key, and
Unkey refuses it from then on (within 30 seconds):
$ curl -i -X DELETE http://localhost:8000/keys/key_... -H "Authorization: Bearer docs_..."
HTTP/1.1 204 No Content
$ curl -i http://localhost:8000/documents -H "Authorization: Bearer docs_..."
HTTP/1.1 401 Unauthorized
www-authenticate: Bearer
content-type: application/json
{"detail":"Not authenticated"}
What to look at¶
- Two root keys: the strategy's may only verify keys,
UnkeyKeys' may create and delete them: a leak of the first can't make keys. - The user from the key: each key is made with
external_id=user.id, andload_usergets it back from every verification. - A strategy per permission: writing has a strategy of its own,
unkey-write, with its permission and its cost, and the same client. - Only the owner deletes a key: a key ID that isn't the user's is a 404.
- Unkey down: the requests get a 503, never let through.
The code¶
"""API keys that Unkey keeps: people sign up and get a key, their scripts send it, and Unkey
checks it on every request, with its permissions, its credits and its rate limit.
uvicorn examples.auth_strategies.unkey_keys:app
Each key reads documents (`documents.read`), has 100 credits a month, of which writing a
document takes 5 and needs `documents.write`, which signing up doesn't give, and 10 requests a
minute. Its owner deletes it with `DELETE /keys/{key_id}`, and Unkey refuses it from then on.
Create an API in Unkey's dashboard, and two root keys: one that may only verify its keys, one
that may create and delete them. Set UNKEY_VERIFY_KEY, UNKEY_ADMIN_KEY and UNKEY_API_ID, in the
environment or a .env file.
"""
import uuid
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from datetime import timedelta
from typing import Annotated
import uvicorn
from fastapi import Depends, FastAPI, HTTPException, status
from pydantic import BaseModel, Field, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict
from fastapi_alcazaba import FastAPIAuth
from fastapi_alcazaba.auth.strategies import UnkeyStrategy
from fastapi_alcazaba.auth.strategies.unkey import (
CreditRefill,
KeyRatelimit,
UnkeyClient,
UnkeyKey,
UnkeyKeys,
)
class Settings(BaseSettings):
"""The configuration, from environment variables or a .env file."""
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")
unkey_verify_key: SecretStr = SecretStr("set UNKEY_VERIFY_KEY")
unkey_admin_key: SecretStr = SecretStr("set UNKEY_ADMIN_KEY")
unkey_api_id: str = "set_UNKEY_API_ID"
settings = Settings()
class User(BaseModel):
id: str
username: str
class SignUp(BaseModel):
username: str = Field(pattern=r"^[a-z0-9_]{3,30}$")
class NewKey(BaseModel):
key: str # shown once: Unkey keeps only its hash
key_id: str
class Document(BaseModel):
title: str = Field(max_length=200)
owner: str = ""
class Database:
"""Stands in for the database: the users, and the ids of their keys."""
def __init__(self) -> None:
self.users: dict[str, User] = {}
self.keys_of: dict[str, set[str]] = {}
self.documents = [Document(title="Style guide", owner="unkey")]
db = Database()
# A root key for each use, with only the permissions it needs.
verifier = UnkeyClient(settings.unkey_verify_key)
admin = UnkeyClient(settings.unkey_admin_key)
keys = UnkeyKeys(admin, api_id=settings.unkey_api_id, prefix="docs")
def load_user(key: UnkeyKey) -> User | None:
"""The owner of a valid key: the user its external ID names."""
return db.users.get(key.external_id) if key.external_id is not None else None
auth = FastAPIAuth(
UnkeyStrategy(verifier, load_user, permissions="documents.read"),
# Writing needs another permission and costs more: a strategy of its own, the same client.
UnkeyStrategy(verifier, load_user, name="unkey-write", permissions="documents.write", cost=5),
)
Reader = Annotated[User, Depends(auth.authenticate("unkey"))]
Writer = Annotated[User, Depends(auth.authenticate("unkey-write"))]
@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncIterator[None]:
yield
await verifier.aclose()
await admin.aclose()
app = FastAPI(lifespan=lifespan)
@app.post("/signup", status_code=status.HTTP_201_CREATED)
async def sign_up(form: SignUp) -> NewKey:
"""A new user and its first key, which reads documents: shown once."""
if any(user.username == form.username for user in db.users.values()):
raise HTTPException(409, "That username is taken")
user = User(id=f"user_{uuid.uuid4().hex[:12]}", username=form.username)
created = await keys.create(
external_id=user.id,
name=f"{user.username}'s first key",
credits_left=100,
refill=CreditRefill(100, day=1),
ratelimits=[KeyRatelimit("requests", 10, timedelta(minutes=1))],
permissions=["documents.read"],
)
db.users[user.id] = user
db.keys_of[user.id] = {created.key_id}
return NewKey(key=created.key.get_secret_value(), key_id=created.key_id)
@app.get("/documents")
async def list_documents(user: Reader) -> list[Document]:
"""One credit of the key, and one request of its rate limit."""
del user # authenticated: that's all it needs
return db.documents
@app.post("/documents", status_code=status.HTTP_201_CREATED)
async def write_document(document: Document, user: Writer) -> Document:
"""For keys with documents.write: five credits."""
written = Document(title=document.title, owner=user.username)
db.documents.append(written)
return written
@app.delete("/keys/{key_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_key(key_id: str, user: Reader) -> None:
"""Deletes one of the user's keys, this one too: Unkey refuses it from then on."""
if key_id not in db.keys_of.get(user.id, set()):
raise HTTPException(404, "No such key") # someone else's keys don't exist for them
await keys.delete(key_id)
db.keys_of[user.id].discard(key_id)
if __name__ == "__main__":
uvicorn.run("unkey_keys:app", reload=True)