From 392eaea09c68752f2ad0f08fdefdd07cfb1ad290 Mon Sep 17 00:00:00 2001 From: Alexander Onnikov Date: Mon, 3 Feb 2025 10:07:53 +0400 Subject: [PATCH] fix: allow access to any workspace with system email (#7865) Signed-off-by: Alexander Onnikov --- dev/tool/src/index.ts | 4 +- dev/tool/src/storage.ts | 15 +++--- server/datalake/src/client.ts | 95 ++++++++++++++--------------------- server/datalake/src/index.ts | 31 ++++-------- services/love/src/storage.ts | 8 +-- 5 files changed, 62 insertions(+), 91 deletions(-) diff --git a/dev/tool/src/index.ts b/dev/tool/src/index.ts index 64bef7b067..64f853a095 100644 --- a/dev/tool/src/index.ts +++ b/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) } } }) diff --git a/dev/tool/src/storage.ts b/dev/tool/src/storage.ts index f1aec4f0f0..9d550abdbc 100644 --- a/dev/tool/src/storage.ts +++ b/dev/tool/src/storage.ts @@ -271,7 +271,6 @@ export async function copyToDatalake ( config: S3Config, adapter: S3Service, datalake: DatalakeClient, - token: string, params: CopyDatalakeParams ): Promise { 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 { 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() diff --git a/server/datalake/src/client.ts b/server/datalake/src/client.ts index 151f9668ec..3955b39897 100644 --- a/server/datalake/src/client.ts +++ b/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 + + 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 { + async getObject (ctx: MeasureContext, workspace: WorkspaceId, objectName: string): Promise { 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 { 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 { @@ -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 { + async deleteObject (ctx: MeasureContext, workspace: WorkspaceId, objectName: string): Promise { 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 { 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 { - 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 { + async getR2UploadParams (ctx: MeasureContext, workspace: WorkspaceId): Promise { 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 { + private async signObjectSign (ctx: MeasureContext, workspace: WorkspaceId, objectName: string): Promise { 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 { 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 { + private async signObjectDelete (ctx: MeasureContext, workspace: WorkspaceId, objectName: string): Promise { 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') diff --git a/server/datalake/src/index.ts b/server/datalake/src/index.ts index 1605de6a92..47b050a3a9 100644 --- a/server/datalake/src/index.ts +++ b/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 {} @@ -84,10 +84,9 @@ export class DatalakeService implements StorageAdapter { @withContext('remove') async remove (ctx: MeasureContext, workspaceId: WorkspaceId, objectNames: string[]): Promise { - 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 { - 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 { 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 { - 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 { - 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 { - 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 { - 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 { diff --git a/services/love/src/storage.ts b/services/love/src/storage.ts index 2a8bb4cd39..d5ed9d147f 100644 --- a/services/love/src/storage.ts +++ b/services/love/src/storage.ts @@ -107,8 +107,8 @@ async function getS3UploadParamsDatalake ( s3config: S3Config ): Promise { 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 { 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) }