feat(user-tokens): implement user tokens (#203)

* user token start

* tests

* refactor auth

* caching

* api

* fix test

* comment

* expiry

* batching

* remove user token implementation

* Revert "remove user token implementation"

This reverts commit 1e1dd78.

* refact auth

* remove file

* fix

* refact client auth function

* refact function

* refact

* fix

* remove file

* bring changes

* remove user token api

* revert

* remove correct user token api

* bring changes

* extract constant

* verifyToken

* fix tests

* disable server if no expiremental socket enabled

* fix test

* remove dead code

packages/server/src/api.ts

@@ -31,7 +31,15 @@ export class Api {
     this.sockets = new SocketManager(this.app.sockets)
     this.syncs = new SyncApi(this.router, this.auth, this.app.syncs, this.app.clients, this.app.channels)
     this.health = new HealthApi(this.router, this.auth, this.app.health)
-    this.users = new UserApi(this.router, this.auth, this.app.clients, this.sockets, this.app.users, this.app.sockets)
+    this.users = new UserApi(
+      this.router,
+      this.auth,
+      this.app.clients,
+      this.sockets,
+      this.app.users,
+      this.app.userTokens,
+      this.app.sockets
+    )
     this.conversations = new ConversationApi(
       this.router,
       this.auth,

packages/server/src/app.ts

@@ -22,6 +22,7 @@ import { SocketService } from './socket/service'
 import { StatusService } from './status/service'
 import { StreamService } from './stream/service'
 import { SyncService } from './sync/service'
+import { UserTokenService } from './user-tokens/service'
 import { UserService } from './users/service'
 import { WebhookService } from './webhooks/service'
 
@@ -42,6 +43,7 @@ export class App {
   kvs: KvsService
   conduits: ConduitService
   users: UserService
+  userTokens: UserTokenService
   conversations: ConversationService
   messages: MessageService
   converse: ConverseService
@@ -70,6 +72,7 @@ export class App {
     this.kvs = new KvsService(this.database, this.caching)
     this.conduits = new ConduitService(this.database, this.crypto, this.caching, this.channels, this.providers)
     this.users = new UserService(this.database, this.caching, this.batching)
+    this.userTokens = new UserTokenService(this.database, this.crypto, this.caching, this.batching, this.users)
     this.conversations = new ConversationService(this.database, this.caching, this.batching, this.users)
     this.messages = new MessageService(this.database, this.caching, this.batching, this.conversations)
     this.converse = new ConverseService(this.messages)
@@ -139,6 +142,7 @@ export class App {
     await this.kvs.setup()
     await this.conduits.setup()
     await this.users.setup()
+    await this.userTokens.setup()
     await this.conversations.setup()
     await this.messages.setup()
     await this.converse.setup()

packages/server/src/user-tokens/service.ts

@@ -0,0 +1,139 @@
+import { uuid } from '@botpress/messaging-base'
+import crypto from 'crypto'
+import { v4 as uuidv4 } from 'uuid'
+import yn from 'yn'
+import { Service } from '../base/service'
+import { Batcher } from '../batching/batcher'
+import { BatchingService } from '../batching/service'
+import { ServerCache } from '../caching/cache'
+import { CachingService } from '../caching/service'
+import { CryptoService } from '../crypto/service'
+import { DatabaseService } from '../database/service'
+import { UserService } from '../users/service'
+import { UserTokenTable } from './table'
+import { UserToken } from './types'
+
+export const USER_TOKEN_LENGTH = 66
+
+export class UserTokenService extends Service {
+  public batcher!: Batcher<UserToken>
+
+  private table: UserTokenTable
+  private cacheById!: ServerCache<uuid, UserToken>
+  private cacheTokens!: ServerCache<uuid, string>
+
+  constructor(
+    private db: DatabaseService,
+    private cryptoService: CryptoService,
+    private cachingService: CachingService,
+    private batchingService: BatchingService,
+    private userService: UserService
+  ) {
+    super()
+    this.table = new UserTokenTable()
+  }
+
+  async setup() {
+    if (!yn(process.env.ENABLE_EXPERIMENTAL_SOCKETS)) {
+      // let's not create a table we don't need for now
+      return
+    }
+
+    this.batcher = await this.batchingService.newBatcher(
+      'batcher_user_tokens',
+      [this.userService.batcher],
+      this.handleBatchFlush.bind(this)
+    )
+
+    this.cacheById = await this.cachingService.newServerCache('cache_user_token_by_id')
+    this.cacheTokens = await this.cachingService.newServerCache('cache_user_token_raw')
+
+    await this.db.registerTable(this.table)
+  }
+
+  private async handleBatchFlush(batch: UserToken[]) {
+    const rows = batch.map((x) => this.serialize(x))
+    await this.query().insert(rows)
+  }
+
+  async generateToken(): Promise<string> {
+    return crypto.randomBytes(USER_TOKEN_LENGTH).toString('base64')
+  }
+
+  async create(userId: uuid, token: string, expiry: Date | undefined): Promise<UserToken> {
+    const userToken = {
+      id: uuidv4(),
+      userId,
+      token: await this.cryptoService.hash(token),
+      expiry
+    }
+
+    await this.batcher.push(userToken)
+    this.cacheById.set(userToken.id, userToken)
+
+    return userToken
+  }
+
+  async getById(id: uuid): Promise<UserToken | undefined> {
+    const cached = this.cacheById.get(id)
+    if (cached) {
+      return cached
+    }
+
+    await this.batcher.flush()
+    const rows = await this.query().where({ id })
+
+    if (rows?.length) {
+      const userToken = this.deserialize(rows[0])
+      this.cacheById.set(id, userToken)
+      return userToken
+    } else {
+      return undefined
+    }
+  }
+
+  async verifyToken(id: string, token: string): Promise<UserToken | undefined> {
+    const userToken = await this.getById(id)
+    if (!userToken) {
+      return undefined
+    }
+
+    if (userToken.expiry && Date.now() > userToken.expiry.getTime()) {
+      return undefined
+    }
+
+    const cachedToken = this.cacheTokens.get(id)
+    if (cachedToken) {
+      if (token === cachedToken) {
+        return userToken
+      } else {
+        return undefined
+      }
+    }
+
+    if (await this.cryptoService.compareHash(userToken.token, token)) {
+      this.cacheTokens.set(id, token)
+      return userToken
+    } else {
+      return undefined
+    }
+  }
+
+  public query() {
+    return this.db.knex(this.table.id)
+  }
+
+  private serialize(userToken: Partial<UserToken>) {
+    return {
+      ...userToken,
+      expiry: this.db.setDate(userToken.expiry)
+    }
+  }
+
+  private deserialize(userToken: any): UserToken {
+    return {
+      ...userToken,
+      expiry: userToken.expiry ? this.db.getDate(userToken.expiry) : undefined
+    }
+  }
+}

packages/server/src/user-tokens/table.ts

@@ -0,0 +1,15 @@
+import { Knex } from 'knex'
+import { Table } from '../base/table'
+
+export class UserTokenTable extends Table {
+  get id() {
+    return 'msg_user_tokens'
+  }
+
+  create(table: Knex.CreateTableBuilder) {
+    table.uuid('id').primary()
+    table.uuid('userId').references('id').inTable('msg_users')
+    table.string('token').notNullable()
+    table.timestamp('expiry').nullable()
+  }
+}

packages/server/src/user-tokens/types.ts

@@ -0,0 +1,8 @@
+import { uuid } from '@botpress/messaging-base'
+
+export interface UserToken {
+  id: uuid
+  userId: uuid
+  token: string
+  expiry: Date | undefined
+}

packages/server/src/users/api.ts

@@ -4,6 +4,7 @@ import { Auth } from '../base/auth/auth'
 import { ClientService } from '../clients/service'
 import { SocketManager } from '../socket/manager'
 import { SocketService } from '../socket/service'
+import { UserTokenService } from '../user-tokens/service'
 import { AuthUserSocketSchema, GetUserSchema } from './schema'
 import { UserService } from './service'
 
@@ -14,6 +15,7 @@ export class UserApi {
     private clients: ClientService,
     private sockets: SocketManager,
     private users: UserService,
+    private userTokens: UserTokenService,
     private socketService: SocketService
   ) {}
 
@@ -54,21 +56,37 @@ export class UserApi {
         return this.sockets.reply(socket, message, { error: true, message: error.message })
       }
 
-      const { clientId, userId, userToken }: { clientId: uuid; userId: uuid; userToken: string } = message.data
+      const { clientId, id: userId, token: userTokenRaw }: { clientId: uuid; id: uuid; token: string } = message.data
 
       const client = await this.clients.getById(clientId)
       if (!client) {
         return this.sockets.reply(socket, message, { error: true, message: 'client not found' })
       }
 
-      // TODO: use user token to validate user
-      let user = userId && (await this.users.get(userId))
+      // TODO: refactor here
+
+      let success = true
+      let user = userId ? await this.users.get(userId) : undefined
+      let token = undefined
+
       if (!user || user.clientId !== client.id) {
-        user = await this.users.create(client.id)
+        success = false
+      } else {
+        const [userTokenId, userTokenToken] = userTokenRaw.split('.')
+        if (!(await this.userTokens.verifyToken(userTokenId, userTokenToken))) {
+          success = false
+        }
+      }
+
+      if (!success) {
+        user = await this.users.create(clientId)
+        const tokenRaw = await this.userTokens.generateToken()
+        const userToken = await this.userTokens.create(user.id, tokenRaw, undefined)
+        token = `${userToken.id}.${tokenRaw}`
       }
 
-      this.socketService.registerForUser(socket, user.id)
-      this.sockets.reply(socket, message, user)
+      this.socketService.registerForUser(socket, user!.id)
+      this.sockets.reply(socket, message, { id: user!.id, token })
     })
   }
 }

packages/server/src/users/schema.ts

@@ -2,8 +2,8 @@ import Joi from 'joi'
 
 export const AuthUserSocketSchema = Joi.object({
   clientId: Joi.string().guid().required(),
-  userId: Joi.string().guid().optional(),
-  userToken: Joi.string().optional()
+  id: Joi.string().guid().optional(),
+  token: Joi.string().optional()
 })
 
 export const GetUserSchema = Joi.object({