Securing API Events using JWTs

Intro

The Zero Trust API Events architecture article described a design pattern for resuming asynchronous flows while maintaining identity data in a digitally verifiable manner. This tutorial provides an end-to-end code example using the popular Apache Kafka event streaming platform as the messaging middleware.

Components

The overall behavior is summarized in the below diagram, where a client application creates an order on behalf of a user. This in turn triggers an OrderCreated event, which is consumed asynchronously by a separate Payments API. The flow enables the Orders API to communicate identity to the Payments API using a JWT access token, which has a long lifetime, but very limited permissions:

In some scenarios the asynchronous processing might continue after a delay. Examples include end of day batch jobs, or workflow processes that are approved by an administrator. The same design pattern can be used for all asynchronous flows and any other system that uses event messages.

GitHub Repository

Clone the GitHub repository, which contains a number of resources. The main focus of this tutorial will be the orders-api and payments-api, which implement the API to API event processing using simple Node.js code. A minimal Node.js client application is used to initiate event processing with a user level access token.

The repo also deploys the Kafka system, an instance of the Curity Identity Server, and an API gateway that implements the Phantom Token Pattern, to swap opaque access tokens sent by the client for JWT access tokens used by APIs.

Prerequisites

First ensure that Docker and Node.js are installed. Then copy a license file for the Curity Identity Server into the root folder of the project. If you do not have one you can get a free community edition license by signing into the Curity Developer Portal with your Github account.

Deploy the System

The repo deploys all infrastructure resources, including a demo level installation of the Apache Kafka system, to a Docker Compose network. Execute the following commands to run the deployment:

./build.sh
./deploy.sh

The APIs run on the local development computer, so that log information can be easily viewed. If required, developers can add console.log statements to the main OrdersService and PaymentsService modules, then restart the APIs by running npm start from the respective folders:

The contactable URLs of the deployed system are summarized below. You can login to the Admin UI with credentials admin / Password1:

Run the Event Flow

The client application is a simple console app, run via the following commands. This triggers a user login via the system browser, and the console app listens for the login response on a loopback HTTP address at http://localhost:3003. This flow is described further in OAuth for Native Apps.

cd console-client
npm install
npm start

When prompted, sign in with the credentials demouser / Password1. This will return a user level access token to the console app, which is then sent to the Orders API along with some example order details. The console client then simply exits.

The initial access token issued to the user facing app has a payload similar to the following. It contains multiple scopes and a short lifetime of 15 minutes:

{
  "jti": "2038fb12-8089-4ebb-bca7-efdd179adc72",
  "delegationId": "e39a700d-3f3d-469e-a72f-aa02d55d5d54",
  "exp": 1657215870,
  "nbf": 1657215570,
  "scope": "openid profile orders trigger_payments",
  "iss": "http://localhost:8443/oauth/v2/oauth-anonymous",
  "sub": "demouser",
  "aud": "api.example.com",
  "iat": 1657215570,
  "purpose": "access_token"
}

You will then see the Orders API receive and create a transaction. The Orders API calls the Identity Server to get a new JWT access token, then includes it in the event message published to Kafka:

Creating Order Transaction ...
{
  "orderTransactionID": "b362998a-f132-ff80-33b3-f239706b8e4e",
  "userID": "demouser",
  "utcTime": "2022-07-08T08:00:08.069Z",
  "items": [
    {
      "itemID": 1,
      "quantity": 1,
      "price": 100
    },
    {
      "itemID": 2,
      "quantity": 2,
      "price": 100
    }
  ]
}
Performing Token Exchange ...
Publishing OrderCreated Event ...
Orders API published an OrderCreated event ...

Shortly after, the Payments API will receive the event message. It will first validate the JWT access token and make some additional checks, to prove the integrity of the event data. The Payments API will then create its own transaction:

Payments API is consuming an OrderCreated event ...
Consuming OrderCreated Event ...
Creating Payment Transaction ...
{
  "paymentTransactionID": "df91cf1b-2d63-a1fd-c2de-8f8241544aa7",
  "orderTransactionID": "b362998a-f132-ff80-33b3-f239706b8e4e",
  "userID": "demouser",
  "utcTime": "2022-07-08T08:00:08.442Z",
  "amount": 200
}

The end result is that the user identity has flowed securely between microservices. As we shall see, this is done with only simple code, and without the need for any additional infrastructure.

Event Publishing with Token Exchange

The Orders API forms an OrderCreated event message with the following structure. It includes a long lived JWT access token in the event message. This JWT can only be used at a single event processing endpoint in consuming APIs, and can only be used for this specific event message:

export interface OrderCreatedEvent {
    accessToken: string;
    payload: {
        orderTransactionID: string;
        utcTime: number;
        items: OrderItem[];
    }
}

The Orders API sends a Token Exchange message to the Identity Server with request details similar to the following:

POST /oauth/v2/oauth-token HTTP/1.1
Host: localhost:8443
Content-Type: application/x-www-form-urlencoded
 
grant_type=https://curity.se/grant/accesstoken&
client_id=orders-api-client&
client_secret=Password1&
scope=trigger_payments&
token=eyJraWQiOiItMTcyNTQxNzE2NyIsIng...&
order_transaction_id=1b6d215b-7ce2-4e0b-9079-f4e1266f57b1&
event_payload_hash=7e6d3d4b2608625f144f9c1a988da170504a368b647a6609ac4ec6c939496be1

This includes a JWT claim with a hash of the event data, which binds the event message to a claim in the access token. This is done using the cryptographic private key of the Identity Server, which no other party has access to.

The custom parameters sent are added to the long lived access token using a Token Procedure with the following logic. In the Admin UI you can locate this by navigating to System / Procedures / Token Procedures:

function result(context) {
    var accessTokenData = context.getDefaultAccessTokenData(context.delegation);

    var issuedAccessToken;
    if (context.client.properties['custom_jwt_token_exchange'] == 'true') {

        var order_transaction_id = context.request.getFormParameter('order_transaction_id');
        var event_payload_hash = context.request.getFormParameter('event_payload_hash');
        if (order_transaction_id && event_payload_hash) {
            accessTokenData.order_transaction_id = order_transaction_id;
            accessTokenData.event_payload_hash = event_payload_hash;
        }

        var now = new Date();
        var future = new Date();
        future.setFullYear(now.getFullYear() + 1);
        accessTokenData.exp = Math.round(future.getTime() / 1000);

        issuedAccessToken = context.getDefaultAccessTokenJwtIssuer().issue(accessTokenData, context.delegation);

    } else {

        issuedAccessToken = context.accessTokenIssuer.issue(accessTokenData, context.delegation);
    }

    return {
        scope: accessTokenData.scope,
        access_token: issuedAccessToken,
        token_type: 'bearer',
        expires_in: secondsUntil(accessTokenData.exp)
    };
}

The long lived access token has claims similar to this. The scope has been reduced to trigger_payments alone, which is a scope designed to only be used at a single endpoint for consumers. The access token also has also been given a long expiry time of 1 year in the future:

{
 "jti": "afa3d2b3-3b97-4a0a-9e31-ad8e69aac9ba",
 "delegationId": "e39a700d-3f3d-469e-a72f-aa02d55d5d54",
 "exp": 1688751570,
 "nbf": 1657215570,
 "scope": "trigger_payments",
 "iss": "http://localhost:8443/oauth/v2/oauth-anonymous",
 "sub": "demouser",
 "aud": "api.example.com",
 "iat": 1657215570,
 "purpose": "access_token",
 "order_transaction_id": "1b6d215b-7ce2-4e0b-9079-f4e1266f57b1",
 "event_payload_hash": "7e6d3d4b2608625f144f9c1a988da170504a368b647a6609ac4ec6c939496be1"
}

The overall publishing code looks like this in the OrdersService class and is straightforward to code:

export async function publishOrderCreated(orderTransaction: OrderTransaction, accessToken: string, producer: Kafka.Producer) {

    const payload = {
        orderTransactionID: orderTransaction.orderTransactionID,
        userID: orderTransaction.userID,
        utcTime: orderTransaction.utcTime.getTime(),
        items: orderTransaction.items,
    };

    console.log('Performing Token Exchange ...');

    const eventPayloadHash = hash.sha256(JSON.stringify(payload));
    const longLivedReducedScopeAccessToken = await tokenExchange(accessToken, orderTransaction.orderTransactionID, eventPayloadHash);

    const orderCreatedEvent = {
        accessToken: longLivedReducedScopeAccessToken,
        payload,
    } as OrderCreatedEvent;

    console.log('Publishing OrderCreated Event ...');

    producer.produce('OrderCreated', null, Buffer.from(JSON.stringify(orderCreatedEvent)));
}

Event Consuming with JWT Validation

The example APIs already perform access token JWT validation for HTTP requests, using the following authorizer function:

async function authorize(accessToken: string): Promise<ClaimsPrincipal> {

    const options = {
        algorithms: [oauthConfiguration.algorithm],
        issuer: oauthConfiguration.issuer,
        audience: oauthConfiguration.audience,
    };
    
    let result: JWTVerifyResult;
    try {
        result = await jwtVerify(accessToken, remoteJWKSet, options);
    } catch (e: any) {
        throw new OrderServiceError(401, 'authentication_error', 'Missing, invalid or expired access token', e)
    }

    const claimsPrincipal: ClaimsPrincipal = {
        userID: result.payload.sub as string,
        scope: (result.payload.scope as string).split(' '),
    }

    return claimsPrincipal;
}

The same function is then reused in the Payments API before processing OrderCreated event messages, where the overall process contains the following logic. Only event messages containing an access token that is cryptographically signed with the private key of the Identity Server are accepted:

const consumer = new Kafka.KafkaConsumer({
    'group.id': 'payments-api-consumer',
    'client.id': 'payments-api-consumer',
    'metadata.broker.list': host,
    event_cb: true,
    }, {});
consumer
    .on('data', async (message: any) => {

        const orderEvent = JSON.parse(message.value.toString());
        console.log(`Payments API is consuming an OrderCreated event ...`);

        try {
            const claims = await authorize(orderEvent.accessToken);
            createPaymentTransaction(orderEvent, claims);

        } catch (e: any) {

            const error = e as PaymentServiceError;
            logError(error);
        }
    });

The Payments API then makes some additional authorization checks specific to its OrderCreated event processing endpoint. Firstly the trigger_payments scope is only used for this particular type of event message, and would be rejected if received at the API's HTTP endpoints. Secondly the hash of the event data is compared against the JWT's event_payload_hash claim. If the received event data hash does not match the JWT claim, the incoming event is rejected:

export function authorizePayment(event: OrderCreatedEvent, claims: ClaimsPrincipal) {

    if (claims.scope.indexOf('trigger_payments') === -1) {
        throw new PaymentServiceError(403, 'authorization_error', 'The token has insufficient scope');
    }

    const eventPayloadHash = hash.sha256(JSON.stringify(event.payload));
    if (claims.eventPayloadHash != eventPayloadHash) {
        throw new PaymentServiceError(403, 'invalid_event_message', 'The event message contains an unexpected payload');
    }

    if (claims.orderTransactionID !== event.payload.orderTransactionID) {
        throw new PaymentServiceError(403, 'invalid_event_transaction', 'The event message contain unexpected transaction data');
    }
}

All of this ensures that a malicious party cannot send event messages to the Payments API, or change the event data of genuine messages. This provides strong protection against internal threats, to ensure zero trust.

Event Payload Serialization

Care is needed to ensure that the publisher and consumer can calculate the hash of the event payload in the same way. The code example uses the same technology stack for both APIs, and will always produce the same string for the same input object. In some setups, such as when publishing and consuming APIs use different technology stacks, you may need to look into alternative options, such as determistic JSON stringifiers.

Conclusion

This tutorial showed how to use JWT access tokens to secure event messages, so that strong identity guarantees can be used when resuming asynchronous workflows. OAuth secured APIs already verify JWTs issued by the Identity Server when receiving HTTP requests. Therefore the same code and key management infrastructure can be used to deal with event messages, for an easy to manage solution.