Threading Together Fine-Grained Auth With Stytch and Cerbos | Cerbos

Threading together fine-grained auth with Stytch and Cerbos

Sam Lock December 07, 2022 8 min read

Stytch is an all-in-one platform for authentication and authorization. They provide a fully integrated suite offering a variety of auth solutions; including magic-links, passwords, OTP, session management, other hosted logins and more. Cerbos is an open source access control system that can handle all of your complicated business logic through simple configuration. It allows you to plug it into your existing stack as a decoupled service.

In this demo, we'll be setting up a Python FastAPI service, using Stytch's Email Magic Links for authentication, and Session Management for "just-in-time" identity verification. We'll be rolling it all up with Cerbos' decoupled authorization for fine-grained access control.

Prerequisites

You'll need the following:

You can find the code for this demo here.

Setting up your Stytch project

This is easy-peasy:

  1. Go to the Stytch Dashboard, sign up (if you haven't) and log in.
  2. Retrieve your project_id and secret from the API keys page. Set them as the following environment variables, respectively:
   STYTCH_PROJECT_ID
   STYTCH_SECRET
  1. Go to the Redirect URLs page, and set the Login and Sign-up redirect URLs to http://localhost:3000/callback.

A note on role management...

Stytch doesn't have an explicit method to store roles against an identity. What it does have, however, is support for arbitrary JSON objects within specific metadata fields.

It offers two types of metadata, trusted_metadata and untrusted_metadata. We only want Direct API integrations to be able to write to our metadata, therefore we use trusted_metadata. The metadata objects can store up to 20 top level keys. We'll use one of them to store a list of roles.

The code

Some initial points:

Let's take a look at some of the more interesting components in more detail below.

Streamlining (de)serialization

The User and TrustedMetadata classes here are Python dataclass instances:

@dataclass_json
@dataclass
class TrustedMetadata:
    roles: set[str] = field(default_factory=set)

@dataclass_json
@dataclass
class User:
    user_id: str
    trusted_metadata: TrustedMetadata = field(default_factory=TrustedMetadata)

@property
    def roles(self) -> set[str]:
        return self.trusted_metadata.roles

def add_role(self, role: str):
        self.trusted_metadata.roles.add(role)

Notice how TrustedMetadata is nested as a child of User. Structuring our dataclasses like this allows us to map to the expected structure of requests and responses to and from Stytch. This in turn allows us to effortlessly serialize/deserialize (and validate) payloads into convenient data structures, via User.from_dict(...) and User.to_dict(...). We'll see how this works below.

Login/sign-up

Navigating to the root page at localhost:3000/ will hit the default / route and trigger the index function. If you're not already signed in, it'll load the sign-in page (otherwise, on successful authentication, you'll find yourself redirected to the /user page - more on that below).

You'll notice parameters in the login_or_create_user function signature that default to an instance of Form():

@app.post("/login_or_create_user")
async def login_or_create_user(request: Request, email: str = Form(), role: str = Form()):
    ...

FastAPI maps these parameter names to those in HTML:

<form action="{{ url_for('login_or_create_user') }}" method="POST">
  <input type="text" name="email" placeholder="example@email.com" />
  <label for="role">Role</label>
  <select name="role" id="role">
    <option value="user">User</option>
    <option value="admin">Admin</option>
  </select>
  <button type="submit">Sign In</button>
</form>

On submit, FastAPI retrieves the values for each and passes them in to the route handler functions as the respective parameters.

Next, we need to call the login_or_create Stytch endpoint with the provided email. Behind the scenes, Stytch will then attempt to lookup the user, or create a new one if they don't exist. It'll then send them a magic link by email. This is as simple as:

resp = stytch_client.magic_links.email.login_or_create(
    email=email,
    login_magic_link_url=MAGIC_LINK_URL,
    signup_magic_link_url=MAGIC_LINK_URL,
)

if resp.status_code != 200:
    raise Exception

After a successful call, we can infer that a user now exists within Stytch. It's at this point that we send a separate request to populate that user with an appropriate role, via the push_role_to_stytch function:

def push_role_to_stytch(user_id: str, role: str):
    u = User(user_id=user_id)
    u.add_role(role)
    try:
        resp = stytch_client.users.update(
            **u.to_dict(),
        )
        ...

You can see how a User instance is created, a role added as a top-level key in the trusted_metadata dict, and the User serialized to json.

Authentication callback

We now have a User stored in Stytch (with roles attributed), and an email with the magic link sitting in the User's inbox. What happens when the link is clicked?

Stytch knows where to redirect to because we specified the URLs in the configuration step above. Clicking the link redirects to our callback function. There's a few steps here, so I'll break them down.

The session_token is passed as a query parameter. It's retrieved from the request object passed in to the route function:

@app.route("/callback")
async def callback(request: Request):
    token = request.query_params["token"]
    ...

We use this token to authenticate the user against Stytch's session auth API, along with a couple of other parameters:

try:
    data = {
        "token": request.query_params["token"],
        "session_duration_minutes": 60,
    }

if (t := request.session.get("session_token")) is not None:
        data["session_token"] = t

resp = stytch_client.magic_links.authenticate(**data)

if resp.status_code != 200:
        raise Exception

except Exception:
    ...

You'll notice that we retrieve a locally stored session token, if available, and add that to the authentication payload. This isn't necessary, but it allows us to extend the duration of the existing token.

Protected endpoint -> /user

This route has a double-whammy of protection. Firstly, we check that the user has permission to access the route at all, using Stytch's session authentication.

Cerbos policies

Cerbos policies are simple and extensible configuration files, written in YAML or JSON. Cerbos PDPs can load them from a number of sources, and automatically update themselves when the policies change. Cerbos is stateless, so once instantiated, any authorization requests made to it are evaluated using only the contextual information contained in the request.

Let's take a quick look at the policy defined in this demo:

apiVersion: api.cerbos.dev/v1
resourcePolicy:
  version: default
  resource: contact
  rules:
    - actions: ["read", "create"]
      effect: EFFECT_ALLOW
      roles:
        - admin
        - user

- actions: ["update", "delete"]
      effect: EFFECT_ALLOW
      roles:
        - admin

- actions: ["update", "delete"]
      effect: EFFECT_ALLOW
      roles:
        - user
      condition:
        match:
          expr: request.resource.attr.owner_id == request.principal.id

The resource: contact specifies that this particular policy applies to resources of type: "contact". Below that, you can see three separate rule groupings; for each, you might notice that they apply to roles of either admin or user. The third grouping specifies that any user is given additional permissions on a resource providing that they are the owner of the resource.

Authorization with Cerbos

Now we know the user is who they say they are and what roles they have, we can check which user permissions they have against a specific resource using the Cerbos PDP.

Firstly, we construct a Principal object using the User data:

principal = Principal(
    user.user_id,
    roles=user.roles,
)

The principal and resource_list objects are then used to construct the request to Cerbos:

with CerbosClient(host=CERBOS_HOST) as c:
    try:
        resp = c.check_resources(principal=principal, resources=resource_list)
        resp.raise_if_failed()
    except Exception:
        ...

Cerbos evaluates the principal against each resource in the list, and returns the access decisions in the response. And with that, we're done!