Part 5: Authentication and Authorization

Knowing Who Is Calling and What They Can Do

Most GraphQL tutorials add auth as an afterthought. In production microservices, it is a first-class concern that shapes the schema design, context factory, and resolver structure.

This part covers:

• JWT verification in the Apollo Server context function

• Field-level authorization using a resolver guard pattern

• A schema directive (@auth, @hasRole) approach

• Protecting subscriptions at connection time

• Auth in the Python Strawberry service

JWT in the Apollo Context

Every request to the Apollo Server passes through the context function before any resolver runs. This is where I verify the JWT and attach the user to the context.

// src/auth.ts
import * as jose from 'jose';

export interface AuthUser {
  id: string;
  email: string;
  role: 'ADMIN' | 'EDITOR' | 'VIEWER';
}

const SECRET = new TextEncoder().encode(
  process.env.JWT_SECRET ?? 'change-this-secret-in-production',
);

export async function verifyToken(token: string): Promise<AuthUser> {
  const { payload } = await jose.jwtVerify(token, SECRET, {
    algorithms: ['HS256'],
  });

  return {
    id: payload.sub as string,
    email: payload.email as string,
    role: payload.role as AuthUser['role'],
  };
}

export function extractBearerToken(authHeader: string | undefined): string | null {
  if (!authHeader?.startsWith('Bearer ')) return null;
  return authHeader.slice(7);
}

// src/context.ts
import { PrismaClient } from '@prisma/client';
import { createDataLoaders } from './dataloaders';
import { AuthUser, verifyToken, extractBearerToken } from './auth';

export interface AppContext {
  prisma: PrismaClient;
  loaders: ReturnType<typeof createDataLoaders>;
  user: AuthUser | null;
}

const prisma = new PrismaClient();

export async function createContext({
  req,
}: {
  req: { headers: Record<string, string | undefined> };
}): Promise<AppContext> {
  let user: AuthUser | null = null;

  const token = extractBearerToken(req.headers.authorization);
  if (token) {
    try {
      user = await verifyToken(token);
    } catch {
      // Invalid/expired token — treat as unauthenticated, not as an error.
      // If a resolver requires auth, it will throw there.
      user = null;
    }
  }

  return {
    prisma,
    loaders: createDataLoaders(prisma),
    user,
  };
}

Key design decision: an invalid token sets user to null rather than throwing. This means public endpoints (queries that do not require auth) continue to work. Resolvers that require auth throw explicitly — which gives better error messages than a blanket 401 at the transport layer.

Field-Level Authorization: Resolver Guards

The simplest pattern is a guard function called at the top of each protected resolver:

// src/auth.ts (additions)
import { ForbiddenError } from './errors';

export function requireAuth(ctx: { user: AuthUser | null }): AuthUser {
  if (!ctx.user) {
    throw new ForbiddenError('Authentication required');
  }
  return ctx.user;
}

export function requireRole(
  ctx: { user: AuthUser | null },
  ...roles: AuthUser['role'][]
): AuthUser {
  const user = requireAuth(ctx);
  if (!roles.includes(user.role)) {
    throw new ForbiddenError(
      `This operation requires one of: ${roles.join(', ')}. Your role: ${user.role}`,
    );
  }
  return user;
}

Apply at the resolver level:

// src/resolvers/product.ts
import { requireAuth, requireRole } from '../auth';

export const productResolvers = {
  // ...
  Mutation: {
    createProduct: async (_parent, { input }, ctx) => {
      requireRole(ctx, 'ADMIN', 'EDITOR');  // throws ForbiddenError if not met

      const data = validateCreateProduct(input);
      // ...
    },

    deleteProduct: async (_parent, { id }, ctx) => {
      requireRole(ctx, 'ADMIN');  // only admins can delete

      await ctx.prisma.product.delete({ where: { id } });
      return true;
    },

    updateStock: async (_parent, { productId, quantity }, ctx) => {
      requireAuth(ctx);  // any authenticated user can update stock

      return ctx.prisma.product.update({
        where: { id: productId },
        data: { stockCount: quantity },
      });
    },
  },
};

This pattern is explicit and easy to follow. Each protected resolver documents its own requirements at the top of the function body — there is no hidden magic.

Schema Directive Approach

A more declarative approach uses @auth and @hasRole directives in the SDL itself. This makes security requirements visible in the schema and enforces them via a mapper transformer rather than per-resolver calls.

# schema additions
directive @auth on FIELD_DEFINITION
directive @hasRole(role: Role!) on FIELD_DEFINITION | OBJECT

enum Role { ADMIN EDITOR VIEWER }

type Mutation {
  createProduct(input: CreateProductInput!): Product! @hasRole(role: EDITOR)
  updateProduct(id: UUID!, input: UpdateProductInput!): Product! @hasRole(role: EDITOR)
  deleteProduct(id: UUID!): Boolean! @hasRole(role: ADMIN)
  updateStock(productId: UUID!, quantity: Int!): Product! @auth
}

Implement the directive transformer:

// src/directives/auth.ts
import { MapperKind, mapSchema, getDirective } from '@graphql-tools/utils';
import { defaultFieldResolver, GraphQLSchema } from 'graphql';
import { ForbiddenError } from '../errors';
import { AuthUser } from '../auth';

export function authDirectiveTransformer(schema: GraphQLSchema): GraphQLSchema {
  return mapSchema(schema, {
    [MapperKind.OBJECT_FIELD]: (fieldConfig) => {
      const authDirective = getDirective(schema, fieldConfig, 'auth')?.[0];
      const hasRoleDirective = getDirective(schema, fieldConfig, 'hasRole')?.[0];

      if (!authDirective && !hasRoleDirective) return fieldConfig;

      const { resolve = defaultFieldResolver } = fieldConfig;

      return {
        ...fieldConfig,
        async resolve(source, args, context: { user: AuthUser | null }, info) {
          if (authDirective && !context.user) {
            throw new ForbiddenError('Authentication required');
          }

          if (hasRoleDirective) {
            const requiredRole = hasRoleDirective['role'] as AuthUser['role'];
            if (!context.user) {
              throw new ForbiddenError('Authentication required');
            }

            const roleHierarchy: Record<AuthUser['role'], number> = {
              VIEWER: 0,
              EDITOR: 1,
              ADMIN: 2,
            };

            if (roleHierarchy[context.user.role] < roleHierarchy[requiredRole]) {
              throw new ForbiddenError(
                `Role ${requiredRole} required. Your role: ${context.user.role}`,
              );
            }
          }

          return resolve(source, args, context, info);
        },
      };
    },
  });
}

Apply the transformer when building the schema:

// src/index.ts
import { makeExecutableSchema } from '@graphql-tools/schema';
import { authDirectiveTransformer } from './directives/auth';

const schema = authDirectiveTransformer(
  makeExecutableSchema({ typeDefs, resolvers }),
);

I prefer the resolver guard pattern for most projects because it is explicit and does not require schema transformer setup. The directive approach is worth it when you have many protected fields and want the security policy visible in the schema introspection.

Protecting Subscriptions

Subscriptions use a persistent WebSocket connection, not HTTP requests. The context is established at connection time, not per-message.

// src/index.ts (WebSocket server setup)
const serverCleanup = useServer(
  {
    schema,
    context: async (ctx) => {
      // ctx.connectionParams is sent by the client on connect
      const token = ctx.connectionParams?.authToken as string | undefined;

      let user: AuthUser | null = null;
      if (token) {
        try {
          user = await verifyToken(token);
        } catch {
          // Invalid token — close the connection
          throw new Error('Invalid authentication token');
        }
      }

      return {
        prisma,
        loaders: createDataLoaders(prisma),
        user,
      };
    },
    onConnect: async (ctx) => {
      // Optional: enforce auth at connect time (reject unauthenticated connections entirely)
      const token = ctx.connectionParams?.authToken as string | undefined;
      if (!token) {
        // Return false to reject, or throw to close with an error
        return false;
      }
      return true;
    },
  },
  wsServer,
);

The TypeScript Apollo Client sends the auth token in connection params:

const wsLink = new GraphQLWsLink(
  createClient({
    url: 'ws://localhost:4001/graphql',
    connectionParams: () => ({
      authToken: localStorage.getItem('authToken') ?? '',
    }),
  }),
);

JWT Signing Utility

For development and testing, a minimal signing function:

// src/auth.ts (additions)
export async function signToken(user: AuthUser, expiresIn = '8h'): Promise<string> {
  return new jose.SignJWT({
    email: user.email,
    role: user.role,
  })
    .setProtectedHeader({ alg: 'HS256' })
    .setSubject(user.id)
    .setIssuedAt()
    .setExpirationTime(expiresIn)
    .sign(SECRET);
}

In production, JWTs typically come from a dedicated identity service (not issued by the product service). The product service only verifies.

Python Strawberry: Auth with Permissions

In the Python Strawberry service (which becomes a subgraph in Part 6), authorization uses Strawberry's PermissionExtension:

# src/permissions.py
import strawberry
from strawberry.permission import BasePermission
from strawberry.types import Info
from typing import Any


class IsAuthenticated(BasePermission):
    message = "Authentication required"

    def has_permission(self, source: Any, info: Info, **kwargs: Any) -> bool:
        return info.context.get("user") is not None


class IsAdmin(BasePermission):
    message = "Admin role required"

    def has_permission(self, source: Any, info: Info, **kwargs: Any) -> bool:
        user = info.context.get("user")
        return user is not None and user.get("role") == "ADMIN"


class IsEditorOrAbove(BasePermission):
    message = "Editor role or above required"

    def has_permission(self, source: Any, info: Info, **kwargs: Any) -> bool:
        user = info.context.get("user")
        if not user:
            return False
        return user.get("role") in ("EDITOR", "ADMIN")

Apply permissions to resolvers in the schema definition:

# src/schema.py
import strawberry
from strawberry.permission import PermissionExtension
from .permissions import IsAuthenticated, IsEditorOrAbove, IsAdmin


@strawberry.type
class Mutation:
    @strawberry.mutation(extensions=[PermissionExtension(permissions=[IsEditorOrAbove()])])
    async def create_product(self, input: CreateProductInput, info: strawberry.types.Info) -> Product:
        ctx = info.context
        # ... create logic
        pass

    @strawberry.mutation(extensions=[PermissionExtension(permissions=[IsAdmin()])])
    async def delete_product(self, id: strawberry.ID, info: strawberry.types.Info) -> bool:
        ctx = info.context
        # ... delete logic
        pass

Build the Strawberry context from the FastAPI request:

# src/context.py
import os
from fastapi import Request
from jose import jwt, JWTError
from typing import Optional


def get_user_from_request(request: Request) -> Optional[dict]:
    auth = request.headers.get("Authorization", "")
    if not auth.startswith("Bearer "):
        return None

    token = auth[7:]
    try:
        payload = jwt.decode(
            token,
            os.environ["JWT_SECRET"],
            algorithms=["HS256"],
        )
        return {
            "id": payload.get("sub"),
            "email": payload.get("email"),
            "role": payload.get("role"),
        }
    except JWTError:
        return None


async def get_context(request: Request) -> dict:
    return {
        "user": get_user_from_request(request),
        # db session injected here in Part 6
    }

Wire the context into the Strawberry FastAPI router:

# src/main.py
from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter
from .schema import schema
from .context import get_context

app = FastAPI()

graphql_app = GraphQLRouter(schema, context_getter=get_context)
app.include_router(graphql_app, prefix="/graphql")

Authorization Summary

Pattern	TypeScript	Python
Auth context	jose.jwtVerify in context factory	python-jose in get_context
Guard approach	requireAuth(ctx) / requireRole(ctx, ...)	PermissionExtension classes
Directive approach	@auth / @hasRole schema directives	n/a (Strawberry uses decorators)
Subscription auth	connectionParams.authToken verified on connect	Not applicable in this series
Token issuance	Separate identity service	Separate identity service

What's Next

Authentication and authorization are now handled end to end — from JWT verification in the context function to field-level guards in both TypeScript and Python.

In Part 6, we wire everything together with Apollo Federation v2 — the TypeScript product service and the Python analytics service become independent subgraphs, unified by an Apollo Router supergraph.