Authentication and Authorization

DBOS Python supports modular, declarative security. This is a cooperative effort between the request framework (such as FastAPI) and DBOS Transact; the server framework performs authentication and forwards the authenticated user and roles on to DBOS functions, which then check for authorization. Authentication information is forwarded via the DBOS context.

The following tutorial shows use of FastAPI middleware to collect and authenticate user information, and shows how to set function- and class-level authorization in DBOS.

Authentication Middleware

Authentication information may arrive in various ways, depending on the approach, protocol, and framework used. It is common to use some sort of "middleware" that sits between the request handler and the application code that processes the authentication information in each request.

A simple middleware for FastAPI may pass authentication information to DBOS in a manner similar to this:

    @app.middleware("http")
    async def authMiddleware(
        request: Request, call_next: Callable[[Request], Awaitable[Response]]
    ) -> Response:
        with DBOSContextSetAuth("user1", ["user", "engineer"]):
            response = await call_next(request)
            return response

There are several things happening in this code snippet.

• authMiddleware - The middleware function responsible for taking information from the Request and providing it to DBOS via the context
• @app.middleware("http") - Registers the middleware function with app, which is a FastAPI instance, so that the function is called on inbound requests
• with DBOSContextEnsure(): - Ensures that a DBOS context is associated with the request, creating one if none exists
• DBOS.set_authentication - Sets the authenticated user and roles into the DBOS context, for use in authorization determinations
• response = await call_next(request) - Proceeds down the handler call chain, with the authentication information in place

Authorization Decorators

DBOS Python uses decorators to declare the roles required to authorize access to functions and methods.

• required_roles is used at the function/method level to list roles required for function access. Users who were authenticated with any role on the list are allowed to access the function.
• default_required_roles can be used to set a list of roles that applies to each method in class, as a default. Any methods decorated with required_roles will use the list provided by required_roles instead of the default list provided to default_required_roles.

For example, most methods in a class may require "user" access, with a few exceptions, such as "login", which requires no authentication/authorization, or a few administrative functions that require the "admin" role:

    @DBOS.default_required_roles(["user"])
    class DBOSTestClass:
        @staticmethod
        @DBOS.workflow()
        def user_function(var: str) -> str:
            # "user" role required due to default_required_roles
            return var

        @staticmethod
        @DBOS.workflow()
        @DBOS.required_roles(["admin"])
        def admin_function(var: str) -> str:
            # "admin" role required due to method-level override
            return var

        @staticmethod
        @DBOS.required_roles([])
        @DBOS.workflow()
        def login_function(var: str) -> str:
            # No role, or any authentication, required due to method-level override
            return var

Example

In this example, we demonstrate how to use JWT tokens with DBOS declarative security. Here, the JWT tokens may be generated by a different part of the stack, separating out the concern of robust user management and authentication credentials, which can then be handled in specialized libraries or services.

    @app.middleware("http")
    async def jwtAuthMiddleware(
        request: Request, call_next: Callable[[Request], Awaitable[Response]]
    ) -> Response:
        user: Optional[str] = None
        roles: Optional[List[str]] = None
        try:
            token = await oauth2_scheme(request)
            if token is not None:
                tdata = decode_jwt(token)
                user = tdata.username
                roles = tdata.roles
        except Exception as e:
            pass

        with DBOSContextSetAuth(user, roles):
            response = await call_next(request)
            return response

As with the simpler example in Authorization Decorators above, @app.middleware("http") is used to insert the jwtAuthMiddleware function between the FastAPI app and DBOS. Use of DBOSContextEnsure, DBOS.set_authentication, and call_next is also the same.

What is different is that oauth2_scheme and decode_jwt are used to extract token contents. The resulting information is used to set the DBOS user and roles. If the user / roles are stored in different fields in the token, adjust the access to tdata accordingly.

The user and roles can then be used in decorated DBOS workflow, transaction, and step functions.

    @app.get("/open/{var1}")
    @DBOS.required_roles([])
    @DBOS.workflow()
    def test_open_endpoint(var1: str) -> str:
        # This function can be called with any user/role, or none at all
        #   This is true because:
        #     The `required_roles` list is empty
        #     The middleware above allows the request to be processed even if no token is present
        pass

    @app.get("/user/{var1}")
    @DBOS.required_roles(["user"])
    @DBOS.workflow()
    def test_user_endpoint(var1: str) -> str:
        # This function can be called only by a user with "user" role
        #   The `required_roles` list contains "user"
        #   Even though the middleware above allows the request to be processed
        #     even if no token is present, DBOS will block the call because the
        #     roles are not set.
        pass