From d2654cb5927b6eb6052401f55174153a793d0e26 Mon Sep 17 00:00:00 2001 From: Andrew Gaul Date: Sat, 25 Jul 2026 15:31:48 -0700 Subject: [PATCH] Implement Put Block From URL Stage the block by fetching the copy source over loopback so that SAS authentication, x-ms-source-range, and the x-ms-source-if-* conditions are enforced by the existing download path, then persist it through the same extent flow as Put Block. Only sources on the same Azurite instance are supported, matching copyFromURL. The response carries the MD5 of the staged content and source condition failures return 412 SourceConditionNotMet as on the real service. Validated with the blockblob test suite and end to end with S3Proxy's native multipart part copy, which previously fell back to streamed emulation on Azurite's 501. Co-Authored-By: Claude Fable 5 --- src/blob/errors/StorageErrorFactory.ts | 9 + src/blob/handlers/BlockBlobHandler.ts | 196 +++++++++++++++++- tests/blob/apis/blockblob.test.ts | 266 +++++++++++++++++++++++++ 3 files changed, 470 insertions(+), 1 deletion(-) diff --git a/src/blob/errors/StorageErrorFactory.ts b/src/blob/errors/StorageErrorFactory.ts index c44776fb5..357cdb9b1 100644 --- a/src/blob/errors/StorageErrorFactory.ts +++ b/src/blob/errors/StorageErrorFactory.ts @@ -754,6 +754,15 @@ export default class StorageErrorFactory { ); } + public static getSourceConditionNotMet(contextID: string): StorageError { + return new StorageError( + 412, + "SourceConditionNotMet", + "The source condition specified using HTTP conditional header(s) is not met.", + contextID + ); + } + public static getInvalidResourceName(contextID: string = ""): StorageError { return new StorageError( 400, diff --git a/src/blob/handlers/BlockBlobHandler.ts b/src/blob/handlers/BlockBlobHandler.ts index ec9b11a12..4f9ce45d4 100644 --- a/src/blob/handlers/BlockBlobHandler.ts +++ b/src/blob/handlers/BlockBlobHandler.ts @@ -1,3 +1,6 @@ +import axios, { AxiosResponse } from "axios"; +import { IncomingMessage } from "http"; + import { convertRawHeadersToMetadata } from "../../common/utils/utils"; import { getMD5FromStream, @@ -272,7 +275,198 @@ export default class BlockBlobHandler options: Models.BlockBlobStageBlockFromURLOptionalParams, context: Context ): Promise { - throw new NotImplementedError(context.contextId); + const blobCtx = new BlobStorageContext(context); + const accountName = blobCtx.account!; + const containerName = blobCtx.container!; + const blobName = blobCtx.blob!; + const date = blobCtx.startTime!; + + // Put Block From URL carries no request body. + if (contentLength !== 0) { + throw StorageErrorFactory.getInvalidHeaderValue(context.contextId, { + HeaderName: "Content-Length", + HeaderValue: contentLength.toString() + }); + } + + this.validateBlockId(blockId, blobCtx); + + await this.metadataStore.checkContainerExist( + context, + accountName, + containerName + ); + + let url: URL; + try { + url = new URL(sourceUrl); + } catch { + throw StorageErrorFactory.getInvalidHeaderValue(context.contextId, { + HeaderName: "x-ms-copy-source", + HeaderValue: sourceUrl + }); + } + + // Only sources within the same Azurite instance are supported, as with + // copyFromURL. + // Hostnames compare case-insensitively and new URL() lowercases its + // host, so normalize the client-supplied header before comparing. + const currentServer = (blobCtx.request!.getHeader("Host") || "") + .toLowerCase(); + if (currentServer !== url.host) { + this.logger.error( + `BlockBlobHandler:stageBlockFromURL() Source ${url} is not on the same Azurite instance as target account ${accountName}`, + context.contextId + ); + throw StorageErrorFactory.getCannotVerifyCopySource( + context.contextId!, + 404, + "The specified resource does not exist" + ); + } + + // The Host header above is client-controlled, so never fetch the + // caller-supplied URL directly; pin the outbound request to the + // loopback address and port this server is actually bound to, keeping + // only the caller's path and query. + const rawRequest = blobCtx.request!.getBodyStream(); + if (!(rawRequest instanceof IncomingMessage) || + rawRequest.socket.localPort === undefined) { + throw StorageErrorFactory.getCannotVerifyCopySource( + context.contextId!, + 404, + "The specified resource does not exist" + ); + } + const scheme = "encrypted" in rawRequest.socket ? "https" : "http"; + // Use the local address this request arrived on rather than a + // hard-coded loopback so non-loopback --blobHost binds keep working; + // IPv6 literals need brackets in URLs. + const localAddress = rawRequest.socket.localAddress || "127.0.0.1"; + const localHost = localAddress.includes(":") ? + `[${localAddress}]` : localAddress; + const pinnedUrl = + `${scheme}://${localHost}:${rawRequest.socket.localPort}` + + `${url.pathname}${url.search}`; + + // Fetch the source range over loopback so that SAS authentication, + // range handling, and source conditions reuse the download path. + // Preserve the source URL's host so product-style source URLs still + // resolve their account from the Host header; the connection itself + // stays pinned to this server's bound address. + const headers: { [key: string]: string } = { host: url.host }; + if (options.sourceRange !== undefined) { + // The download path ignores malformed Range headers, which would + // silently stage the entire source blob; reject them up front. + const rangeMatch = /^bytes=(\d+)-(\d*)$/.exec(options.sourceRange); + if (rangeMatch === null || + (rangeMatch[2] !== "" && + Number.parseInt(rangeMatch[2], 10) < + Number.parseInt(rangeMatch[1], 10))) { + throw StorageErrorFactory.getInvalidHeaderValue(context.contextId, { + HeaderName: "x-ms-source-range", + HeaderValue: options.sourceRange + }); + } + headers.range = options.sourceRange; + } + const sourceConditions = options.sourceModifiedAccessConditions || {}; + if (sourceConditions.sourceIfMatch !== undefined) { + headers["if-match"] = sourceConditions.sourceIfMatch; + } + if (sourceConditions.sourceIfNoneMatch !== undefined) { + headers["if-none-match"] = sourceConditions.sourceIfNoneMatch; + } + if (sourceConditions.sourceIfModifiedSince !== undefined) { + headers["if-modified-since"] = new Date( + sourceConditions.sourceIfModifiedSince + ).toUTCString(); + } + if (sourceConditions.sourceIfUnmodifiedSince !== undefined) { + headers["if-unmodified-since"] = new Date( + sourceConditions.sourceIfUnmodifiedSince + ).toUTCString(); + } + // Note: unlike the Copy Blob operations, Put Block From URL has no + // x-ms-source-if-tags condition; the generated operation spec does not + // deserialize one. + + const sourceResponse: AxiosResponse = await axios.get(pinnedUrl, { + headers, + responseType: "stream", + validateStatus: () => true + }); + + if (sourceResponse.status === 304 || sourceResponse.status === 412) { + sourceResponse.data.destroy(); + throw StorageErrorFactory.getSourceConditionNotMet(context.contextId!); + } + if (sourceResponse.status === 404) { + sourceResponse.data.destroy(); + throw StorageErrorFactory.getCannotVerifyCopySource( + context.contextId!, + 404, + "The specified resource does not exist" + ); + } + if (sourceResponse.status !== 200 && sourceResponse.status !== 206) { + sourceResponse.data.destroy(); + throw StorageErrorFactory.getCannotVerifyCopySource( + context.contextId!, + sourceResponse.status, + "Could not verify the copy source within the specified time." + ); + } + + const persistency = await this.extentStore.appendExtent( + sourceResponse.data, + context.contextId + ); + + // Put Block From URL echoes the MD5 of the staged content. + const stream = await this.extentStore.readExtent( + persistency, + context.contextId + ); + const calculatedContentMD5 = await getMD5FromStream(stream); + if (options.sourceContentMD5 !== undefined) { + if ( + !Buffer.from(options.sourceContentMD5).equals(calculatedContentMD5) + ) { + throw StorageErrorFactory.getInvalidOperation( + context.contextId!, + "Provided contentMD5 doesn't match." + ); + } + } + + const block: BlockModel = { + accountName, + containerName, + blobName, + isCommitted: false, + name: blockId, + size: persistency.count, + persistency + }; + + await this.metadataStore.stageBlock( + context, + block, + options.leaseAccessConditions + ); + + const response: Models.BlockBlobStageBlockFromURLResponse = { + statusCode: 201, + contentMD5: calculatedContentMD5, + requestId: blobCtx.contextId, + version: BLOB_API_VERSION, + date, + isServerEncrypted: true, + clientRequestId: options.requestId + }; + + return response; } public async commitBlockList( diff --git a/tests/blob/apis/blockblob.test.ts b/tests/blob/apis/blockblob.test.ts index fa31f2ace..b52338c6e 100644 --- a/tests/blob/apis/blockblob.test.ts +++ b/tests/blob/apis/blockblob.test.ts @@ -5,6 +5,7 @@ import { BlobSASPermissions, Tags } from "@azure/storage-blob"; +import axios from "axios"; import * as assert from "assert"; import * as crypto from "crypto"; @@ -242,6 +243,271 @@ describe("BlockBlobAPIs", () => { ); }); + it("stageBlockFromURL @loki @sql", async () => { + const content = "HelloWorldFromSourceBlob"; + const sourceClient = containerClient.getBlockBlobClient( + getUniqueName("source") + ); + await sourceClient.upload(content, content.length); + const sourceUrl = await sourceClient.generateSasUrl({ + permissions: BlobSASPermissions.parse("r"), + expiresOn: new Date(Date.now() + 60 * 60 * 1000) + }); + + const resultStage = await blockBlobClient.stageBlockFromURL( + base64encode("1"), + sourceUrl, + 0, + 10 + ); + const expectedMD5 = await getMD5FromString(content.substring(0, 10)); + assert.deepStrictEqual( + Buffer.from(resultStage.contentMD5!), + Buffer.from(expectedMD5) + ); + + await blockBlobClient.stageBlockFromURL( + base64encode("2"), + sourceUrl, + 10, + content.length - 10 + ); + + const listResponse = await blockBlobClient.getBlockList("uncommitted"); + assert.equal(listResponse.uncommittedBlocks!.length, 2); + assert.equal(listResponse.uncommittedBlocks![0].size, 10); + assert.equal( + listResponse.uncommittedBlocks![1].size, + content.length - 10 + ); + + await blockBlobClient.commitBlockList([ + base64encode("1"), + base64encode("2") + ]); + const result = await blobClient.download(0); + assert.equal(await bodyToString(result, content.length), content); + }); + + it("stageBlockFromURL without range copies the entire source @loki @sql", async () => { + const content = "HelloWorldFromSourceBlob"; + const sourceClient = containerClient.getBlockBlobClient( + getUniqueName("source") + ); + await sourceClient.upload(content, content.length); + const sourceUrl = await sourceClient.generateSasUrl({ + permissions: BlobSASPermissions.parse("r"), + expiresOn: new Date(Date.now() + 60 * 60 * 1000) + }); + + await blockBlobClient.stageBlockFromURL(base64encode("1"), sourceUrl); + await blockBlobClient.commitBlockList([base64encode("1")]); + const result = await blobClient.download(0); + assert.equal(await bodyToString(result, content.length), content); + }); + + it("stageBlockFromURL rejects an unmet source condition @loki @sql", async () => { + const content = "HelloWorldFromSourceBlob"; + const sourceClient = containerClient.getBlockBlobClient( + getUniqueName("source") + ); + await sourceClient.upload(content, content.length); + const sourceUrl = await sourceClient.generateSasUrl({ + permissions: BlobSASPermissions.parse("r"), + expiresOn: new Date(Date.now() + 60 * 60 * 1000) + }); + + // @azure/storage-blob does not expose x-ms-source-if-* on + // stageBlockFromURL, so issue the request directly. + const destinationUrl = await blockBlobClient.generateSasUrl({ + permissions: BlobSASPermissions.parse("rw"), + expiresOn: new Date(Date.now() + 60 * 60 * 1000) + }); + const response = await axios.put( + destinationUrl + + "&comp=block&blockid=" + + encodeURIComponent(base64encode("1")), + undefined, + { + headers: { + "x-ms-copy-source": sourceUrl, + "x-ms-source-if-match": '"0x0000000000000000"', + "Content-Length": "0" + }, + validateStatus: () => true + } + ); + assert.deepStrictEqual(response.status, 412); + assert.ok(response.data.includes("SourceConditionNotMet")); + }); + + it("stageBlockFromURL rejects a request body @loki @sql", async () => { + const content = "HelloWorldFromSourceBlob"; + const sourceClient = containerClient.getBlockBlobClient( + getUniqueName("source") + ); + await sourceClient.upload(content, content.length); + const sourceUrl = await sourceClient.generateSasUrl({ + permissions: BlobSASPermissions.parse("r"), + expiresOn: new Date(Date.now() + 60 * 60 * 1000) + }); + const destinationUrl = await blockBlobClient.generateSasUrl({ + permissions: BlobSASPermissions.parse("rw"), + expiresOn: new Date(Date.now() + 60 * 60 * 1000) + }); + + const response = await axios.put( + destinationUrl + + "&comp=block&blockid=" + + encodeURIComponent(base64encode("1")), + "unexpected body", + { + headers: { + "x-ms-copy-source": sourceUrl + }, + validateStatus: () => true + } + ); + assert.deepStrictEqual(response.status, 400); + assert.ok(response.data.includes("InvalidHeaderValue")); + }); + + it("stageBlockFromURL rejects a malformed source range @loki @sql", async () => { + const content = "HelloWorldFromSourceBlob"; + const sourceClient = containerClient.getBlockBlobClient( + getUniqueName("source") + ); + await sourceClient.upload(content, content.length); + const sourceUrl = await sourceClient.generateSasUrl({ + permissions: BlobSASPermissions.parse("r"), + expiresOn: new Date(Date.now() + 60 * 60 * 1000) + }); + const destinationUrl = await blockBlobClient.generateSasUrl({ + permissions: BlobSASPermissions.parse("rw"), + expiresOn: new Date(Date.now() + 60 * 60 * 1000) + }); + + for (const badRange of ["bytes=abc", "bytes=5-2", "0-10"]) { + const response = await axios.put( + destinationUrl + + "&comp=block&blockid=" + + encodeURIComponent(base64encode("1")), + undefined, + { + headers: { + "x-ms-copy-source": sourceUrl, + "x-ms-source-range": badRange, + "Content-Length": "0" + }, + validateStatus: () => true + } + ); + assert.deepStrictEqual(response.status, 400, badRange); + assert.ok(response.data.includes("InvalidHeaderValue"), badRange); + } + }); + + it("stageBlockFromURL with product-style source URL @loki @sql", async () => { + const content = "HelloWorldFromSourceBlob"; + const sourceName = getUniqueName("source"); + const sourceClient = containerClient.getBlockBlobClient(sourceName); + await sourceClient.upload(content, content.length); + const sourceSasQuery = (await sourceClient.generateSasUrl({ + permissions: BlobSASPermissions.parse("r"), + expiresOn: new Date(Date.now() + 60 * 60 * 1000) + })).split("?")[1]; + const destinationSasQuery = (await blockBlobClient.generateSasUrl({ + permissions: BlobSASPermissions.parse("rw"), + expiresOn: new Date(Date.now() + 60 * 60 * 1000) + })).split("?")[1]; + + // Both requests use product-style URLs, where the account comes from + // the Host header rather than the path. + const productHost = + `${EMULATOR_ACCOUNT_NAME}.localhost:${server.config.port}`; + const response = await axios.put( + `http://${server.config.host}:${server.config.port}` + + `/${containerName}/${blobName}?${destinationSasQuery}` + + "&comp=block&blockid=" + + encodeURIComponent(base64encode("1")), + undefined, + { + headers: { + host: productHost, + "x-ms-copy-source": + `http://${productHost}/${containerName}/${sourceName}` + + `?${sourceSasQuery}`, + "Content-Length": "0" + }, + validateStatus: () => true + } + ); + assert.deepStrictEqual(response.status, 201); + + const listResponse = await blockBlobClient.getBlockList("uncommitted"); + assert.equal(listResponse.uncommittedBlocks!.length, 1); + assert.equal(listResponse.uncommittedBlocks![0].size, content.length); + }); + + it("stageBlockFromURL accepts a mixed-case Host header @loki @sql", async () => { + const content = "HelloWorldFromSourceBlob"; + const sourceName = getUniqueName("source"); + const sourceClient = containerClient.getBlockBlobClient(sourceName); + await sourceClient.upload(content, content.length); + const sourceSasQuery = (await sourceClient.generateSasUrl({ + permissions: BlobSASPermissions.parse("r"), + expiresOn: new Date(Date.now() + 60 * 60 * 1000) + })).split("?")[1]; + const destinationSasQuery = (await blockBlobClient.generateSasUrl({ + permissions: BlobSASPermissions.parse("rw"), + expiresOn: new Date(Date.now() + 60 * 60 * 1000) + })).split("?")[1]; + + // Host headers are case-insensitive; the lowercase source URL host + // must match despite the client's casing. + const response = await axios.put( + `http://${server.config.host}:${server.config.port}` + + `/${EMULATOR_ACCOUNT_NAME}/${containerName}/${blobName}` + + `?${destinationSasQuery}` + + "&comp=block&blockid=" + + encodeURIComponent(base64encode("1")), + undefined, + { + headers: { + host: `LocalHost:${server.config.port}`, + "x-ms-copy-source": + `http://localhost:${server.config.port}` + + `/${EMULATOR_ACCOUNT_NAME}/${containerName}/${sourceName}` + + `?${sourceSasQuery}`, + "Content-Length": "0" + }, + validateStatus: () => true + } + ); + assert.deepStrictEqual(response.status, 201); + }); + + it("stageBlockFromURL from a missing source returns 404 @loki @sql", async () => { + const missingClient = containerClient.getBlockBlobClient( + getUniqueName("missing") + ); + const sourceUrl = await missingClient.generateSasUrl({ + permissions: BlobSASPermissions.parse("r"), + expiresOn: new Date(Date.now() + 60 * 60 * 1000) + }); + + try { + await blockBlobClient.stageBlockFromURL( + base64encode("1"), + sourceUrl + ); + assert.fail(); + } catch (err: any) { + assert.deepStrictEqual(err.statusCode, 404); + assert.deepStrictEqual(err.details.errorCode, "CannotVerifyCopySource"); + } + }); + it("stageBlock with double commit block should work @loki @sql", async () => { const body = "HelloWorld";