Skip to content

Casbin

Casbin User to Your app: cris: POST …/3/publish. Your app to Casbin policy: (cris, articles, publish)? Casbin policy to Your app: deny: a writer. Your app to User: 403 Forbidden. User to Your app: ben: POST …/3/publish. Your app to Casbin policy: (ben, articles, publish)? Casbin policy to Your app: allow: an editor. Your app to User: 200, published by ben. Who may do what is in the policy, not in the routes. User Your app Casbin policy cris: POST …/3/publish (cris, articles, publish)? deny: a writer 403 Forbidden ben: POST …/3/publish (ben, articles, publish)? allow: an editor 200, published by ben Who may do what is in the policy, not in the routes
What it's for: a Casbin policy (roles and their permissions) decides who may do what, for each route or each action, instead of checks spread over the code.

A newsroom API whose policy, not its routes, says who may do what. Anybody reads the published articles; writers write drafts, editors edit and publish them, and admins give people roles. The FastAPIAuthz dependencies check each request against Casbin's policy: on the path and method of the route, or on a named action such as (ben, articles, publish).

Users log in with a fixed token each, which stands in for your authentication (JWTs, sessions, API keys): ana-token (admin), ben-token (editor), cris-token (writer) and dan-token (no role yet).

Run it

From a clone of the repository:

make deps                 # .venv, with every extra
source .venv/bin/activate
uvicorn examples.authz.main:app

Try it

Visitors read the published articles. Cris, a writer, reads the drafts too:

$ curl http://127.0.0.1:8000/articles
[{"title":"Olive harvest starts","body":"...","id":1,"author":"cris","published_by":"ben"}]

$ curl http://127.0.0.1:8000/articles -H "Authorization: Bearer cris-token"
[{"title":"Olive harvest starts",...,"published_by":"ben"},{"title":"Draft: new mill","body":"...","id":2,"author":"cris","published_by":null}]

Writing needs an account, and the writer role:

$ curl -i -X POST http://127.0.0.1:8000/articles -H "Content-Type: application/json" \
    -d '{"title": "Mill opens", "body": "On Monday."}'
HTTP/1.1 401 Unauthorized
www-authenticate: Bearer

{"detail":"Not authenticated"}

$ curl -X POST http://127.0.0.1:8000/articles -H "Authorization: Bearer cris-token" \
    -H "Content-Type: application/json" -d '{"title": "Mill opens", "body": "On Monday."}'
{"title":"Mill opens","body":"On Monday.","id":3,"author":"cris","published_by":null}

Publishing is a named action, (user, articles, publish), which only editors may do:

$ curl -i -X POST http://127.0.0.1:8000/articles/3/publish -H "Authorization: Bearer cris-token"
HTTP/1.1 403 Forbidden

{"detail":"Forbidden"}

$ curl -X POST http://127.0.0.1:8000/articles/3/publish -H "Authorization: Bearer ben-token"
{"title":"Mill opens","body":"On Monday.","id":3,"author":"cris","published_by":"ben"}

Dan has an account but no role, and so reads what visitors read, and nothing more. Ana, an admin, makes Dan a writer, which takes effect at once:

$ curl -i -X POST http://127.0.0.1:8000/articles -H "Authorization: Bearer dan-token" \
    -H "Content-Type: application/json" -d '{"title": "Hello", "body": "First post."}'
HTTP/1.1 403 Forbidden

{"detail":"Forbidden"}

$ curl -i -X PUT http://127.0.0.1:8000/users/dan/roles/writer -H "Authorization: Bearer ana-token"
HTTP/1.1 204 No Content

$ curl -s -o /dev/null -w "%{http_code}\n" -X POST http://127.0.0.1:8000/articles \
    -H "Authorization: Bearer dan-token" -H "Content-Type: application/json" \
    -d '{"title": "Hello", "body": "First post."}'
201

Admins only grant the roles the policy knows:

$ curl -X PUT http://127.0.0.1:8000/users/dan/roles/root -H "Authorization: Bearer ana-token"
{"detail":[{"type":"enum","loc":["path","role"],"msg":"Input should be 'writer', 'editor' or 'admin'",...}]}

A wrong token is refused, rather than taken for a visitor without one:

$ curl -i http://127.0.0.1:8000/articles -H "Authorization: Bearer nonsense"
HTTP/1.1 401 Unauthorized
www-authenticate: Bearer

{"detail":"Invalid token"}

What to look at

  • The policy is data: policy.csv has a p line per permission of a role and a g line per role of a user (or of another role: editors are writers too). Changing who may do what doesn't change the code; in production the policy comes from a database, through one of Casbin's async adapters.
  • Two ways to ask: Depends(authz) asks about the route, as (user, path, method): Casbin's keyMatch2 takes /articles/:id as a pattern. members.require("articles", "publish") asks about a named action, the same on whatever route it guards.
  • Visitors and members: authz has the policy check visitors without a token as *anonymous, a name no user can have, so that the policy opens the reading routes to them. members refuses them with a 401 before asking the policy, and so always returns a user name: the routes for members don't deal with None.
  • What visitors may do, everybody may: the matcher grants the rules of *anonymous to every subject (p.sub == "*anonymous" || g(r.sub, p.sub)), so Dan, without a role, isn't worse off than a visitor.
  • Checks that depend on the data go in the route, with await authz.enforce(): everybody may list the articles, but only those who may read the drafts see them. A draft is a 404 for the others, not a 403, which would tell that it exists.
  • Roles granted at runtime take effect at once in the process that grants them: an AsyncEnforcer changes its policy in the event loop, where the checks run. The file adapter of the example doesn't keep them; a database adapter does, and a Casbin watcher tells the other processes to load the policy again.

The code

examples/authz/main.py
"""A newsroom API whose policy, not its routes, says who may do what: Casbin's RBAC.

    uvicorn examples.authz.main:app

Anybody reads the published articles; writers write drafts, editors edit and publish them, and
admins give people roles. Who may do what is in policy.csv (`p` lines: what a role may do; `g`
lines: who has which role, and which roles include others), checked by FastAPIAuthz: on the path
and method of a route (`Depends(authz)`), or on a named action (`Depends(authz.require(...))`).

The policy is read from files here; in production, from a database through one of Casbin's
async adapters (casbin-async-sqlalchemy-adapter...), which also keeps the roles granted at runtime.
"""

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from enum import StrEnum
from pathlib import Path
from typing import Annotated

import casbin
import uvicorn
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from pydantic import BaseModel, Field

from fastapi_alcazaba import FastAPIAuthz

HERE = Path(__file__).parent


class Role(StrEnum):
    # An enum: admins can only grant the roles the policy knows.
    WRITER = "writer"
    EDITOR = "editor"
    ADMIN = "admin"


class Draft(BaseModel):
    title: str = Field(max_length=200)
    body: str = Field(max_length=10_000)


class Article(Draft):
    id: int
    author: str
    published_by: str | None = None


class Database:
    """Stands in for the database."""

    def __init__(self) -> None:
        self.articles = {
            1: Article(
                id=1, title="Olive harvest starts", body="...", author="cris", published_by="ben"
            ),
            2: Article(id=2, title="Draft: new mill", body="...", author="cris"),
        }


# Async: its changes to the policy (granting a role) run in the event loop, like the checks.
enforcer = casbin.AsyncEnforcer(str(HERE / "model.conf"), str(HERE / "policy.csv"))
db = Database()
bearer = HTTPBearer(auto_error=False)
# Stands in for the app's authentication (JWTs, sessions, API keys): a token per user.
TOKENS = {"ana-token": "ana", "ben-token": "ben", "cris-token": "cris", "dan-token": "dan"}


async def current_user(
    credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer)],
) -> str | None:
    """Who makes the request; None for a visitor without a token."""
    if credentials is None:
        return None
    user = TOKENS.get(credentials.credentials)
    if user is None:  # a wrong token isn't a visitor without one
        raise HTTPException(401, "Invalid token", headers={"WWW-Authenticate": "Bearer"})
    return user


# Routes open to visitors when the policy says so: the policy knows them as "*anonymous", a name
# no user can have (the * isn't allowed in user names), so nobody signs up to get its rules.
authz = FastAPIAuthz(enforcer, subject=current_user, anonymous="*anonymous")
# Routes for people with an account: a 401 without one, before asking the policy.
members = FastAPIAuthz(enforcer, subject=current_user)
Visitor = Annotated[str | None, Depends(authz)]
Member = Annotated[str, Depends(members)]


@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncIterator[None]:
    await enforcer.load_policy()  # an AsyncEnforcer reads it when told
    yield


app = FastAPI(lifespan=lifespan)


async def reads_drafts(user: str | None) -> bool:
    # A check inside the route: the route is open to everybody, the drafts aren't.
    return user is not None and await authz.enforce(user, "drafts", "read")


@app.get("/articles")
async def list_articles(user: Visitor) -> list[Article]:
    """The published articles; the drafts too for those who may read them."""
    drafts = await reads_drafts(user)
    return [article for article in db.articles.values() if article.published_by or drafts]


@app.get("/articles/{article_id}")
async def read_article(article_id: int, user: Visitor) -> Article:
    article = db.articles.get(article_id)
    if article is None or not (article.published_by or await reads_drafts(user)):
        raise HTTPException(404, "No such article")  # for readers, a draft doesn't exist
    return article


@app.post("/articles", status_code=status.HTTP_201_CREATED)
async def write_article(draft: Draft, user: Member) -> Article:
    article = Article(id=max(db.articles) + 1, author=user, **draft.model_dump())
    db.articles[article.id] = article
    return article


@app.put("/articles/{article_id}")
async def edit_article(article_id: int, draft: Draft, user: Member) -> Article:
    article = db.articles.get(article_id)
    if article is None:
        raise HTTPException(404, "No such article")
    if article.author != user and not await authz.enforce(user, "articles", "edit"):
        raise HTTPException(403, "Only its author or an editor edits an article")
    article.title, article.body = draft.title, draft.body
    return article


@app.post("/articles/{article_id}/publish")
async def publish_article(
    article_id: int,
    editor: Annotated[str, Depends(members.require("articles", "publish"))],
) -> Article:
    """A named action, not a path: (editor, articles, publish)."""
    article = db.articles.get(article_id)
    if article is None:
        raise HTTPException(404, "No such article")
    article.published_by = editor
    return article


@app.put(
    "/users/{user}/roles/{role}",
    status_code=status.HTTP_204_NO_CONTENT,
    dependencies=[Depends(members.require("roles", "grant"))],
)
async def grant_role(user: str, role: Role) -> None:
    """Takes effect at once in this process. The other processes see it when they load the policy
    again: a Casbin watcher (casbin-redis-watcher...) tells them."""
    await enforcer.add_role_for_user(user, role)


if __name__ == "__main__":
    uvicorn.run("main:app", reload=True)

examples/authz/model.conf
# RBAC on paths and on named actions: a request is (who, what, how).
#   (ben, /articles/7, PUT): may Ben edit article 7?
#   (ben, articles, publish): may Ben publish articles?
# keyMatch2 takes /articles/:id as a pattern; "articles" matches itself. What visitors
# (*anonymous: no user name starts with *) may do, everybody may: a user without a role reads
# what visitors read.

[request_definition]
r = sub, obj, act

[policy_definition]
p = sub, obj, act

[role_definition]
g = _, _

[policy_effect]
e = some(where (p.eft == allow))

[matchers]
m = (p.sub == "*anonymous" || g(r.sub, p.sub)) && keyMatch2(r.obj, p.obj) && r.act == p.act
examples/authz/policy.csv
p, *anonymous, /articles, GET
p, *anonymous, /articles/:id, GET
p, writer, /articles, POST
p, writer, /articles/:id, PUT
p, writer, drafts, read
p, editor, articles, edit
p, editor, articles, publish
p, admin, roles, grant
g, editor, writer
g, admin, editor
g, ana, admin
g, ben, editor
g, cris, writer