Auth handling in synchronizer-ws-server-durable-object

[legowhales]

The new synchronizer for cloudflare is awesome!

Quick question: how would you handle authentication?
Is there an easy way to send a token from the client to the server, and in the server throw an expired or unauthorized response based on the token and other params?

Thank's!

[Stewartarmbrecht]

Agreed. The Cloudflare durable object synchronizer is AWESOME!

I also have the same question on the best way to implement authentication.

[Stewartarmbrecht]

I got a working solution. Not sure if it is production caliber yet.
You just need to

1. Pass your token to the durable object via a search param.
2. Create a custom fetch method that validates the token or throws a 401.

Apologies for the excess code but I don't have time to edit it to simplify it:

WebSocketClient:

import ReconnectingWebSocket from 'reconnecting-websocket';

import { useCurrentUser } from '@boundbybetter/auth';
import { logCall, logError, logSetup } from '@boundbybetter/shared';
import { createContext, useContext, useEffect, useState } from 'react';

const SERVER = process.env.EXPO_PUBLIC_CLOUDFLARE_WORKERS_URL;  // "wss://my-durable-object.workers.dev"

export interface WebSocketContextType {
  webSocket: ReconnectingWebSocket | undefined;
  status: WebSocketStatus;
}

export enum WebSocketStatus {
  Undefined = 'undefined',
  Initializing = 'initializing',
  Creating = 'creating',
  Created = 'created',
  Connecting = 'connecting',
  Connected = 'connected',
  Disconnected = 'disconnected',
  Stopped = 'stopped',
  Error = 'error',
}
export const WebSocketContext = createContext<WebSocketContextType>({
  webSocket: undefined,
  status: WebSocketStatus.Undefined,
});

export const useWebSocketClient = (): ReconnectingWebSocket | undefined => {
  const context = useContext(WebSocketContext);
  return context?.webSocket as ReconnectingWebSocket | undefined;
};

export const useWebSocketStatus = (): WebSocketStatus => {
  const context = useContext(WebSocketContext);
  return context?.status;
};

export const WebSocketProvider = (props: { children: React.ReactNode }) => {
  const user = useCurrentUser();
  const [webSocket, setWebSocket] = useState<ReconnectingWebSocket | undefined>(
    undefined,
  );
  const [status, setStatus] = useState<WebSocketStatus>(
    WebSocketStatus.Undefined,
  );
  logSetup('WebSocketProvider');

  useEffect(() => {
    setStatus(WebSocketStatus.Initializing);
    const initializeWebSocket = async () => {
      try {
        setStatus(WebSocketStatus.Creating);

        // Add authorization token to WebSocket URL
        // I'm setting the sync level at the individual user for cross device sync in an expo app.
        const wsUrl = new URL(SERVER + '/' + user.userId); 
       // Add the idToken returned by Microsoft Entra External Customer tenant.
        wsUrl.searchParams.append('token', `${user.idToken}`);

        // Initialize the client with the auth token
        const client = new ReconnectingWebSocket(wsUrl.toString());

        setWebSocket(client);
        setStatus(WebSocketStatus.Created);

        client.onopen = (event) => {
          setStatus(WebSocketStatus.Connected);
          logCall('WebSocketProvider', 'Connected to WebPubSub', event);
        };

        client.onclose = (event) => {
          // Check if closure was due to auth failure
          if (event.code === 1008) {
            setStatus(WebSocketStatus.Error);
            logError('WebSocketProvider', 'Authentication failed', event);
          } else {
            setStatus(WebSocketStatus.Disconnected);
            logCall('WebSocketProvider', 'Disconnected from WebPubSub', event);
          }
        };

        client.onmessage = (event) => {
          logCall('WebSocketProvider', 'Received message', event);
        };

        client.onerror = (event) => {
          setStatus(WebSocketStatus.Error);
          logCall('WebSocketProvider', 'WebSocket error', event);
        };
      } catch (error) {
        setStatus(WebSocketStatus.Error);
        logError('WebSocketProvider', 'initializeWebSocket', error);
      }
    };

    if (user?.idToken) {
      initializeWebSocket();
    }

    return () => {
      if (webSocket) {
        webSocket.close();
      }
    };
  }, [user]); // Re-run if user changes

  return (
    <WebSocketContext.Provider value={{ webSocket, status }}>
      {props.children}
    </WebSocketContext.Provider>
  );
};

Durable Object:

      import { createMergeableStore, Id, IdAddedOrRemoved } from 'tinybase';
      import { createDurableObjectStoragePersister } from 'tinybase/persisters/persister-durable-object-storage';
      import {
        getWsServerDurableObjectFetch,
        WsServerDurableObject,
      } from 'tinybase/synchronizers/synchronizer-ws-server-durable-object';
      import * as jose from 'jose';
      
      // Whether to persist data in the Durable Object between client sessions.
      //
      // If false, the Durable Object only provides synchronization between clients
      // (which are assumed to persist their own data).
      const PERSIST_TO_DURABLE_OBJECT = true;
      
      export class GroupSyncServer extends WsServerDurableObject {
        onPathId(pathId: Id, addedOrRemoved: IdAddedOrRemoved) {
          console.info((addedOrRemoved ? 'Added' : 'Removed') + ` path ${pathId}`);
        }
      
        onClientId(pathId: Id, clientId: Id, addedOrRemoved: IdAddedOrRemoved) {
          console.info(
            (addedOrRemoved ? 'Added' : 'Removed') +
              ` client ${clientId} on path ${pathId}`,
          );
        }
      
        createPersister() {
          if (PERSIST_TO_DURABLE_OBJECT) {
            return createDurableObjectStoragePersister(
              createMergeableStore(),
              this.ctx.storage,
            );
          }
        }
      }
      
      // Custom fetch handler that validates the token before upgrading to WebSocket
      const customFetch = async (request: Request, env: any) => {
        const url = new URL(request.url);
        const token = url.searchParams.get('token');
        // In my local dev environment I bypass the login flow for speed.
        const isTest = token === 'test-user-1' && env.IS_DEV === 'true';
      
        if (!token) {
          return new Response('Unauthorized - Missing token', { status: 401 });
        }
      
        if (isTest) {
          return getWsServerDurableObjectFetch('GroupSyncServers')(request, env);
        }
      
        try {
          // Get JWKS from Microsoft Entra ID - Update the URL to include the .well-known/openid-configuration/jwks endpoint
          // https://boundbybettercustomers.ciamlogin.com/0009cc7a-b831-4911-be61-58865b14fccb
          const jwksUrl = new URL(`${env.AUTH_PROVIDER_URL}/discovery/v2.0/keys`);
          console.log('JWKS URL:', jwksUrl.toString());
          const JWKS = jose.createRemoteJWKSet(jwksUrl);
      
          // Verify token using JWKS client
          const decoded = await jose.jwtVerify(token, JWKS);
          console.log('decoded', decoded);
      
          const userId = request.url.split('/').pop()?.split('?')[0];
      
          console.log('userId', userId);
          console.log('decoded.payload.oid', decoded.payload.oid);
      
          if (!userId || decoded.payload.oid !== userId) {
            return new Response('Unauthorized - Invalid token', { status: 401 });
          }
      
          // If token is valid, proceed with WebSocket upgrade
          return getWsServerDurableObjectFetch('GroupSyncServers')(request, env);
        } catch (error) {
          console.error('error', error);
          return new Response('Unauthorized - Invalid token', { status: 401 });
        }
      };
      
      export default {
        fetch: customFetch,
      };

[Stewartarmbrecht]

Here is my toml file:

#:schema node_modules/wrangler/config-schema.json
name = "sync-server-dev"
main = "index.ts"
compatibility_date = "2024-10-11"
route = { pattern = "sync-dev.boundbybetter.com", custom_domain = true }

[[durable_objects.bindings]]
name = "GroupSyncServers"
class_name = "GroupSyncServer"

[[migrations]]
tag = "v1"
new_classes = ["GroupSyncServer"]

# wrangler.toml (wrangler v3.88.0^)
[observability.logs]
enabled = true

# Add IS_DEV env variable
[vars]
IS_DEV = "false"
AUTH_PROVIDER_URL = "https://boundbybettercustomers.ciamlogin.com/0009cc7a-b831-4911-be61-58865b14fccb"

Here is my AuthProvider:

import React, { useState } from 'react';
import { LogInScreen } from './LogInScreen';
import { logCall } from '@boundbybetter/shared';
import { User } from './User';
import { AuthContext } from './AuthContext';

export const AuthProvider: React.FC<{ children: JSX.Element }> = ({
  children,
}) => {
  const skipAuth = process.env.EXPO_PUBLIC_SKIP_AUTH === 'true';
  const testUser: User = {
    userId: 'test-user-1',
    userName: 'Test User',
    userEmail: '[email protected]',
    groups: [],
    accessToken: 'test-user-1',
    idToken: 'test-user-1',
  };
  const initialUser = skipAuth ? testUser : null;
  const [currentUser, setCurrentUser] = useState<User | null>(initialUser);

  logCall('AuthProvider', currentUser);

  const isLoggedIn = () => {
    return currentUser?.userId !== undefined;
  };

  // Method to clear the current user
  const clearCurrentUser = () => {
    setCurrentUser(null);
    logCall('AuthProvider', 'User cleared from context');
  };

  return (
    <AuthContext.Provider
      value={{ currentUser, setCurrentUser, clearCurrentUser }}
    >
      {isLoggedIn() ? children : <LogInScreen onLogin={setCurrentUser} />}
    </AuthContext.Provider>
  );
};

and my LoginScreen uses expo-auth-session:

import * as WebBrowser from 'expo-web-browser';
import React from 'react';
import { tg } from '@boundbybetter/ui';
import { logMessage } from '@boundbybetter/shared';
import { JwtPayload, jwtDecode } from 'jwt-decode';
import Constants from 'expo-constants';

import {
  makeRedirectUri,
  useAutoDiscovery,
  useAuthRequest,
  exchangeCodeAsync,
} from 'expo-auth-session';
import { User } from './User';

WebBrowser.maybeCompleteAuthSession();

// Define a custom interface that extends JwtPayload
interface CustomJwtPayload extends JwtPayload {
  name: string; // Add the expected properties
  email: string; // Add the expected properties
  oid: string;
  upn: string;
  groups?: string[];
}

interface LogInScreenProps {
  onLogin: (user: User) => void;
}

export const LogInScreen = ({ onLogin }: LogInScreenProps) => {
  // const { user, getAccessToken, signIn } = useAuth();

  const theme = tg.useTheme();
  const variant = Constants.expoConfig?.extra?.appVariant || 'nope';
  const redirectUri = makeRedirectUri({
    scheme:
      variant === 'prod'
        ? 'com.boundbybetter.bettertemplateapp'
        : `com.boundbybetter.bettertemplateapp.${variant}`,
    path: 'auth',
  });

  // Endpoint
  const discovery = useAutoDiscovery(process.env.EXPO_PUBLIC_AUTHORITY);

  const clientId = process.env.EXPO_PUBLIC_CLIENT_ID;
  // Request;
  const [authRequest, authResponse, promptAsync] = useAuthRequest(
    {
      clientId,
      scopes: ['openid', 'profile', 'email', 'offline_access'],
      redirectUri,
    },
    discovery,
  );

  const logIn = async (method: string) => {
    if (method === 'Email') {
      const codeResponse = await promptAsync();
      logMessage('codeResponse', codeResponse);
      if (authRequest && codeResponse?.type === 'success' && discovery) {
        logMessage('response useAuthRequest', authResponse);
        const exchangeCodeResponse = await exchangeCodeAsync(
          {
            clientId,
            code: codeResponse.params.code,
            extraParams: authRequest.codeVerifier
              ? { code_verifier: authRequest.codeVerifier }
              : /* istanbul ignore next */ undefined,
            redirectUri,
            scopes: [
              'openid',
              'profile',
              'email',
              'offline_access',
              'Group.Read.All',
              'User.Read',
            ],
          },
          discovery,
        );
        logMessage('response', exchangeCodeResponse);
        // Get the user id from the idToken.
        const idToken = jwtDecode(
          exchangeCodeResponse.idToken,
        ) as CustomJwtPayload;
        logMessage('idToken', idToken);
        const newUser: User = {
          userId: idToken.oid,
          userName: idToken.name,
          userEmail: idToken.email,
          groups: idToken.groups || [],
          idToken: exchangeCodeResponse.idToken,
        };
        onLogin(newUser); // Update the user in AuthProvider
      }
    } else {
      // TODO: Remove buttons and logic for other providers.  Instead just implement each provider in Microsoft Entra.
      onLogin(null);
    }
  };
  return (
    <tg.YStack
      padding="$4"
      space="$2"
      testID="app-log-in"
      backgroundColor={theme.background}
      flex={1}
    >
      <tg.H1>Better Template</tg.H1>
      <tg.H3>Log In</tg.H3>
      <tg.Paragraph>Please select a method for logging in.</tg.Paragraph>
      <tg.Button
        width="$14"
        onPress={() => logIn('Email')}
        disabled={!authRequest}
      >
        Email
      </tg.Button>
      <tg.Button
        width="$14"
        onPress={/* istanbul ignore next */ () => logIn('Microsoft')}
      >
        Microsoft
      </tg.Button>
      <tg.Button
        width="$14"
        onPress={/* istanbul ignore next */ () => logIn('Google')}
      >
        Google
      </tg.Button>
      <tg.Button
        width="$14"
        onPress={/* istanbul ignore next */ () => logIn('Facebook')}
      >
        Facebook
      </tg.Button>
    </tg.YStack>
  );
};

[Stewartarmbrecht]

Oh and my Sync Provider: :)

import { logCall, logError, logSetup } from '@boundbybetter/shared';
import { useCreateSynchronizer } from 'tinybase/ui-react';
import { useCurrentUser } from '@boundbybetter/auth';
import { useStore } from '../store/useStore';
import { createContext, useContext, useState } from 'react';
import {
  createWsSynchronizer,
  WsSynchronizer,
} from 'tinybase/synchronizers/synchronizer-ws-client';
import ReconnectingWebSocket from 'reconnecting-websocket';
import { useWebSocketClient } from '../webSocket/WebSocketProvider';

export enum CloudflareSyncStatus {
  Undefined = 'undefined',
  Initializing = 'initializing',
  Starting = 'starting',
  InboundSyncing = 'inboundSyncing',
  OutboundSyncing = 'outboundSyncing',
  Idle = 'idle',
  Stopping = 'stopping',
  Error = 'error',
}

export interface CloudflareSyncContextType {
  synchronizer: WsSynchronizer<ReconnectingWebSocket> | undefined;
  status: CloudflareSyncStatus;
}

export const CloudflareSyncContext = createContext<CloudflareSyncContextType>({
  synchronizer: undefined,
  status: CloudflareSyncStatus.Undefined,
});

export const useCloudflareSynchronizer = () => {
  const context = useContext(CloudflareSyncContext);
  return context?.synchronizer;
};

export const useCloudflareSyncStatus = () => {
  const context = useContext(CloudflareSyncContext);
  return context?.status;
};

export interface CloudflareSyncProviderProps {
  children: JSX.Element;
}

export const CloudflareSyncProvider = (props: CloudflareSyncProviderProps) => {
  const user = useCurrentUser();
  const store = useStore();
  const webSocketClient = useWebSocketClient();
  const [status, setStatus] = useState<CloudflareSyncStatus>(
    CloudflareSyncStatus.Undefined,
  );
  logSetup('CloudflareSyncProvider');

  const synchronizer = useCreateSynchronizer(
    store,
    async (store) => {
      try {
        if (!user?.idToken || !webSocketClient) {
          return undefined;
        }

        setStatus(CloudflareSyncStatus.Initializing);
        const newSynchronizer = await createWsSynchronizer(
          store,
          webSocketClient,
          1,
        );

        newSynchronizer.addStatusListener((synchronizer, status) => {
          const newStatus =
            status === 0
              ? CloudflareSyncStatus.Idle
              : status === 1
                ? CloudflareSyncStatus.InboundSyncing
                : CloudflareSyncStatus.OutboundSyncing;

          setStatus(newStatus);
          logCall(
            'CloudflareSyncProvider',
            'statusListener',
            `Synchronizer status: ${newStatus}`,
            'Stats',
            synchronizer.getStats(),
          );
        });

        await newSynchronizer.startSync();

        // If the websocket reconnects in the future, do another explicit sync.
        webSocketClient.addEventListener('open', () => {
          newSynchronizer.load().then(() => newSynchronizer.save());
        });

        return newSynchronizer;
      } catch (error) {
        setStatus(CloudflareSyncStatus.Error);
        logError('CloudflareSyncProvider', 'createSynchronizer', error);
        return undefined;
      }
    },
    [user?.idToken, webSocketClient],
    async (synchronizer) => {
      setStatus(CloudflareSyncStatus.Stopping);
      synchronizer.stopSync();
    },
    [],
  );

  return (
    <CloudflareSyncContext.Provider value={{ synchronizer, status }}>
      {props.children}
    </CloudflareSyncContext.Provider>
  );
};

It's all wrapped up in a State Provider:

import { logSetup } from '@boundbybetter/shared';
import { StoreProvider } from './store/StoreProvider';
import { PersistenceProvider } from './persistence/PersistenceProvider';
import { WebSocketProvider } from './webSocket/WebSocketProvider';
import { CloudflareSyncProvider } from './sync/CloudflareSyncProvider';

export interface StateProviderProps {
  children: JSX.Element;
}

export const StateProvider = (props: StateProviderProps) => {
  logSetup('StateProvider');
  return (
    <WebSocketProvider>
      <StoreProvider>
        <PersistenceProvider>
          <CloudflareSyncProvider>{props.children}</CloudflareSyncProvider>
        </PersistenceProvider>
      </StoreProvider>
    </WebSocketProvider>
  );
};

The State Provider is wrapped in my Auth Provider within my applications.