fix: allow access to any workspace with system email (#7865)

Signed-off-by: Alexander Onnikov <Alexander.Onnikov@xored.com>

dev/tool/src/index.ts

@@ -1239,7 +1239,6 @@ export function devTool (
       }
 
       toolCtx.info('using datalake', { datalake: datalakeConfig })
-      const datalake = createDatalakeClient(datalakeConfig as DatalakeConfig)
 
       let workspaces: Workspace[] = []
       await withAccountDatabase(async (db) => {
@@ -1280,10 +1279,11 @@ export function devTool (
         })
         const workspaceId = getWorkspaceId(workspace.workspace)
         const token = generateToken(systemAccountEmail, workspaceId)
+        const datalake = createDatalakeClient(datalakeConfig as DatalakeConfig, token)
 
         for (const config of storages) {
           const storage = new S3Service(config)
-          await copyToDatalake(toolCtx, workspaceId, config, storage, datalake, token, params)
+          await copyToDatalake(toolCtx, workspaceId, config, storage, datalake, params)
         }
       }
     })

dev/tool/src/storage.ts

@@ -271,7 +271,6 @@ export async function copyToDatalake (
   config: S3Config,
   adapter: S3Service,
   datalake: DatalakeClient,
-  token: string,
   params: CopyDatalakeParams
 ): Promise<void> {
   console.log('copying from', config.name, 'concurrency:', params.concurrency)
@@ -312,7 +311,7 @@ export async function copyToDatalake (
   let cursor: string | undefined = ''
   let hasMore = true
   while (hasMore) {
-    const res = await datalake.listObjects(ctx, token, workspaceId, cursor, 1000)
+    const res = await datalake.listObjects(ctx, workspaceId, cursor, 1000)
     cursor = res.cursor
     hasMore = res.cursor !== undefined
     for (const blob of res.blobs) {
@@ -350,7 +349,7 @@ export async function copyToDatalake (
               ctx,
               5,
               async () => {
-                await copyBlobToDatalake(ctx, workspaceId, blob, config, adapter, datalake, token)
+                await copyBlobToDatalake(ctx, workspaceId, blob, config, adapter, datalake)
                 processedCnt += 1
                 processedSize += blob.size
               },
@@ -379,8 +378,7 @@ export async function copyBlobToDatalake (
   blob: ListBlobResult,
   config: S3Config,
   adapter: S3Service,
-  datalake: DatalakeClient,
-  token: string
+  datalake: DatalakeClient
 ): Promise<void> {
   const objectName = blob._id
   if (blob.size < 1024 * 1024 * 64) {
@@ -392,7 +390,7 @@ export async function copyBlobToDatalake (
     const url = concatLink(endpoint, `${bucketId}/${objectId}`)
 
     const params = { url, accessKeyId, secretAccessKey, region }
-    await datalake.uploadFromS3(ctx, token, workspaceId, objectName, params)
+    await datalake.uploadFromS3(ctx, workspaceId, objectName, params)
   } else {
     // Handle huge file
     const stat = await adapter.stat(ctx, workspaceId, objectName)
@@ -407,7 +405,7 @@ export async function copyBlobToDatalake (
       const readable = await adapter.get(ctx, workspaceId, objectName)
       try {
         console.log('uploading huge blob', objectName, Math.round(stat.size / 1024 / 1024), 'MB')
-        await uploadMultipart(ctx, token, datalake, workspaceId, objectName, readable, metadata)
+        await uploadMultipart(ctx, datalake, workspaceId, objectName, readable, metadata)
         console.log('done', objectName)
       } finally {
         readable.destroy()
@@ -418,7 +416,6 @@ export async function copyBlobToDatalake (
 
 function uploadMultipart (
   ctx: MeasureContext,
-  token: string,
   datalake: DatalakeClient,
   workspaceId: WorkspaceId,
   objectName: string,
@@ -449,7 +446,7 @@ function uploadMultipart (
     stream.pipe(passthrough)
 
     datalake
-      .uploadMultipart(ctx, token, workspaceId, objectName, passthrough, metadata)
+      .uploadMultipart(ctx, workspaceId, objectName, passthrough, metadata)
       .then(() => {
         cleanup()
         resolve()

server/datalake/src/client.ts

@@ -80,7 +80,14 @@ export interface R2UploadParams {
 
 /** @public */
 export class DatalakeClient {
-  constructor (private readonly endpoint: string) {}
+  private readonly headers: Record<string, string>
+
+  constructor (
+    private readonly endpoint: string,
+    private readonly token: string
+  ) {
+    this.headers = { Authorization: 'Bearer ' + token }
+  }
 
   getObjectUrl (ctx: MeasureContext, workspace: WorkspaceId, objectName: string): string {
     const path = `/blob/${workspace.name}/${encodeURIComponent(objectName)}`
@@ -89,7 +96,6 @@ export class DatalakeClient {
 
   async listObjects (
     ctx: MeasureContext,
-    token: string,
     workspace: WorkspaceId,
     cursor: string | undefined,
     limit: number = 100
@@ -101,16 +107,16 @@ export class DatalakeClient {
       url.searchParams.append('cursor', cursor)
     }
 
-    const response = await fetchSafe(ctx, url, { headers: { Authorization: 'Bearer ' + token } })
+    const response = await fetchSafe(ctx, url, { headers: { ...this.headers } })
     return (await response.json()) as ListObjectOutput
   }
 
-  async getObject (ctx: MeasureContext, token: string, workspace: WorkspaceId, objectName: string): Promise<Readable> {
+  async getObject (ctx: MeasureContext, workspace: WorkspaceId, objectName: string): Promise<Readable> {
     const url = this.getObjectUrl(ctx, workspace, objectName)
 
     let response
     try {
-      response = await fetchSafe(ctx, url, { headers: { Authorization: 'Bearer ' + token } })
+      response = await fetchSafe(ctx, url, { headers: { ...this.headers } })
     } catch (err: any) {
       if (err.name !== 'NotFoundError') {
         console.error('failed to get object', { workspace, objectName, err })
@@ -128,7 +134,6 @@ export class DatalakeClient {
 
   async getPartialObject (
     ctx: MeasureContext,
-    token: string,
     workspace: WorkspaceId,
     objectName: string,
     offset: number,
@@ -136,7 +141,7 @@ export class DatalakeClient {
   ): Promise<Readable> {
     const url = this.getObjectUrl(ctx, workspace, objectName)
     const headers = {
-      Authorization: 'Bearer ' + token,
+      ...this.headers,
       Range: length !== undefined ? `bytes=${offset}-${offset + length - 1}` : `bytes=${offset}`
     }
 
@@ -160,7 +165,6 @@ export class DatalakeClient {
 
   async statObject (
     ctx: MeasureContext,
-    token: string,
     workspace: WorkspaceId,
     objectName: string
   ): Promise<StatObjectOutput | undefined> {
@@ -170,7 +174,7 @@ export class DatalakeClient {
     try {
       response = await fetchSafe(ctx, url, {
         method: 'HEAD',
-        headers: { Authorization: 'Bearer ' + token }
+        headers: { ...this.headers }
       })
     } catch (err: any) {
       if (err.name === 'NotFoundError') {
@@ -192,12 +196,12 @@ export class DatalakeClient {
     }
   }
 
-  async deleteObject (ctx: MeasureContext, token: string, workspace: WorkspaceId, objectName: string): Promise<void> {
+  async deleteObject (ctx: MeasureContext, workspace: WorkspaceId, objectName: string): Promise<void> {
     const url = this.getObjectUrl(ctx, workspace, objectName)
     try {
       await fetchSafe(ctx, url, {
         method: 'DELETE',
-        headers: { Authorization: 'Bearer ' + token }
+        headers: { ...this.headers }
       })
     } catch (err: any) {
       if (err.name !== 'NotFoundError') {
@@ -209,7 +213,6 @@ export class DatalakeClient {
 
   async putObject (
     ctx: MeasureContext,
-    token: string,
     workspace: WorkspaceId,
     objectName: string,
     stream: Readable | Buffer | string,
@@ -230,11 +233,11 @@ export class DatalakeClient {
     try {
       if (size === undefined || size < 64 * 1024 * 1024) {
         return await ctx.with('direct-upload', {}, (ctx) =>
-          this.uploadWithFormData(ctx, token, workspace, objectName, stream, { ...params, size })
+          this.uploadWithFormData(ctx, workspace, objectName, stream, { ...params, size })
         )
       } else {
         return await ctx.with('signed-url-upload', {}, (ctx) =>
-          this.uploadWithSignedURL(ctx, token, workspace, objectName, stream, { ...params, size })
+          this.uploadWithSignedURL(ctx, workspace, objectName, stream, { ...params, size })
         )
       }
     } catch (err) {
@@ -245,7 +248,6 @@ export class DatalakeClient {
 
   async uploadWithFormData (
     ctx: MeasureContext,
-    token: string,
     workspace: WorkspaceId,
     objectName: string,
     stream: Readable | Buffer | string,
@@ -268,7 +270,7 @@ export class DatalakeClient {
     const response = await fetchSafe(ctx, url, {
       method: 'POST',
       body: form,
-      headers: { Authorization: 'Bearer ' + token }
+      headers: { ...this.headers }
     })
 
     const result = (await response.json()) as BlobUploadResult[]
@@ -287,7 +289,6 @@ export class DatalakeClient {
 
   async uploadMultipart (
     ctx: MeasureContext,
-    token: string,
     workspace: WorkspaceId,
     objectName: string,
     stream: Readable | Buffer | string,
@@ -295,41 +296,40 @@ export class DatalakeClient {
   ): Promise<ObjectMetadata> {
     const chunkSize = 10 * 1024 * 1024
 
-    const multipart = await this.multipartUploadStart(ctx, token, workspace, objectName, params)
+    const multipart = await this.multipartUploadStart(ctx, workspace, objectName, params)
 
     try {
       const parts: MultipartUploadPart[] = []
 
       let partNumber = 1
       for await (const chunk of getChunks(stream, chunkSize)) {
-        const part = await this.multipartUploadPart(ctx, token, workspace, objectName, multipart, partNumber, chunk)
+        const part = await this.multipartUploadPart(ctx, workspace, objectName, multipart, partNumber, chunk)
         parts.push(part)
         partNumber++
       }
 
-      return await this.multipartUploadComplete(ctx, token, workspace, objectName, multipart, parts)
+      return await this.multipartUploadComplete(ctx, workspace, objectName, multipart, parts)
     } catch (err: any) {
-      await this.multipartUploadAbort(ctx, token, workspace, objectName, multipart)
+      await this.multipartUploadAbort(ctx, workspace, objectName, multipart)
       throw err
     }
   }
 
   async uploadWithSignedURL (
     ctx: MeasureContext,
-    token: string,
     workspace: WorkspaceId,
     objectName: string,
     stream: Readable | Buffer | string,
     params: UploadObjectParams
   ): Promise<ObjectMetadata> {
-    const url = await this.signObjectSign(ctx, token, workspace, objectName)
+    const url = await this.signObjectSign(ctx, workspace, objectName)
 
     try {
       await fetchSafe(ctx, url, {
         body: stream,
         method: 'PUT',
         headers: {
-          Authorization: 'Bearer ' + token,
+          ...this.headers,
           'Content-Type': params.type,
           'Content-Length': params.size?.toString() ?? '0'
           // 'x-amz-meta-last-modified': metadata.lastModified.toString()
@@ -337,18 +337,17 @@ export class DatalakeClient {
       })
     } catch (err) {
       ctx.error('failed to upload via signed url', { workspace, objectName, err })
-      await this.signObjectDelete(ctx, token, workspace, objectName)
+      await this.signObjectDelete(ctx, workspace, objectName)
       throw new DatalakeError('Failed to upload via signed URL')
     }
 
-    return await this.signObjectComplete(ctx, token, workspace, objectName)
+    return await this.signObjectComplete(ctx, workspace, objectName)
   }
 
   // S3
 
   async uploadFromS3 (
     ctx: MeasureContext,
-    token: string,
     workspace: WorkspaceId,
     objectName: string,
     params: {
@@ -363,7 +362,7 @@ export class DatalakeClient {
     await fetchSafe(ctx, url, {
       method: 'POST',
       headers: {
-        Authorization: 'Bearer ' + token,
+        ...this.headers,
         'Content-Type': 'application/json'
       },
       body: JSON.stringify(params)
@@ -372,18 +371,17 @@ export class DatalakeClient {
 
   // R2
 
-  async getR2UploadParams (ctx: MeasureContext, token: string, workspace: WorkspaceId): Promise<R2UploadParams> {
+  async getR2UploadParams (ctx: MeasureContext, workspace: WorkspaceId): Promise<R2UploadParams> {
     const path = `/upload/r2/${workspace.name}`
     const url = concatLink(this.endpoint, path)
 
-    const response = await fetchSafe(ctx, url, { headers: { Authorization: 'Bearer ' + token } })
+    const response = await fetchSafe(ctx, url, { headers: { ...this.headers } })
     const json = (await response.json()) as R2UploadParams
     return json
   }
 
   async uploadFromR2 (
     ctx: MeasureContext,
-    token: string,
     workspace: WorkspaceId,
     objectName: string,
     params: {
@@ -396,7 +394,7 @@ export class DatalakeClient {
     await fetchSafe(ctx, url, {
       method: 'POST',
       headers: {
-        Authorization: 'Bearer ' + token,
+        ...this.headers,
         'Content-Type': 'application/json'
       },
       body: JSON.stringify(params)
@@ -405,15 +403,10 @@ export class DatalakeClient {
 
   // Signed URL
 
-  private async signObjectSign (
-    ctx: MeasureContext,
-    token: string,
-    workspace: WorkspaceId,
-    objectName: string
-  ): Promise<string> {
+  private async signObjectSign (ctx: MeasureContext, workspace: WorkspaceId, objectName: string): Promise<string> {
     try {
       const url = this.getSignObjectUrl(workspace, objectName)
-      const response = await fetchSafe(ctx, url, { method: 'POST', headers: { Authorization: 'Bearer ' + token } })
+      const response = await fetchSafe(ctx, url, { method: 'POST', headers: { ...this.headers } })
       return await response.text()
     } catch (err: any) {
       ctx.error('failed to sign object', { workspace, objectName, err })
@@ -423,13 +416,12 @@ export class DatalakeClient {
 
   private async signObjectComplete (
     ctx: MeasureContext,
-    token: string,
     workspace: WorkspaceId,
     objectName: string
   ): Promise<ObjectMetadata> {
     try {
       const url = this.getSignObjectUrl(workspace, objectName)
-      const res = await fetchSafe(ctx, url, { method: 'PUT', headers: { Authorization: 'Bearer ' + token } })
+      const res = await fetchSafe(ctx, url, { method: 'PUT', headers: { ...this.headers } })
       return (await res.json()) as ObjectMetadata
     } catch (err: any) {
       ctx.error('failed to complete signed url upload', { workspace, objectName, err })
@@ -437,15 +429,10 @@ export class DatalakeClient {
     }
   }
 
-  private async signObjectDelete (
-    ctx: MeasureContext,
-    token: string,
-    workspace: WorkspaceId,
-    objectName: string
-  ): Promise<void> {
+  private async signObjectDelete (ctx: MeasureContext, workspace: WorkspaceId, objectName: string): Promise<void> {
     try {
       const url = this.getSignObjectUrl(workspace, objectName)
-      await fetchSafe(ctx, url, { method: 'DELETE', headers: { Authorization: 'Bearer ' + token } })
+      await fetchSafe(ctx, url, { method: 'DELETE', headers: { ...this.headers } })
     } catch (err: any) {
       ctx.error('failed to abort signed url upload', { workspace, objectName, err })
       throw new DatalakeError('Failed to abort signed URL upload')
@@ -461,7 +448,6 @@ export class DatalakeClient {
 
   private async multipartUploadStart (
     ctx: MeasureContext,
-    token: string,
     workspace: WorkspaceId,
     objectName: string,
     params: UploadObjectParams
@@ -471,7 +457,7 @@ export class DatalakeClient {
 
     try {
       const headers = {
-        Authorization: 'Bearer ' + token,
+        ...this.headers,
         'Content-Type': params.type,
         'Content-Length': params.size?.toString() ?? '0',
         'Last-Modified': new Date(params.lastModified).toUTCString()
@@ -486,7 +472,6 @@ export class DatalakeClient {
 
   private async multipartUploadPart (
     ctx: MeasureContext,
-    token: string,
     workspace: WorkspaceId,
     objectName: string,
     multipart: MultipartUpload,
@@ -503,7 +488,7 @@ export class DatalakeClient {
       const response = await fetchSafe(ctx, url, {
         method: 'POST',
         body,
-        headers: { Authorization: 'Bearer ' + token }
+        headers: { ...this.headers }
       })
       return (await response.json()) as MultipartUploadPart
     } catch (err: any) {
@@ -514,7 +499,6 @@ export class DatalakeClient {
 
   private async multipartUploadComplete (
     ctx: MeasureContext,
-    token: string,
     workspace: WorkspaceId,
     objectName: string,
     multipart: MultipartUpload,
@@ -529,7 +513,7 @@ export class DatalakeClient {
       const res = await fetchSafe(ctx, url, {
         method: 'POST',
         body: JSON.stringify({ parts }),
-        headers: { Authorization: 'Bearer ' + token }
+        headers: { ...this.headers }
       })
       return (await res.json()) as ObjectMetadata
     } catch (err: any) {
@@ -540,7 +524,6 @@ export class DatalakeClient {
 
   private async multipartUploadAbort (
     ctx: MeasureContext,
-    token: string,
     workspace: WorkspaceId,
     objectName: string,
     multipart: MultipartUpload
@@ -551,7 +534,7 @@ export class DatalakeClient {
     url.searchParams.append('uploadId', multipart.uploadId)
 
     try {
-      await fetchSafe(ctx, url, { method: 'POST', headers: { Authorization: 'Bearer ' + token } })
+      await fetchSafe(ctx, url, { method: 'POST', headers: { ...this.headers } })
     } catch (err: any) {
       ctx.error('failed to abort multipart upload', { workspace, objectName, err })
       throw new DatalakeError('Failed to abort multipart upload')

server/datalake/src/index.ts

@@ -21,7 +21,6 @@ import core, {
   systemAccountEmail,
   withContext
 } from '@hcengineering/core'
-
 import {
   type BlobStorageIterator,
   type BucketInfo,
@@ -47,9 +46,9 @@ export interface DatalakeConfig extends StorageConfig {
 /**
  * @public
  */
-export function createDatalakeClient (opt: DatalakeConfig): DatalakeClient {
+export function createDatalakeClient (opt: DatalakeConfig, token: string): DatalakeClient {
   const endpoint = Number.isInteger(opt.port) ? `${opt.endpoint}:${opt.port}` : opt.endpoint
-  return new DatalakeClient(endpoint)
+  return new DatalakeClient(endpoint, token)
 }
 
 export const CONFIG_KIND = 'datalake'
@@ -61,7 +60,8 @@ export class DatalakeService implements StorageAdapter {
   private readonly client: DatalakeClient
 
   constructor (readonly opt: DatalakeConfig) {
-    this.client = createDatalakeClient(opt)
+    const token = generateToken(systemAccountEmail, { name: '' }, { service: 'datalake' })
+    this.client = createDatalakeClient(opt, token)
   }
 
   async initialize (ctx: MeasureContext, workspaceId: WorkspaceId): Promise<void> {}
@@ -84,10 +84,9 @@ export class DatalakeService implements StorageAdapter {
 
   @withContext('remove')
   async remove (ctx: MeasureContext, workspaceId: WorkspaceId, objectNames: string[]): Promise<void> {
-    const token = generateToken(systemAccountEmail, workspaceId)
     await Promise.all(
       objectNames.map(async (objectName) => {
-        await this.client.deleteObject(ctx, token, workspaceId, objectName)
+        await this.client.deleteObject(ctx, workspaceId, objectName)
       })
     )
   }
@@ -99,8 +98,6 @@ export class DatalakeService implements StorageAdapter {
 
   @withContext('listStream')
   async listStream (ctx: MeasureContext, workspaceId: WorkspaceId): Promise<BlobStorageIterator> {
-    const token = generateToken(systemAccountEmail, workspaceId)
-
     let hasMore = true
     const buffer: ListBlobResult[] = []
     let cursor: string | undefined
@@ -109,7 +106,7 @@ export class DatalakeService implements StorageAdapter {
       next: async () => {
         try {
           while (hasMore && buffer.length < 50) {
-            const res = await this.client.listObjects(ctx, token, workspaceId, cursor)
+            const res = await this.client.listObjects(ctx, workspaceId, cursor)
             hasMore = res.cursor !== undefined
             cursor = res.cursor
 
@@ -138,8 +135,7 @@ export class DatalakeService implements StorageAdapter {
   @withContext('stat')
   async stat (ctx: MeasureContext, workspaceId: WorkspaceId, objectName: string): Promise<Blob | undefined> {
     try {
-      const token = generateToken(systemAccountEmail, workspaceId)
-      const result = await this.client.statObject(ctx, token, workspaceId, objectName)
+      const result = await this.client.statObject(ctx, workspaceId, objectName)
       if (result !== undefined) {
         return {
           provider: '',
@@ -161,8 +157,7 @@ export class DatalakeService implements StorageAdapter {
 
   @withContext('get')
   async get (ctx: MeasureContext, workspaceId: WorkspaceId, objectName: string): Promise<Readable> {
-    const token = generateToken(systemAccountEmail, workspaceId)
-    return await this.client.getObject(ctx, token, workspaceId, objectName)
+    return await this.client.getObject(ctx, workspaceId, objectName)
   }
 
   @withContext('put')
@@ -174,8 +169,6 @@ export class DatalakeService implements StorageAdapter {
     contentType: string,
     size?: number
   ): Promise<UploadedObjectInfo> {
-    const token = generateToken(systemAccountEmail, workspaceId)
-
     const params: UploadObjectParams = {
       lastModified: Date.now(),
       type: contentType,
@@ -183,7 +176,7 @@ export class DatalakeService implements StorageAdapter {
     }
 
     const { etag } = await ctx.with('put', {}, (ctx) =>
-      withRetry(ctx, 5, () => this.client.putObject(ctx, token, workspaceId, objectName, stream, params))
+      withRetry(ctx, 5, () => this.client.putObject(ctx, workspaceId, objectName, stream, params))
     )
 
     return {
@@ -194,8 +187,7 @@ export class DatalakeService implements StorageAdapter {
 
   @withContext('read')
   async read (ctx: MeasureContext, workspaceId: WorkspaceId, objectName: string): Promise<Buffer[]> {
-    const token = generateToken(systemAccountEmail, workspaceId)
-    const data = await this.client.getObject(ctx, token, workspaceId, objectName)
+    const data = await this.client.getObject(ctx, workspaceId, objectName)
     const chunks: Buffer[] = []
 
     for await (const chunk of data) {
@@ -213,8 +205,7 @@ export class DatalakeService implements StorageAdapter {
     offset: number,
     length?: number
   ): Promise<Readable> {
-    const token = generateToken(systemAccountEmail, workspaceId)
-    return await this.client.getPartialObject(ctx, token, workspaceId, objectName, offset, length)
+    return await this.client.getPartialObject(ctx, workspaceId, objectName, offset, length)
   }
 
   async getUrl (ctx: MeasureContext, workspaceId: WorkspaceId, objectName: string): Promise<string> {

services/love/src/storage.ts

@@ -107,8 +107,8 @@ async function getS3UploadParamsDatalake (
   s3config: S3Config
 ): Promise<S3UploadParams> {
   const token = generateToken(systemAccountEmail, workspaceId)
-  const client = createDatalakeClient(config)
-  const { bucket } = await client.getR2UploadParams(ctx, token, workspaceId)
+  const client = createDatalakeClient(config, token)
+  const { bucket } = await client.getR2UploadParams(ctx, workspaceId)
 
   const endpoint = s3config.endpoint
   const accessKey = s3config.accessKey
@@ -147,13 +147,13 @@ async function saveFileToDatalake (
   filename: string
 ): Promise<Blob | undefined> {
   const token = generateToken(systemAccountEmail, workspaceId)
-  const client = createDatalakeClient(config)
+  const client = createDatalakeClient(config, token)
   const storageAdapter = new DatalakeService(config)
 
   const prefix = rootPrefix(s3config, workspaceId)
   const uuid = stripPrefix(prefix, filename)
 
-  await client.uploadFromR2(ctx, token, workspaceId, uuid, { filename: uuid })
+  await client.uploadFromR2(ctx, workspaceId, uuid, { filename: uuid })
 
   return await storageAdapter.stat(ctx, workspaceId, uuid)
 }