Using AWS Cognito with Cerbos

Cerbos integrates with many authentication providers. Our AWS Cognito integration works with the same principle that all of our other authentication integrations (Okta, Auth0, WorkOS etc).

It relies on getting the identity object and combining it with the resource that user is trying to access and ask the question whether that user is allowed to do that action on the said principal. For example, “can a user who is a manager in the northeast region approve an expense report that belongs to an employee from the south region in the amount of $5000?” or perhaps $10,000, and $50,000 for different groups of users: the point being without having to create a whole new user group for each threshold.

Following is an example of what this may look like in your application code. It simply grabs the profile from Cognito and passes it over to Cerbos in the principal object. You can see a full example on GitHub which showcases how to use it in a Python FastAPI project..

@app.get("/user", response_class=HTMLResponse)
async def user(request: Request, credentials: dict = Depends(get_user_from_session)):
    claims = credentials.claims
    user_id: str = claims["sub"]
    roles: list[str] = claims.get("cognito:groups", [])

    principal = Principal(
        user_id,
        roles=roles,
        policy_version="20210210",
        attr={
            "foo": "bar",
        },
    )

    # resources would usually be retrieved from your data store
    actions = ["read", "update", "delete"]
    resource_list = ResourceList(
        resources=[
            # This resource is owned by the user making the request
            ResourceAction(
                Resource(
                    "abc123",
                    "contact",
                    attr={
                        "owner": user_id,
                    },
                ),
                actions=actions,
            ),
            # This resource is owned by someone else
            ResourceAction(
                Resource(
                    "def456",
                    "contact",
                    attr={
                        "owner": "other_user_id",
                    },
                ),
                actions=actions,
            ),
        ]
    )

    with CerbosClient(host="http://localhost:3592") as c:
        # usually check for a specific action
        action = "read"
        if not c.is_allowed(action, principal, r):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Unauthorized")
        return {
            "id": user_id,
            "foo": "bar",
        }

You can find out more about Cognito on the AWS Developer site, our integration, as well as our other integrations such as Auth0, Okta and WorkOS and more on the Ecosystem page.