GraphQL in Scala: Role-Based Access Control
Securing GraphQL fields in Scala with Caliban
Today, I'm going to answer a question asked by Łukasz Biały on Twitter:
Is there a way to get field-level RBAC (Role-Based Access Control)?
It turns out there is! However, Caliban's approach to authentication and authorization is quite flexible. Instead of enforcing a specific method, Caliban provides tools that you can use to implement access control according to your needs. In this article, we'll go through a complete example that you can reuse or draw inspiration from.
The problem
Let's imagine we have a very basic GraphQL schema:
type Query {
adminData: String! # requires Admin role
userData: String! # requires User role
}
What we want to achieve with Role-Based Access Control is to ensure that clients can only access adminData
if they have the Admin
role and can only access userData
if they have the User
role. Any attempt to request a field without the required role should result in an execution error.
We will tackle this problem in three steps:
Tagging our schema: First, we will define our schema with Caliban and specify which role(s) are required for each field.
Verifying permissions: Next, we will implement the logic to check during query execution whether each field is allowed to be requested or not.
Providing the auth context: Finally, we will see how to extract role information from incoming requests and use it in our execution logic.
Tagging our schema
Let's start by creating our schema with Caliban. If you're not familiar with this, check out my Beginner's Guide to GraphQL in Scala.
case class Query(
adminData: String,
userData: String
) derives Schema.SemiAuto
val api = graphQL(RootResolver(Query("admin", "user")))
A quick call to println(api.render)
allows us to verify that this produces the expected schema. Let's also create our two roles:
enum Role {
case Admin, User
}
Now, how do we define which role is required for each field? We will use GraphQL schema directives. Directives are a great way to add arbitrary data to our schema and access it during execution.
Caliban provides a @GQLDirective
annotation that can be used on types and fields. Let's create a more specific version of this annotation to make a hasRole
directive. This directive will have a single property, role
, which will specify the required role.
val directiveName = "hasRole"
val attributeName = "role"
class HasRoleDirective(role: Role)
extends GQLDirective(
Directive(
directiveName,
Map(attributeName -> StringValue(role.toString))
)
)
I extracted directiveName
and attributeName
into variables because we will need them later when we read those directives.
Once we have this annotation, we can create even more specific annotations to tag fields directly with their exact roles.
case class admin() extends HasRoleDirective(Role.Admin)
case class user() extends HasRoleDirective(Role.User)
This can then be used when we declare the schema:
case class Query(
@admin adminData: String,
@user userData: String
) derives Schema.SemiAuto
Calling api.render
again, we see the directive appear in the generated schema:
type Query {
adminData: String! @hasRole(role: "Admin")
userData: String! @hasRole(role: "User")
}
Great, we now have a simple and minimalist way to mark fields with required roles. This is done at the schema level and does not require any changes to our resolvers. Let's see how we can use this.
Verifying permissions
To implement the verification logic, we will use the concept of FieldWrapper
. A FieldWrapper
is a piece of logic that runs every time a field is resolved. It provides the original execution computation and a FieldInfo
value containing information about the field (its type and, importantly for us, its directives). You can run any code, potentially fail, or just run the original computation.
First, let's implement a function that extracts our roles from the FieldInfo
we receive.
def getRequiredRoles(info: FieldInfo): Set[Role] =
info.directives
.filter(_.name == directiveName)
.flatMap(_.arguments.get(attributeName))
.flatMap {
case StringValue(role) => Try(Role.valueOf(role)).toOption.toList
case _ => Nil
}
.toSet
It is quite straightforward: we get all directives matching our directiveName
, read their role
attribute, and transform all of that into a Set[Role]
.
Next, to compare these required roles with the actual roles of the current user, we need an authorization context that contains these roles. Let's call it AuthContext
.
case class AuthContext(roles: Set[Role])
We can now implement our FieldWrapper
.
It will access AuthContext
in the ZIO environment and check if any required roles are missing. If no roles are missing, we run the original computation query
. If roles are missing, we fail with an error that tells us which roles are missing.
val accessControl: FieldWrapper[AuthContext] =
new FieldWrapper[AuthContext](wrapPureValues = true) {
def wrap[R1 <: AuthContext](
query: ZQuery[R1, ExecutionError, ResponseValue],
info: FieldInfo
): ZQuery[R1, ExecutionError, ResponseValue] =
ZQuery.serviceWithQuery[AuthContext] { ctx =>
val missingRoles = getRequiredRoles(info).diff(ctx.roles)
if (missingRoles.isEmpty) query
else ZQuery.fail(ExecutionError(s"Missing required roles: ${missingRoles.mkString(", ")}."))
}
}
Note that the field computation effect is ZQuery
, not ZIO
. This means that if your field wrapper needs to access a database, it can benefit from the optimizations that ZQuery
provides, such as caching, deduplication, and batching.
You may have also noticed that we used wrapPureValues = true
. This setting decides whether to run this logic on fields that don't need any effects to execute. In this case, we want to wrap all fields because each field may have different permissions. If your schema is clearly divided at the top level between admin fields and user fields, and everything under these inherits their parent permissions, you may not need to wrap pure values (which will slightly improve performance).
We can then apply our wrapper to our API using @@
.
val api = graphQL(RootResolver(Query("admin", "user")))
@@ accessControl
Adding this wrapper changes the type of api
. Previously, it was GraphQL[Any]
, but now it is GraphQL[AuthContext]
. This means we need to provide an AuthContext
to run a GraphQL request. Let's see how.
Providing the auth context
To expose our server, we are going to use the caliban-quick
module which is the simplest and fastest solution and is based on zio-http
. Similar implementations can be done if you use other server libraries such as http4s
or pekko-http
.
The zio-http
library has the concept of Middleware
that can be used to run some code on each request and potentially fail or provide some value to downstream code. That is exactly what customAuthProviding
is doing: given an incoming request, you can either fail or return some context that will be provided as part of the ZIO environment of your request handler.
We're going to use it with our AuthContext
: we are checking if there is a Roles
header in the incoming request, and if there is, we use it to build AuthContext
. Note that we wouldn't do it this way in a real-life scenario; instead, we would get a token from the headers, validate that token, and get the user roles from some kind of backend. But the implementation of this middleware would more or less be the same.
val middleware =
Middleware.customAuthProviding[AuthContext] { req =>
req.headers
.get("Roles")
.map(_.split(",").flatMap(s => Try(Role.valueOf(s)).toOption).toSet)
.map(roles => AuthContext(roles))
}
The rest of the code is straightforward: we convert our GraphQL
object into a route with the path api/graphql
, apply our middleware (using @@
), and run everything on port 8080.
Unsafe.unsafely {
Runtime.default.unsafe.run {
api
.routes("api/graphql")
.map(_ @@ middleware)
.flatMap(_.serve.provideLayer(Server.defaultWithPort(8080)))
}
}
We can now test our API. Let's say our Roles
header is Roles: User
. If we call query { adminData userData }
, we will get the following error:
{
"data": null,
"errors": [
{
"message": "Missing required roles: Admin.",
"locations": [
{
"line": 1,
"column": 9
}
],
"path": [
"adminData"
]
}
]
}
Not only do we get an error as expected, but the error also tells us which field triggered the error and which role was missing. Neat!
On the other hand, if we run query { userData }
, we get the expected result:
{
"data": {
"userData": "user"
}
}
Conclusion
That's it for our example! We have shown how to implement role-based field access control using a custom directive to mark our fields, a field wrapper to run the verification logic, and an HTTP middleware to extract our auth context. These simple steps can be customized in many ways to fit your specific needs.
The complete code for this example can be found in this Gist.
You can run it directly with Scala CLI using this command:
scala-cli https://gist.github.com/ghostdogpr/f8dc3c33a3d2d85f6f8dfa1fb23ce1ab