fix: validate cloud range responses Retain provider responses and reject mismatched ranges, object versions, content encodings, and byte counts before Arrow interprets the bytes. Keep the released CloudStore request paths and disable HTTP decompression so returned bytes retain their stored offsets. Cover equal-length wrong-column responses and protocol faults through both provider clients, and exercise the signed S3 and Azure Table and Stream paths. Generated-by: OpenAI Codex
diff --git a/docs/src/manual.md b/docs/src/manual.md index b6c8b45..7709381 100644 --- a/docs/src/manual.md +++ b/docs/src/manual.md
@@ -346,6 +346,14 @@ tbl = Arrow.Table(obj; scan = Scan(select = (:id,), filter = col(:day) > 20)) ``` +Cloud reads pin a strong ETag before fetching bytes. If the object's ETag is +missing, Arrow refreshes its metadata once and requires the size to agree. +Each nonempty request, including a whole-object range, must return HTTP 206 +with the requested byte range, total object size, ETag, and byte count. +Overwritten objects and inconsistent responses fail the read. HTTP content +encodings such as gzip are rejected to preserve byte offsets; Arrow's IPC +compression is still supported. Empty reads send no request. + Any other transport is two methods away: ```julia
diff --git a/ext/ArrowCloudStoreExt.jl b/ext/ArrowCloudStoreExt.jl index 9c4e300..aea7e65 100644 --- a/ext/ArrowCloudStoreExt.jl +++ b/ext/ArrowCloudStoreExt.jl
@@ -22,6 +22,8 @@ using Arrow using CloudStore: CloudStore, Object +const HTTP = CloudStore.API.HTTP + """ CloudObjectSource(obj::CloudStore.Object) <: Arrow.AbstractArrowSource @@ -34,6 +36,11 @@ If the ETag is missing, construction refreshes metadata once. A changed size or missing, weak, or malformed ETag throws before any ranges are read. + +Every nonempty read, including a whole-object read, requires a 206 response +whose range, total size, ETag, and body length match the request. HTTP content +encodings such as gzip are rejected; Arrow's IPC compression remains supported. +An empty read returns an empty vector without a request. """ struct CloudObjectSource{O<:Object} <: Arrow.AbstractArrowSource obj::O @@ -70,17 +77,32 @@ len == 0 && return UInt8[] last = off + len - 1 obj = s.obj - headers = ["Range" => "bytes=$(off)-$(last)"] + headers = ["Range" => "bytes=$(off)-$(last)", "Accept-Encoding" => "identity"] etag = _ifmatch(obj.eTag) etag === nothing || push!(headers, "If-Match" => etag) - bytes = CloudStore.get( + response = CloudStore.API.getObject( obj.store, - obj.key; + CloudStore.API.makeURL(obj.store, obj.key), + headers; credentials=obj.credentials, - headers=headers, - allowMultipart=false, - objectMaxSize=Int(len), + decompress=false, ) + response.status == 206 || + throw(Arrow.ValidationError("cloud range request did not return a 206 response")) + range = + match(r"^bytes (\d+)-(\d+)/(\d+)\z"i, HTTP.header(response, "Content-Range", "")) + range !== nothing && + (tryparse(Int64, range[1]), tryparse(Int64, range[2]), tryparse(Int64, range[3])) == + (off, last, obj.size) || throw( + Arrow.ValidationError("cloud response does not match the requested byte range"), + ) + HTTP.header(response, "ETag", "") == etag || + throw(Arrow.ValidationError("cloud response does not match the source ETag")) + lowercase(strip(HTTP.header(response, "Content-Encoding", ""))) in ("", "identity") || + throw(Arrow.ValidationError("cloud range response has an HTTP content encoding")) + bytes = response.body + length(bytes) == len || + throw(Arrow.ValidationError("cloud range response has an unexpected body length")) return bytes isa Vector{UInt8} ? bytes : Vector{UInt8}(bytes) end
diff --git a/test/Project.toml b/test/Project.toml index 964f189..53e0859 100644 --- a/test/Project.toml +++ b/test/Project.toml
@@ -16,6 +16,7 @@ [deps] Serialization = "9e88b42a-f829-5b0c-bbe9-9e923198166b" +Sockets = "6462fe0b-24de-5631-8697-dd941f90decc" DataDecimals = "3e2245cb-6932-498d-a7cf-39d39de8bde3" Durations = "41ac09f5-0a95-4810-aa18-c88bf48ee130" Aqua = "4c88cf16-eb10-579e-8560-4a9242c79595"
diff --git a/test/cloudstore_ranges.jl b/test/cloudstore_ranges.jl new file mode 100644 index 0000000..7b66ac8 --- /dev/null +++ b/test/cloudstore_ranges.jl
@@ -0,0 +1,203 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +using Sockets +using CodecZlib + +# Keep the deliberately inconsistent headers on the wire unchanged by a server +# library. Both providers still use their real HTTP clients and signing paths. +function with_range_response(f, mode, data) + listener = Sockets.listen(Sockets.IPv4("127.0.0.1"), 0) + host = "http://127.0.0.1:$(Sockets.getsockname(listener)[2])" + requests = Any[] + request_lock = ReentrantLock() + server = @async begin + @sync while isopen(listener) + socket = try + Sockets.accept(listener) + catch + isopen(listener) && rethrow() + break + end + @async try + line = readline(socket) + isempty(line) && return nothing + method, target, _ = split(line; limit=3) + headers = Dict{String,String}() + while true + line = readline(socket) + isempty(line) && break + name, value = split(line, ':'; limit=2) + headers[lowercase(name)] = strip(value) + end + lock(request_lock) do + push!(requests, (; method, target, headers)) + end + range = match(r"^bytes=(\d+)-(\d+)\z", headers["range"]) + lo, hi = parse.(Int, range.captures) + width = hi - lo + 1 + shift = + mode == :wrong_offset ? 1 : + mode == :swapped_column && + width >= 100000 && + hi + width < length(data) ? width : 0 + body = data[(lo + shift + 1):(hi + shift + 1)] + total = mode == :wrong_total ? length(data) + 1 : length(data) + range_end = hi + shift + (mode == :wrong_end) + content_range = "bytes $(lo + shift)-$range_end/$total" + mode == :malformed_range && (content_range *= " garbage") + mode == :unknown_total && (content_range = "bytes $lo-$hi/*") + mode == :overflow && (content_range = "bytes $lo-$hi/$(big(2)^100)") + status = 206 + encoding = "" + tag = + mode == :changed_tag ? "\"v2\"" : + mode == :weak_tag ? "W/\"v1\"" : "\"v1\"" + if mode == :ignored + status, body = 200, data + elseif mode == :short + body = body[1:(end - 1)] + elseif mode == :long + body = vcat(body, 0xff) + elseif mode == :encoded + body = transcode(GzipCompressor, body) + encoding = "gzip" + elseif mode == :identity + encoding = "Identity" + content_range = replace(content_range, "bytes" => "Bytes") + end + response_headers = ["Connection" => "close"] + mode == :missing_range || + push!(response_headers, "Content-Range" => content_range) + mode == :missing_tag || push!(response_headers, "ETag" => tag) + isempty(encoding) || + push!(response_headers, "Content-Encoding" => encoding) + push!( + response_headers, + mode == :chunked ? "Transfer-Encoding" => "chunked" : + "Content-Length" => string(length(body)), + ) + write(socket, "HTTP/1.1 $status fixture\r\n") + for (name, value) in response_headers + write(socket, name, ": ", value, "\r\n") + end + write(socket, "\r\n") + if mode == :chunked + write(socket, string(length(body); base=16), "\r\n", body, "\r\n0\r\n\r\n") + else + write(socket, body) + end + finally + close(socket) + end + end + end + stores = ( + ( + CloudStore.S3.Bucket("ranges", "us-east-1"; host), + CloudStore.AWS.Credentials("fixture", "fixture"), + ), + ( + CloudStore.Blobs.Container("ranges", "fixture"; host), + CloudStore.Azure.Credentials("fixture", "Zml4dHVyZQ=="), + ), + ) + try + return f(stores, requests) + finally + close(listener) + wait(server) + end +end + +@testset "Cloud range response integrity" begin + ext = Base.get_extension(Arrow, :ArrowCloudStoreExt) + data = collect(codeunits("abcdefghijkl")) + for mode in (:valid, :identity, :chunked) + with_range_response(mode, data) do stores, requests + for (store, credentials) in stores + source = ext.CloudObjectSource( + CloudStore.Object(store, credentials, "data", length(data), "v1"), + ) + @test Arrow.readrange(source, 4, 4) == data[5:8] + @test Arrow.readrange(source, 0, length(data)) == data + before = length(requests) + @test Arrow.readrange(source, 0, 0) == UInt8[] + @test Arrow.readrange(source, length(data), 0) == UInt8[] + @test length(requests) == before + end + @test all(r -> r.headers["if-match"] == "\"v1\"", requests) + @test all(r -> get(r.headers, "accept-encoding", "") == "identity", requests) + end + end + for mode in ( + :wrong_offset, + :wrong_end, + :wrong_total, + :missing_range, + :malformed_range, + :unknown_total, + :overflow, + :ignored, + :missing_tag, + :changed_tag, + :weak_tag, + :short, + :long, + :encoded, + ) + with_range_response(mode, data) do stores, requests + for (store, credentials) in stores + source = ext.CloudObjectSource( + CloudStore.Object(store, credentials, "data", length(data), "v1"), + ) + @test_throws Arrow.ValidationError Arrow.readrange(source, 4, 4) + # A 200 response is also rejected when a Range requests the whole object. + if mode == :ignored + @test_throws Arrow.ValidationError Arrow.readrange( + source, + 0, + length(data), + ) + end + end + end + end +end + +@testset "Cloud scans reject bytes from another column" begin + expected = collect(Int64, 1:30000) + other = fill(Int64(-7), length(expected)) + io = IOBuffer() + Arrow.write(io, (a=expected, b=other); file=true) + data = take!(io) + for mode in (:valid, :swapped_column, :encoded) + with_range_response(mode, data) do stores, requests + for (store, credentials) in stores + object = CloudStore.Object(store, credentials, "data", length(data), "v1") + if mode == :valid + @test collect(Arrow.Table(object; scan=Tables.Scan(select=(:a,))).a) == + expected + else + @test_throws Arrow.ValidationError Arrow.Table( + object; + scan=Tables.Scan(select=(:a,)), + ) + end + end + end + end +end
diff --git a/test/cloudstore_tests.jl b/test/cloudstore_tests.jl index cf4a061..1006459 100644 --- a/test/cloudstore_tests.jl +++ b/test/cloudstore_tests.jl
@@ -141,45 +141,52 @@ sio = IOBuffer() Arrow.write(sio, Tables.partitioner([part1, part2]); file=false) streambytes = take!(sio) - Minio.with() do conf - credentials, bucket = conf.credentials, conf.store - CloudStore.put(bucket, "t.arrow", filebytes; credentials=credentials) - CloudStore.put(bucket, "t.arrows", streambytes; credentials=credentials) - obj = CloudStore.Object(bucket, "t.arrow"; credentials=credentials) - @test Arrow.sourcelength(ext.CloudObjectSource(obj)) == length(filebytes) - # one range, byte-exact against the object; a bounded concurrency - src = ext.CloudObjectSource(obj) - @test Arrow.readrange(src, 8, 16) == filebytes[9:24] - @test Arrow.readrange(src, length(filebytes) - 6, 6) == filebytes[(end - 5):end] - @test Arrow.readrange(src, 0, 0) == UInt8[] - @test Arrow.concurrentreads(src) == ext.CONCURRENT_RANGE_READS > 1 - # a projected, windowed scan over the object - t = Arrow.Table(obj; scan=Tables.Scan(select=(:b,), limit=3, offset=1500)) - @test Tables.columnnames(t) == [:b] - @test t.b == ["v1501", "v1502", "v1503"] - # a filter that the footer statistics can prune to the second batch - t2 = Arrow.Table(obj; scan=Tables.Scan(select=(:a,), filter=Tables.col(:a) > 1990)) - @test t2.a == collect(1991:2000) - # whole-object reads agree with the in-memory reads - full = Arrow.Table(obj) - mem = Arrow.Table(filebytes) - @test Tables.columnnames(full) == Tables.columnnames(mem) - @test full.a == mem.a && full.b == mem.b && full.c == mem.c - sobj = CloudStore.Object(bucket, "t.arrows"; credentials=credentials) - st = Arrow.Table(sobj) - @test st.a == mem.a && st.b == mem.b - @test [length(batch.a) for batch in Arrow.Stream(sobj)] == [1000, 1000] - @test [length(batch.a) for batch in Arrow.Stream(obj)] == [1000, 1000] - # A handle is pinned to the object version it was made from: after the - # key is overwritten, its next range read fails (If-Match) rather than - # mixing bytes of two versions across request rounds. - stale = ext.CloudObjectSource(obj) - CloudStore.put(bucket, "t.arrow", streambytes; credentials=credentials) - @test_throws Exception Arrow.readrange(stale, 8, 16) - @test_throws Exception Arrow.Table(obj; scan=Tables.Scan(select=(:a,))) - fresh = CloudStore.Object(bucket, "t.arrow"; credentials=credentials) - @test Arrow.Table(fresh).a == mem.a + for emulator in (Minio, Azurite) + emulator.with() do conf + credentials, bucket = conf.credentials, conf.store + CloudStore.put(bucket, "t.arrow", filebytes; credentials=credentials) + CloudStore.put(bucket, "t.arrows", streambytes; credentials=credentials) + obj = CloudStore.Object(bucket, "t.arrow"; credentials=credentials) + @test Arrow.sourcelength(ext.CloudObjectSource(obj)) == length(filebytes) + # one range, byte-exact against the object; a bounded concurrency + src = ext.CloudObjectSource(obj) + @test Arrow.readrange(src, 8, 16) == filebytes[9:24] + @test Arrow.readrange(src, length(filebytes) - 6, 6) == filebytes[(end - 5):end] + @test Arrow.readrange(src, 0, 0) == UInt8[] + @test Arrow.concurrentreads(src) == ext.CONCURRENT_RANGE_READS > 1 + # a projected, windowed scan over the object + t = Arrow.Table(obj; scan=Tables.Scan(select=(:b,), limit=3, offset=1500)) + @test Tables.columnnames(t) == [:b] + @test t.b == ["v1501", "v1502", "v1503"] + # a filter that the footer statistics can prune to the second batch + t2 = Arrow.Table( + obj; + scan=Tables.Scan(select=(:a,), filter=Tables.col(:a) > 1990), + ) + @test t2.a == collect(1991:2000) + # whole-object reads agree with the in-memory reads + full = Arrow.Table(obj) + mem = Arrow.Table(filebytes) + @test Tables.columnnames(full) == Tables.columnnames(mem) + @test full.a == mem.a && full.b == mem.b && full.c == mem.c + sobj = CloudStore.Object(bucket, "t.arrows"; credentials=credentials) + st = Arrow.Table(sobj) + @test st.a == mem.a && st.b == mem.b + @test [length(batch.a) for batch in Arrow.Stream(sobj)] == [1000, 1000] + @test [length(batch.a) for batch in Arrow.Stream(obj)] == [1000, 1000] + # A handle is pinned to the object version it was made from: after the + # key is overwritten, its next range read fails (If-Match) rather than + # mixing bytes of two versions across request rounds. + stale = ext.CloudObjectSource(obj) + CloudStore.put(bucket, "t.arrow", streambytes; credentials=credentials) + @test_throws Exception Arrow.readrange(stale, 8, 16) + @test_throws Exception Arrow.Table(obj; scan=Tables.Scan(select=(:a,))) + fresh = CloudStore.Object(bucket, "t.arrow"; credentials=credentials) + @test Arrow.Table(fresh).a == mem.a + end end end +include("cloudstore_ranges.jl") + end # module
diff --git a/test/runtests.jl b/test/runtests.jl index 69548ea..c8fe2d8 100644 --- a/test/runtests.jl +++ b/test/runtests.jl
@@ -52,7 +52,7 @@ # Shared acceptance/conformance support contracts and adapter composition. include("conformance_support_tests.jl") -# The CloudStore.jl extension against a local S3-compatible server. +# The CloudStore.jl extension against local S3 and Azure emulators. include("cloudstore_tests.jl") # The adapter acceptance batteries: assertion-dense scripts over the