fix: tenant s3 credentials fixes and refactor

migrations/multitenant/0017-tenants-s3-credentials-fix-notify-key.sql

@@ -0,0 +1,21 @@
+CREATE OR REPLACE FUNCTION tenants_s3_credentials_update_notify_trigger ()
+    RETURNS TRIGGER
+AS $$
+BEGIN
+    PERFORM
+        pg_notify('tenants_s3_credentials_update', '"' || NEW.tenant_id || ':' || NEW.access_key || '"');
+    RETURN NULL;
+END;
+$$
+    LANGUAGE plpgsql;
+
+CREATE OR REPLACE FUNCTION tenants_s3_credentials_delete_notify_trigger ()
+    RETURNS TRIGGER
+AS $$
+BEGIN
+    PERFORM
+        pg_notify('tenants_s3_credentials_update', '"' || OLD.tenant_id || ':' || OLD.access_key || '"');
+    RETURN NULL;
+END;
+$$
+    LANGUAGE plpgsql;

src/http/error-handler.ts

@@ -64,7 +64,7 @@ export const setErrorHandler = (app: FastifyInstance) => {
     // Fastify errors
     if ('statusCode' in error) {
       const err = error as FastifyError
-      return reply.status((error as any).statusCode || 500).send({
+      return reply.status(err.statusCode || 500).send({
         statusCode: `${err.statusCode}`,
         error: err.name,
         message: err.message,

src/http/plugins/db.ts

@@ -171,7 +171,7 @@ export const migrations = fastifyPlugin(
     })
 
     if (dbMigrationStrategy === MultitenantMigrationStrategy.ON_REQUEST) {
-      const migrationsMutex = createMutexByKey()
+      const migrationsMutex = createMutexByKey<void>()
 
       fastify.addHook('preHandler', async (request) => {
         // migrations are handled via async migrations

src/http/plugins/jwt.ts

@@ -25,7 +25,7 @@ export const jwt = fastifyPlugin(
     fastify.decorateRequest('jwt', '')
     fastify.decorateRequest('jwtPayload', undefined)
 
-    fastify.addHook('preHandler', async (request, reply) => {
+    fastify.addHook('preHandler', async (request) => {
       request.jwt = (request.headers.authorization || '').replace(BEARER, '')
 
       if (!request.jwt && request.routeOptions.config.allowInvalidJwt) {
@@ -41,13 +41,14 @@ export const jwt = fastifyPlugin(
         request.jwtPayload = payload
         request.owner = payload.sub
         request.isAuthenticated = true
-      } catch (err: any) {
+      } catch (e) {
         request.jwtPayload = { role: 'anon' }
         request.isAuthenticated = false
 
         if (request.routeOptions.config.allowInvalidJwt) {
           return
         }
+        const err = e as Error
         throw ERRORS.AccessDenied(err.message, err)
       }
     })

src/http/plugins/signature-v4.ts

@@ -1,6 +1,6 @@
 import { FastifyInstance, FastifyRequest } from 'fastify'
 import fastifyPlugin from 'fastify-plugin'
-import { getJwtSecret, getS3CredentialsByAccessKey, getTenantConfig } from '@internal/database'
+import { getJwtSecret, getTenantConfig, s3CredentialsManager } from '@internal/database'
 import { ClientSignature, SignatureV4 } from '@storage/protocols/s3'
 import { signJWT, verifyJWT } from '@internal/auth'
 import { ERRORS } from '@internal/errors'
@@ -160,7 +160,7 @@ async function createServerSignature(tenantId: string, clientSignature: ClientSi
   }
 
   if (isMultitenant) {
-    const credential = await getS3CredentialsByAccessKey(
+    const credential = await s3CredentialsManager.getS3CredentialsByAccessKey(
       tenantId,
       clientSignature.credentials.accessKey
     )

src/http/routes/admin/jwks.ts

@@ -110,7 +110,7 @@ export default async function routes(fastify: FastifyInstance) {
       }
 
       const result = await jwksManager.addJwk(params.tenantId, body.jwk, body.kind)
-      return reply.send(result)
+      return reply.status(201).send(result)
     }
   )
 

src/http/routes/admin/s3.ts

@@ -1,7 +1,9 @@
 import { FastifyInstance, RequestGenericInterface } from 'fastify'
 import apiKey from '../../plugins/apikey'
-import { createS3Credentials, deleteS3Credential, listS3Credentials } from '@internal/database'
+import { s3CredentialsManager } from '@internal/database'
 import { FromSchema } from 'json-schema-to-ts'
+import { isUuid } from '@storage/limits'
+import { ERRORS } from '@internal/errors'
 
 const createCredentialsSchema = {
   description: 'Create S3 Credentials',
@@ -88,7 +90,7 @@ export default async function routes(fastify: FastifyInstance) {
       schema: createCredentialsSchema,
     },
     async (req, reply) => {
-      const credentials = await createS3Credentials(req.params.tenantId, {
+      const credentials = await s3CredentialsManager.createS3Credentials(req.params.tenantId, {
         description: req.body.description,
         claims: req.body.claims,
       })
@@ -106,8 +108,7 @@ export default async function routes(fastify: FastifyInstance) {
     '/:tenantId/credentials',
     { schema: listCredentialsSchema },
     async (req, reply) => {
-      const credentials = await listS3Credentials(req.params.tenantId)
-
+      const credentials = await s3CredentialsManager.listS3Credentials(req.params.tenantId)
       return reply.send(credentials)
     }
   )
@@ -116,7 +117,10 @@ export default async function routes(fastify: FastifyInstance) {
     '/:tenantId/credentials',
     { schema: deleteCredentialsSchema },
     async (req, reply) => {
-      await deleteS3Credential(req.params.tenantId, req.body.id)
+      if (!isUuid(req.body.id)) {
+        throw ERRORS.InvalidParameter('id not uuid')
+      }
+      await s3CredentialsManager.deleteS3Credential(req.params.tenantId, req.body.id)
 
       return reply.code(204).send()
     }

src/internal/concurrency/mutex.ts

@@ -1,9 +1,9 @@
 import { Semaphore } from '@shopify/semaphore'
 
-export function createMutexByKey() {
+export function createMutexByKey<T>() {
   const semaphoreMap = new Map<string, { semaphore: Semaphore; count: number }>()
 
-  return async (key: string, fn: () => Promise<any>) => {
+  return async (key: string, fn: () => Promise<T>) => {
     let entry = semaphoreMap.get(key)
     if (!entry) {
       entry = { semaphore: new Semaphore(1), count: 0 }

src/internal/database/jwks-manager/manager.ts

@@ -9,7 +9,7 @@ const TENANTS_JWKS_UPDATE_CHANNEL = 'tenants_jwks_update'
 const JWK_KIND_STORAGE_URL_SIGNING = 'storage-url-signing-key'
 const JWK_KID_SEPARATOR = '_'
 
-const tenantJwksMutex = createMutexByKey()
+const tenantJwksMutex = createMutexByKey<JwksConfig>()
 const tenantJwksConfigCache = new Map<string, JwksConfig>()
 
 function createJwkKid({ kind, id }: { id: string; kind: string }): string {

src/internal/database/tenant.ts

@@ -1,16 +1,15 @@
-import crypto from 'node:crypto'
 import { getConfig, JwksConfig, JwksConfigKey, JwksConfigKeyOCT } from '../../config'
-import { decrypt, encrypt, verifyJWT } from '../auth'
+import { decrypt, verifyJWT } from '../auth'
 import { multitenantKnex } from './multitenant-db'
 import { JwtPayload } from 'jsonwebtoken'
 import { PubSubAdapter } from '../pubsub'
 import { createMutexByKey } from '../concurrency'
-import { LRUCache } from 'lru-cache'
-import objectSizeOf from 'object-sizeof'
 import { ERRORS } from '@internal/errors'
 import { DBMigration } from '@internal/database/migrations'
 import { JWKSManager } from './jwks-manager'
 import { JWKSManagerStoreKnex } from './jwks-manager/store-knex'
+import { S3CredentialsManagerStoreKnex } from '../../storage/protocols/s3/credentials-manager/store-knex'
+import { S3CredentialsManager } from '../../storage/protocols/s3/credentials-manager'
 
 interface TenantConfig {
   anonKey?: string
@@ -51,28 +50,16 @@ export enum TenantMigrationStatus {
   FAILED_STALE = 'FAILED_STALE',
 }
 
-interface S3Credentials {
-  accessKey: string
-  secretKey: string
-  claims: { role: string; sub?: string; [key: string]: unknown }
-}
-
 const { isMultitenant, dbServiceRole, serviceKey, jwtSecret } = getConfig()
 
 const tenantConfigCache = new Map<string, TenantConfig>()
 
-const tenantS3CredentialsCache = new LRUCache<string, S3Credentials>({
-  maxSize: 1024 * 1024 * 50, // 50MB
-  ttl: 1000 * 60 * 60, // 1 hour
-  sizeCalculation: (value) => objectSizeOf(value),
-  updateAgeOnGet: true,
-  allowStale: false,
-})
-
-const tenantMutex = createMutexByKey()
-const s3CredentialsMutex = createMutexByKey()
+const tenantMutex = createMutexByKey<TenantConfig>()
 
 export const jwksManager = new JWKSManager(new JWKSManagerStoreKnex(multitenantKnex))
+export const s3CredentialsManager = new S3CredentialsManager(
+  new S3CredentialsManagerStoreKnex(multitenantKnex)
+)
 
 const singleTenantServiceKey:
   | {
@@ -107,15 +94,15 @@ export async function getTenantConfig(tenantId: string): Promise<TenantConfig> {
   }
 
   if (tenantConfigCache.has(tenantId)) {
-    return tenantConfigCache.get(tenantId) as TenantConfig
+    return tenantConfigCache.get(tenantId)!
   }
 
   return tenantMutex(tenantId, async () => {
     if (tenantConfigCache.has(tenantId)) {
-      return tenantConfigCache.get(tenantId) as TenantConfig
+      return tenantConfigCache.get(tenantId)!
     }
 
-    const tenant = await multitenantKnex('tenants').first().where('id', tenantId)
+    const tenant = await multitenantKnex.table('tenants').first().where('id', tenantId)
     if (!tenant) {
       throw ERRORS.MissingTenantConfig(tenantId)
     }
@@ -173,7 +160,7 @@ export async function getTenantConfig(tenantId: string): Promise<TenantConfig> {
     }
     tenantConfigCache.set(tenantId, config)
 
-    return tenantConfigCache.get(tenantId)
+    return tenantConfigCache.get(tenantId)!
   })
 }
 
@@ -249,7 +236,6 @@ export async function getFeatures(tenantId: string): Promise<Features> {
 }
 
 const TENANTS_UPDATE_CHANNEL = 'tenants_update'
-const TENANTS_S3_CREDENTIALS_UPDATE_CHANNEL = 'tenants_s3_credentials_update'
 
 /**
  * Keeps the in memory config cache up to date
@@ -258,130 +244,6 @@ export async function listenForTenantUpdate(pubSub: PubSubAdapter): Promise<void
   await pubSub.subscribe(TENANTS_UPDATE_CHANNEL, (cacheKey) => {
     tenantConfigCache.delete(cacheKey)
   })
-  await pubSub.subscribe(TENANTS_S3_CREDENTIALS_UPDATE_CHANNEL, (cacheKey) => {
-    tenantS3CredentialsCache.delete(cacheKey)
-  })
-
+  await s3CredentialsManager.listenForTenantUpdate(pubSub)
   await jwksManager.listenForTenantUpdate(pubSub)
 }
-
-/**
- * Create S3 Credential for a tenant
- * @param tenantId
- * @param data
- */
-export async function createS3Credentials(
-  tenantId: string,
-  data: { description: string; claims?: S3Credentials['claims'] }
-) {
-  const existingCount = await countS3Credentials(tenantId)
-
-  if (existingCount >= 50) {
-    throw ERRORS.MaximumCredentialsLimit()
-  }
-
-  const secretAccessKeyId = crypto.randomBytes(32).toString('hex').slice(0, 32)
-  const secretAccessKey = crypto.randomBytes(64).toString('hex').slice(0, 64)
-
-  if (data.claims) {
-    delete data.claims.iss
-    delete data.claims.issuer
-    delete data.claims.exp
-    delete data.claims.iat
-  }
-
-  data.claims = {
-    ...(data.claims || {}),
-    role: data.claims?.role ?? dbServiceRole,
-    issuer: `supabase.storage.${tenantId}`,
-    sub: data.claims?.sub,
-  }
-
-  const credentials = await multitenantKnex
-    .table('tenants_s3_credentials')
-    .insert({
-      tenant_id: tenantId,
-      description: data.description,
-      access_key: secretAccessKeyId,
-      secret_key: encrypt(secretAccessKey),
-      claims: JSON.stringify(data.claims),
-    })
-    .returning('id')
-
-  return {
-    id: credentials[0].id,
-    access_key: secretAccessKeyId,
-    secret_key: secretAccessKey,
-  }
-}
-
-export async function getS3CredentialsByAccessKey(
-  tenantId: string,
-  accessKey: string
-): Promise<S3Credentials> {
-  const cacheKey = `${tenantId}:${accessKey}`
-  const cachedCredentials = tenantS3CredentialsCache.get(cacheKey)
-
-  if (cachedCredentials) {
-    return cachedCredentials
-  }
-
-  return s3CredentialsMutex(cacheKey, async () => {
-    const cachedCredentials = tenantS3CredentialsCache.get(cacheKey)
-
-    if (cachedCredentials) {
-      return cachedCredentials
-    }
-
-    const data = await multitenantKnex
-      .table('tenants_s3_credentials')
-      .select('access_key', 'secret_key', 'claims')
-      .where('tenant_id', tenantId)
-      .where('access_key', accessKey)
-      .first()
-
-    if (!data) {
-      throw ERRORS.MissingS3Credentials()
-    }
-
-    const secretKey = decrypt(data.secret_key)
-
-    tenantS3CredentialsCache.set(cacheKey, {
-      accessKey: data.access_key,
-      secretKey: secretKey,
-      claims: data.claims,
-    })
-
-    return {
-      accessKey: data.access_key,
-      secretKey: secretKey,
-      claims: data.claims,
-    }
-  })
-}
-
-export function deleteS3Credential(tenantId: string, credentialId: string) {
-  return multitenantKnex
-    .table('tenants_s3_credentials')
-    .where('tenant_id', tenantId)
-    .where('id', credentialId)
-    .delete()
-    .returning('id')
-}
-
-export function listS3Credentials(tenantId: string) {
-  return multitenantKnex
-    .table('tenants_s3_credentials')
-    .select('id', 'description', 'access_key', 'created_at')
-    .where('tenant_id', tenantId)
-    .orderBy('created_at', 'asc')
-}
-
-export async function countS3Credentials(tenantId: string) {
-  const data = await multitenantKnex
-    .table('tenants_s3_credentials')
-    .count<{ count: number }>('id')
-    .where('tenant_id', tenantId)
-
-  return Number(data?.count || 0)
-}

src/internal/streams/monitor.ts

@@ -12,7 +12,7 @@ export function monitorStream(dataStream: Readable) {
   const byteCounter = createByteCounterStream()
   const span = trace.getActiveSpan()
 
-  let measures: any[] = []
+  let measures: object[] = []
 
   // Handle the 'speed' event to collect speed measurements
   speedMonitor.on('speed', (bps) => {