Stream attachment uploads and downloads through storage
Uploads are authenticated before the body is read: token, proof headers, device and account are checked first, then Content-Length (required, 411 without it) against attachmentMaxBytes and the rate limits. The body then streams to storage while it is counted and hashed. The proof signature, which covers the body's SHA-256, is checked after the last byte, and a blob whose proof fails, or whose body ends short of Content-Length, is deleted. S3 PUTs are signed UNSIGNED-PAYLOAD and sent with a known length through AsyncHTTPClient. The digest is unknown until the last byte, so x-amz-checksum-sha256 cannot go in a header, and B2 refuses aws-chunked. GETs stream the bucket's body to the client, Range still asked of the bucket. The disk store writes to a .partial file and renames it. Vapor runs a streamed body's closure a second time when writing the response fails; the bucket body is handed out once, because a second read traps the process. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Paul Hitt
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
cf1ce36602
commit
ea1b2f228e
@@ -19,6 +19,10 @@ let package = Package(
|
||||
.package(url: "https://github.com/apple/swift-crypto.git", from: "3.0.0"),
|
||||
// Wake-up fan-out for long-polling.
|
||||
.package(url: "https://github.com/vapor/redis.git", from: "4.11.0"),
|
||||
// Already in the graph through Vapor. Named here because the
|
||||
// attachment store streams through its request and response types,
|
||||
// which Vapor does not re-export.
|
||||
.package(url: "https://github.com/swift-server/async-http-client.git", from: "1.21.0"),
|
||||
],
|
||||
targets: [
|
||||
.executableTarget(
|
||||
@@ -31,6 +35,7 @@ let package = Package(
|
||||
.product(name: "Redis", package: "redis"),
|
||||
.product(name: "Leaf", package: "leaf"),
|
||||
.product(name: "Crypto", package: "swift-crypto"),
|
||||
.product(name: "AsyncHTTPClient", package: "async-http-client"),
|
||||
]
|
||||
),
|
||||
.testTarget(
|
||||
@@ -38,6 +43,7 @@ let package = Package(
|
||||
dependencies: [
|
||||
"FlamenetRelay",
|
||||
.product(name: "VaporTesting", package: "vapor"),
|
||||
.product(name: "AsyncHTTPClient", package: "async-http-client"),
|
||||
// Only PostgresMigrationTests uses it, and only when
|
||||
// FN_TEST_POSTGRES names a reachable server: CI has no
|
||||
// Postgres service, so the dialect production runs on is
|
||||
|
||||
@@ -1,10 +1,16 @@
|
||||
import AsyncHTTPClient
|
||||
import Crypto
|
||||
import Foundation
|
||||
import NIOCore
|
||||
import Vapor
|
||||
|
||||
/// Where sealed attachment blobs live.
|
||||
protocol AttachmentStore: Sendable {
|
||||
func write(_ token: String, data: Data) async throws
|
||||
/// Store exactly `length` bytes from `body`, one chunk at a time. When
|
||||
/// `body` throws or ends at any other length, nothing is left stored.
|
||||
func write<Body: AsyncSequence & Sendable>(
|
||||
_ token: String, length: Int, body: Body
|
||||
) async throws where Body.Element == ByteBuffer
|
||||
/// The stored size, or nil when the backend cannot say cheaply.
|
||||
func storedSize(_ token: String) async -> Int?
|
||||
/// The blob as a Response, or nil when the bytes are gone.
|
||||
@@ -18,6 +24,24 @@ protocol AttachmentStore: Sendable {
|
||||
func reachability() async -> StorageReachability
|
||||
}
|
||||
|
||||
extension AttachmentStore {
|
||||
/// The whole blob at once, for callers that already hold it in memory.
|
||||
func write(_ token: String, data: Data) async throws {
|
||||
let buffer = ByteBuffer(data: data)
|
||||
let body = AsyncStream<ByteBuffer> {
|
||||
$0.yield(buffer)
|
||||
$0.finish()
|
||||
}
|
||||
try await write(token, length: data.count, body: body)
|
||||
}
|
||||
}
|
||||
|
||||
/// A body that ended at a length other than the one declared for it.
|
||||
struct AttachmentLengthMismatch: Error {
|
||||
let declared: Int
|
||||
let received: Int
|
||||
}
|
||||
|
||||
/// What a status probe of the blob backend can conclude.
|
||||
enum StorageReachability: Sendable {
|
||||
case reachable
|
||||
@@ -32,12 +56,45 @@ enum StorageReachability: Sendable {
|
||||
struct DirectoryAttachmentStore: AttachmentStore {
|
||||
let directory: String
|
||||
|
||||
func write(_ token: String, data: Data) async throws {
|
||||
/// Written to a `.partial` file and renamed into place after the last
|
||||
/// byte, so a reader never sees half a blob.
|
||||
func write<Body: AsyncSequence & Sendable>(
|
||||
_ token: String, length: Int, body: Body
|
||||
) async throws where Body.Element == ByteBuffer {
|
||||
try FileManager.default.createDirectory(
|
||||
atPath: directory, withIntermediateDirectories: true)
|
||||
try data.write(
|
||||
to: URL(fileURLWithPath: attachmentPath(directory, token)),
|
||||
options: .atomic)
|
||||
let path = attachmentPath(directory, token)
|
||||
let partial = path + ".partial"
|
||||
guard FileManager.default.createFile(atPath: partial, contents: nil) else {
|
||||
throw Abort(.internalServerError, reason: "cannot create \(partial)")
|
||||
}
|
||||
do {
|
||||
let handle = try FileHandle(forWritingTo: URL(fileURLWithPath: partial))
|
||||
var written = 0
|
||||
do {
|
||||
for try await chunk in body {
|
||||
written += chunk.readableBytes
|
||||
guard written <= length else { break }
|
||||
try handle.write(contentsOf: chunk.readableBytesView)
|
||||
}
|
||||
try handle.close()
|
||||
} catch {
|
||||
try? handle.close()
|
||||
throw error
|
||||
}
|
||||
guard written == length else {
|
||||
throw AttachmentLengthMismatch(declared: length, received: written)
|
||||
}
|
||||
// rename(2) replaces an existing file in one step; `moveItem`
|
||||
// refuses to.
|
||||
guard rename(partial, path) == 0 else {
|
||||
throw Abort(
|
||||
.internalServerError, reason: "cannot rename \(partial): errno \(errno)")
|
||||
}
|
||||
} catch {
|
||||
try? FileManager.default.removeItem(atPath: partial)
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
func storedSize(_ token: String) async -> Int? {
|
||||
@@ -75,8 +132,28 @@ struct DirectoryAttachmentStore: AttachmentStore {
|
||||
|
||||
// MARK: - S3
|
||||
|
||||
/// The HTTP hop to the bucket. Production is AsyncHTTPClient; tests swap in a
|
||||
/// bucket held in memory.
|
||||
protocol S3Transport: Sendable {
|
||||
/// The response body may still be arriving when this returns.
|
||||
func execute(_ request: HTTPClientRequest) async throws -> HTTPClientResponse
|
||||
}
|
||||
|
||||
/// AsyncHTTPClient streams a request body and a response body. Vapor's
|
||||
/// `Client` collects both into memory.
|
||||
struct HTTPClientS3Transport: S3Transport {
|
||||
let client: HTTPClient
|
||||
/// The whole exchange, body included. A download is paced by the phone
|
||||
/// reading it, and 10 MB at 20 KB/s takes about 8 minutes.
|
||||
var timeout: TimeAmount = .minutes(15)
|
||||
|
||||
func execute(_ request: HTTPClientRequest) async throws -> HTTPClientResponse {
|
||||
try await client.execute(request, timeout: timeout)
|
||||
}
|
||||
}
|
||||
|
||||
/// An S3-compatible bucket, spoken to directly with SigV4, no SDK, because
|
||||
/// three verbs against one bucket do not need one.
|
||||
/// four verbs against one bucket do not need one.
|
||||
struct S3AttachmentStore: AttachmentStore {
|
||||
struct S3Config: Sendable {
|
||||
let endpoint: String // e.g. https://s3.us-west-004.backblazeb2.com
|
||||
@@ -88,13 +165,13 @@ struct S3AttachmentStore: AttachmentStore {
|
||||
}
|
||||
|
||||
let s3: S3Config
|
||||
let client: any Client
|
||||
let transport: any S3Transport
|
||||
/// The migration fallback, the directory blobs used to live in.
|
||||
let localFallback: DirectoryAttachmentStore
|
||||
|
||||
private func key(_ token: String) -> String { "\(s3.prefix)\(token).bin" }
|
||||
private func url(_ token: String) -> URI {
|
||||
URI(string: "\(s3.endpoint)/\(s3.bucket)/\(key(token))")
|
||||
private func url(_ token: String) -> String {
|
||||
"\(s3.endpoint)/\(s3.bucket)/\(key(token))"
|
||||
}
|
||||
|
||||
/// Nil, deliberately.
|
||||
@@ -118,8 +195,7 @@ struct S3AttachmentStore: AttachmentStore {
|
||||
/// nothing, reads nothing, and cannot return anybody's data.
|
||||
func reachability() async -> StorageReachability {
|
||||
do {
|
||||
let response = try await signedRequest(
|
||||
.HEAD, token: "status-probe-does-not-exist", body: Data())
|
||||
let response = try await send(.HEAD, token: "status-probe-does-not-exist")
|
||||
switch response.status.code {
|
||||
case 200, 204, 404: return .reachable
|
||||
default: return .unreachable
|
||||
@@ -129,8 +205,32 @@ struct S3AttachmentStore: AttachmentStore {
|
||||
}
|
||||
}
|
||||
|
||||
func write(_ token: String, data: Data) async throws {
|
||||
let response = try await signedRequest(.PUT, token: token, body: data)
|
||||
/// One PUT, the body passed through as it arrives.
|
||||
///
|
||||
/// The payload is signed as `UNSIGNED-PAYLOAD`. A SigV4 payload hash has
|
||||
/// to be in the headers before the first byte, and the hash is known only
|
||||
/// after the last. B2 refuses the `aws-chunked` signing that signs chunk
|
||||
/// by chunk. `x-amz-checksum-sha256` has the same problem as a header, and
|
||||
/// as a trailer it needs `aws-chunked` too. Three things protect the bytes
|
||||
/// instead: TLS to the bucket, the `Content-Length` the bucket holds the
|
||||
/// upload to, and the client's proof, which signs the SHA-256 of the body
|
||||
/// and is checked against the bytes that went past.
|
||||
func write<Body: AsyncSequence & Sendable>(
|
||||
_ token: String, length: Int, body: Body
|
||||
) async throws where Body.Element == ByteBuffer {
|
||||
var request = try signedRequest(.PUT, token: token, payloadHash: "UNSIGNED-PAYLOAD")
|
||||
request.body = .stream(body, length: .known(Int64(length)))
|
||||
let response: HTTPClientResponse
|
||||
do {
|
||||
response = try await transport.execute(request)
|
||||
} catch {
|
||||
// The body ended short, failed, or the connection dropped. A
|
||||
// bucket does not keep a PUT it did not receive in full; the
|
||||
// DELETE makes sure of it.
|
||||
try? await delete(token)
|
||||
throw error
|
||||
}
|
||||
await Self.discardBody(response)
|
||||
guard response.status == .ok else {
|
||||
throw Abort(
|
||||
.internalServerError,
|
||||
@@ -138,7 +238,8 @@ struct S3AttachmentStore: AttachmentStore {
|
||||
}
|
||||
}
|
||||
|
||||
/// The range is asked of the bucket, not applied after the fact.
|
||||
/// The range is asked of the bucket, not applied after the fact, and the
|
||||
/// bytes go to the client as they arrive from the bucket.
|
||||
func respond(
|
||||
_ token: String, request: Request,
|
||||
range: ClosedRange<Int>?
|
||||
@@ -150,9 +251,7 @@ struct S3AttachmentStore: AttachmentStore {
|
||||
value: AttachmentRange.upstreamHeader(
|
||||
first: range.lowerBound, last: range.upperBound))
|
||||
}
|
||||
let response = try await signedRequest(
|
||||
.GET, token: token, body: Data(),
|
||||
extraHeaders: extra)
|
||||
let response = try await send(.GET, token: token, extraHeaders: extra)
|
||||
// 206 when a range was asked for, 200 when it was not.
|
||||
if response.status == .ok || response.status == .partialContent {
|
||||
var headers = HTTPHeaders()
|
||||
@@ -160,9 +259,20 @@ struct S3AttachmentStore: AttachmentStore {
|
||||
if let contentRange = response.headers.first(name: .contentRange) {
|
||||
headers.replaceOrAdd(name: .contentRange, value: contentRange)
|
||||
}
|
||||
let body = response.body.map { Response.Body(buffer: $0) } ?? .empty
|
||||
// -1 sends it chunked when the bucket gave no length.
|
||||
let count = response.headers.first(name: .contentLength).flatMap { Int($0) } ?? -1
|
||||
let upstream = TakeOnce(response.body)
|
||||
let body = Response.Body(
|
||||
managedAsyncStream: { writer in
|
||||
// Vapor runs this closure a second time, with a writer
|
||||
// that only fails, when writing the response to the
|
||||
// client fails. A bucket body traps on a second read.
|
||||
guard let upstream = upstream.take() else { return }
|
||||
for try await chunk in upstream { try await writer.writeBuffer(chunk) }
|
||||
}, count: count)
|
||||
return Response(status: response.status, headers: headers, body: body)
|
||||
}
|
||||
await Self.discardBody(response)
|
||||
if response.status == .notFound {
|
||||
// Pre-switch blobs still on disk.
|
||||
return try await localFallback.respond(token, request: request, range: range)
|
||||
@@ -171,7 +281,7 @@ struct S3AttachmentStore: AttachmentStore {
|
||||
}
|
||||
|
||||
func delete(_ token: String) async throws {
|
||||
let response = try await signedRequest(.DELETE, token: token, body: Data())
|
||||
let response = try await send(.DELETE, token: token)
|
||||
// 204 = deleted; 404 = already gone.
|
||||
guard
|
||||
response.status == .noContent || response.status == .ok
|
||||
@@ -186,21 +296,43 @@ struct S3AttachmentStore: AttachmentStore {
|
||||
|
||||
// MARK: SigV4
|
||||
|
||||
/// A request with no body. The reply body is left unread for a GET and
|
||||
/// read and dropped for anything else.
|
||||
private func send(
|
||||
_ method: HTTPMethod, token: String,
|
||||
extraHeaders: HTTPHeaders = [:]
|
||||
) async throws -> HTTPClientResponse {
|
||||
let response = try await transport.execute(
|
||||
signedRequest(
|
||||
method, token: token, payloadHash: Self.emptyPayloadHash,
|
||||
extraHeaders: extraHeaders))
|
||||
if method != .GET { await Self.discardBody(response) }
|
||||
return response
|
||||
}
|
||||
|
||||
/// An error document or nothing. Reading it returns the connection to the
|
||||
/// pool; dropping it unread closes the connection.
|
||||
private static func discardBody(_ response: HTTPClientResponse) async {
|
||||
_ = try? await response.body.collect(upTo: 64 * 1024)
|
||||
}
|
||||
|
||||
/// SHA-256 of zero bytes.
|
||||
private static let emptyPayloadHash = SHA256.hash(data: Data()).hexString
|
||||
|
||||
/// `extraHeaders` are sent but NOT signed, and that is correct rather than
|
||||
/// a shortcut: SigV4 covers whatever `SignedHeaders` names, and `Range` is
|
||||
/// deliberately not among them.
|
||||
private func signedRequest(
|
||||
_ method: HTTPMethod, token: String,
|
||||
body: Data,
|
||||
payloadHash: String,
|
||||
extraHeaders: HTTPHeaders = [:]
|
||||
) async throws -> ClientResponse {
|
||||
) throws -> HTTPClientRequest {
|
||||
guard let host = URL(string: s3.endpoint)?.host else {
|
||||
throw Abort(.internalServerError, reason: "S3 endpoint has no host")
|
||||
}
|
||||
let now = Date()
|
||||
let amzDate = Self.amzDateFormatter.string(from: now)
|
||||
let dateStamp = String(amzDate.prefix(8))
|
||||
let payloadHash = SHA256.hash(data: body).hexString
|
||||
|
||||
let canonicalURI = "/\(s3.bucket)/\(key(token))"
|
||||
let canonicalHeaders =
|
||||
@@ -242,11 +374,10 @@ struct S3AttachmentStore: AttachmentStore {
|
||||
+ "SignedHeaders=\(signedHeaders), Signature=\(signature)")
|
||||
for header in extraHeaders { headers.add(name: header.name, value: header.value) }
|
||||
|
||||
return try await client.send(method, headers: headers, to: url(token)) { req in
|
||||
if !body.isEmpty {
|
||||
req.body = ByteBuffer(data: body)
|
||||
}
|
||||
}
|
||||
var request = HTTPClientRequest(url: url(token))
|
||||
request.method = method
|
||||
request.headers = headers
|
||||
return request
|
||||
}
|
||||
|
||||
/// 20260829T012345Z.
|
||||
@@ -259,6 +390,22 @@ struct S3AttachmentStore: AttachmentStore {
|
||||
}()
|
||||
}
|
||||
|
||||
/// Hands its value out once.
|
||||
private final class TakeOnce<Value: Sendable>: @unchecked Sendable {
|
||||
private let lock = NSLock()
|
||||
private var value: Value?
|
||||
|
||||
init(_ value: Value) { self.value = value }
|
||||
|
||||
func take() -> Value? {
|
||||
lock.lock()
|
||||
defer { lock.unlock() }
|
||||
let taken = value
|
||||
value = nil
|
||||
return taken
|
||||
}
|
||||
}
|
||||
|
||||
extension SHA256.Digest {
|
||||
fileprivate var hexString: String { map { String(format: "%02x", $0) }.joined() }
|
||||
}
|
||||
|
||||
@@ -212,7 +212,7 @@ func configure(_ app: Application) async throws {
|
||||
endpoint: endpoint, bucket: config.s3Bucket,
|
||||
region: config.s3Region, accessKey: config.s3AccessKey,
|
||||
secretKey: config.s3SecretKey, prefix: config.s3Prefix),
|
||||
client: app.client,
|
||||
transport: HTTPClientS3Transport(client: app.http.client.shared),
|
||||
localFallback: directoryStore)
|
||||
app.logger.notice(
|
||||
"attachments in S3, bucket \(config.s3Bucket), reads fall back to disk for pre-switch blobs"
|
||||
@@ -1425,14 +1425,11 @@ func register(routes r: Routes, on app: Application) throws {
|
||||
try await r.setDeliveryTokens(req, body: body(of: req), now: unixNow())
|
||||
}
|
||||
|
||||
// §8.7, the body is a raw ciphertext blob, not JSON, and needs its own
|
||||
// collection limit: the global default is deliberately small so that an
|
||||
// unauthenticated sealed submission cannot ask for much memory.
|
||||
e2e.on(
|
||||
.POST, "attachments",
|
||||
body: .collect(maxSize: ByteCount(value: r.config.attachmentMaxBytes))
|
||||
) { req in
|
||||
try await r.uploadAttachment(req, body: body(of: req), now: unixNow())
|
||||
// §8.7, the body is a raw ciphertext blob, not JSON, and is streamed to
|
||||
// storage rather than collected. `uploadAttachment` authenticates and
|
||||
// checks Content-Length against the size limit before reading a byte.
|
||||
e2e.on(.POST, "attachments", body: .stream) { req in
|
||||
try await r.uploadAttachment(req, now: unixNow())
|
||||
}
|
||||
e2e.get("attachments", ":token") { req in
|
||||
guard let token = req.parameters.get("token") else {
|
||||
|
||||
@@ -8,16 +8,41 @@ extension Routes {
|
||||
|
||||
// MARK: - POST /e2e/attachments
|
||||
|
||||
func uploadAttachment(_ request: Request, body: Data, now: Int) async throws -> Response {
|
||||
// The proof signs a digest of the body, which here is the whole blob.
|
||||
let caller = try await authenticate(
|
||||
request, body: body, path: "/e2e/attachments", now: now)
|
||||
/// The route's entry: the body is the request's own stream, never
|
||||
/// collected.
|
||||
func uploadAttachment(_ request: Request, now: Int) async throws -> Response {
|
||||
try await uploadAttachment(
|
||||
request, body: request.body, length: Self.declaredLength(of: request), now: now)
|
||||
}
|
||||
|
||||
/// `Content-Length`, or the size of a body that is already in memory.
|
||||
static func declaredLength(of request: Request) -> Int? {
|
||||
if let header = request.headers.first(name: .contentLength) { return Int(header) }
|
||||
return request.body.data?.readableBytes
|
||||
}
|
||||
|
||||
/// Up to 10 MB, streamed to storage without being held.
|
||||
///
|
||||
/// Everything that does not need the body is checked before its first
|
||||
/// byte is read: the token, the proof headers, the device, the account,
|
||||
/// the size, and the rate limits. The proof's signature covers the
|
||||
/// SHA-256 of the body, so it is checked after the last byte, against a
|
||||
/// hash taken as the bytes went past, and a blob whose proof fails is
|
||||
/// deleted again.
|
||||
func uploadAttachment<Body: AsyncSequence & Sendable>(
|
||||
_ request: Request, body: Body, length: Int?, now: Int
|
||||
) async throws -> Response where Body.Element == ByteBuffer {
|
||||
let path = "/e2e/attachments"
|
||||
let (claims, caller) = try await authenticateBeforeBody(request, path: path, now: now)
|
||||
|
||||
guard let length else {
|
||||
throw WireError("length_required", "Content-Length is required.", 411)
|
||||
}
|
||||
// An AES-GCM sealed box is at least nonce(12) + tag(16).
|
||||
guard body.count >= 28 else {
|
||||
guard length >= 28 else {
|
||||
throw WireError.badRequest("Attachment body required.")
|
||||
}
|
||||
guard body.count <= config.attachmentMaxBytes else {
|
||||
guard length <= config.attachmentMaxBytes else {
|
||||
throw WireError("too_large", "Attachment exceeds the size limit.", 413)
|
||||
}
|
||||
|
||||
@@ -25,7 +50,7 @@ extension Routes {
|
||||
[
|
||||
(key: "att:count:\(caller.userID)", amount: 1, limit: config.attachmentRatePerHour),
|
||||
(
|
||||
key: "att:bytes:\(caller.userID)", amount: body.count,
|
||||
key: "att:bytes:\(caller.userID)", amount: length,
|
||||
limit: config.attachmentBytesPerHour
|
||||
),
|
||||
], now: now)
|
||||
@@ -35,7 +60,7 @@ extension Routes {
|
||||
}
|
||||
// The month-scale budget (C3).
|
||||
let withinBudget = await attachmentQuota.consume(
|
||||
"att:month:\(caller.userID)", amount: body.count,
|
||||
"att:month:\(caller.userID)", amount: length,
|
||||
limit: config.attachmentBytesPerMonth, now: now)
|
||||
guard withinBudget else {
|
||||
throw WireError(
|
||||
@@ -48,17 +73,34 @@ extension Routes {
|
||||
// travels inside the ratchet, so it has to be unguessable, not merely
|
||||
// unique.
|
||||
let token = randomHexToken()
|
||||
let metered = MeteredBody(base: body, declared: length)
|
||||
do {
|
||||
try await attachmentStore.write(token, data: body)
|
||||
try await attachmentStore.write(token, length: length, body: metered)
|
||||
} catch {
|
||||
if let mismatch = metered.mismatch {
|
||||
// Usually the phone dropped the connection part way.
|
||||
request.logger.notice(
|
||||
"attachment upload ended at \(mismatch.received) of \(mismatch.declared) bytes")
|
||||
throw WireError(
|
||||
"incomplete_body",
|
||||
"The attachment body did not match its Content-Length.", 400)
|
||||
}
|
||||
request.logger.error("attachment write failed: \(error)")
|
||||
throw WireError("storage_error", "Could not store attachment.", 500)
|
||||
}
|
||||
|
||||
do {
|
||||
try await authenticateAfterBody(
|
||||
request, claims: claims, path: path, digest: metered.digest, now: now)
|
||||
} catch {
|
||||
try? await attachmentStore.delete(token)
|
||||
throw error
|
||||
}
|
||||
|
||||
do {
|
||||
let row = Attachment()
|
||||
row.id = token
|
||||
row.bytes = body.count
|
||||
row.bytes = length
|
||||
try await row.create(on: request.db)
|
||||
} catch {
|
||||
// The row is the only thing retention can see.
|
||||
@@ -67,7 +109,68 @@ extension Routes {
|
||||
throw WireError("storage_error", "Could not index attachment.", 500)
|
||||
}
|
||||
|
||||
return try .json(AttachmentUploadResponse(ok: true, id: token, bytes: body.count))
|
||||
return try .json(AttachmentUploadResponse(ok: true, id: token, bytes: length))
|
||||
}
|
||||
|
||||
/// `authenticated` without the proof's signature, which needs the body.
|
||||
/// The same checks in the same order, and the same errors.
|
||||
private func authenticateBeforeBody(
|
||||
_ request: Request, path: String, now: Int
|
||||
) async throws -> (TokenClaims, Caller) {
|
||||
guard let token = request.headers.bearerAuthorization?.token else {
|
||||
throw WireError.unauthorized()
|
||||
}
|
||||
do {
|
||||
let claims = try verifier.verify(token: token, now: now)
|
||||
let caller = try verifier.checkProofHeaders(
|
||||
claims: claims,
|
||||
timestamp: request.headers.first(name: Headers.timestamp),
|
||||
nonce: request.headers.first(name: Headers.nonce),
|
||||
signature: request.headers.first(name: Headers.signature),
|
||||
now: now)
|
||||
if let known = try await Device.find(caller.deviceID, on: request.db) {
|
||||
guard known.ikSig == caller.identityKey.base64EncodedString() else {
|
||||
throw AuthError.deviceMismatch
|
||||
}
|
||||
} else if try await KeyLog.isRetired(deviceID: caller.deviceID, on: request.db) {
|
||||
throw AuthError.deviceRetired
|
||||
}
|
||||
if let user = try await User.find(caller.userID, on: request.db) {
|
||||
if user.suspendedAt != nil { throw User.suspendedRefusal }
|
||||
} else if config.issuerSigningKeyRaw != nil {
|
||||
throw WireError.unauthorized("This account no longer exists.")
|
||||
}
|
||||
return (claims, caller)
|
||||
} catch AuthError.deviceRetired {
|
||||
throw WireError.deviceRevoked
|
||||
} catch let e as AuthError {
|
||||
throw WireError.unauthorized(e.description)
|
||||
}
|
||||
}
|
||||
|
||||
/// The signature over the streamed body's digest, then the replay check.
|
||||
private func authenticateAfterBody(
|
||||
_ request: Request, claims: TokenClaims, path: String, digest: String, now: Int
|
||||
) async throws {
|
||||
let proofSig = request.headers.first(name: Headers.signature)
|
||||
do {
|
||||
_ = try verifier.verifyProof(
|
||||
claims: claims, method: request.method.string, path: path,
|
||||
bodyDigest: digest,
|
||||
timestamp: request.headers.first(name: Headers.timestamp),
|
||||
nonce: request.headers.first(name: Headers.nonce),
|
||||
signature: proofSig, now: now)
|
||||
guard let proofSig, await replay.admit(proofSig, now: now) else {
|
||||
throw AuthError.replayed
|
||||
}
|
||||
} catch AuthError.replayed {
|
||||
throw WireError(
|
||||
"proof_replayed",
|
||||
AuthError.replayed.description + "; sign the request again, the token is fine",
|
||||
401)
|
||||
} catch let e as AuthError {
|
||||
throw WireError.unauthorized(e.description)
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: - GET /e2e/attachments/{id}
|
||||
@@ -144,3 +247,76 @@ func randomHexToken() -> String {
|
||||
raw.map { String(format: "%02x", $0) }.joined()
|
||||
}
|
||||
}
|
||||
|
||||
/// An upload body on its way to storage, counted and hashed as it passes.
|
||||
///
|
||||
/// It throws when the body runs past `declared`, and when it ends short of
|
||||
/// it. A dropped connection ends a Vapor request stream the same way a
|
||||
/// finished one does, so the count is the only thing that tells them apart.
|
||||
struct MeteredBody<Base: AsyncSequence & Sendable>: AsyncSequence, Sendable
|
||||
where Base.Element == ByteBuffer {
|
||||
typealias Element = ByteBuffer
|
||||
|
||||
let base: Base
|
||||
let declared: Int
|
||||
private let state = State()
|
||||
|
||||
init(base: Base, declared: Int) {
|
||||
self.base = base
|
||||
self.declared = declared
|
||||
}
|
||||
|
||||
/// Hex SHA-256 of every byte that went past.
|
||||
var digest: String {
|
||||
state.current.hash.finalize().map { String(format: "%02x", $0) }.joined()
|
||||
}
|
||||
|
||||
/// Set when the body ended at a length other than `declared`.
|
||||
var mismatch: AttachmentLengthMismatch? { state.current.mismatch }
|
||||
|
||||
func makeAsyncIterator() -> Iterator {
|
||||
Iterator(base: base.makeAsyncIterator(), declared: declared, state: state)
|
||||
}
|
||||
|
||||
struct Iterator: AsyncIteratorProtocol {
|
||||
var base: Base.AsyncIterator
|
||||
let declared: Int
|
||||
let state: State
|
||||
|
||||
mutating func next() async throws -> ByteBuffer? {
|
||||
let chunk = try await base.next()
|
||||
try state.record(chunk, declared: declared)
|
||||
return chunk
|
||||
}
|
||||
}
|
||||
|
||||
final class State: @unchecked Sendable {
|
||||
struct Values {
|
||||
var received = 0
|
||||
var hash = SHA256()
|
||||
var mismatch: AttachmentLengthMismatch?
|
||||
}
|
||||
private let lock = NSLock()
|
||||
private var values = Values()
|
||||
|
||||
var current: Values {
|
||||
lock.lock()
|
||||
defer { lock.unlock() }
|
||||
return values
|
||||
}
|
||||
|
||||
/// Nil is the end of the body.
|
||||
func record(_ chunk: ByteBuffer?, declared: Int) throws {
|
||||
lock.lock()
|
||||
defer { lock.unlock() }
|
||||
values.received += chunk?.readableBytes ?? 0
|
||||
if values.received > declared || (chunk == nil && values.received < declared) {
|
||||
let mismatch = AttachmentLengthMismatch(
|
||||
declared: declared, received: values.received)
|
||||
values.mismatch = mismatch
|
||||
throw mismatch
|
||||
}
|
||||
if let chunk { values.hash.update(data: chunk.readableBytesView) }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -131,6 +131,62 @@ struct TokenVerifier: Sendable {
|
||||
signature: String?,
|
||||
now: Int
|
||||
) throws -> Caller {
|
||||
let digest = SHA256.hash(data: body).map { String(format: "%02x", $0) }.joined()
|
||||
return try verifyProof(
|
||||
claims: claims, method: method, path: path, bodyDigest: digest,
|
||||
timestamp: timestamp, nonce: nonce, signature: signature, now: now)
|
||||
}
|
||||
|
||||
/// The same check over a digest the caller computed, for a body that was
|
||||
/// streamed to storage instead of held in memory (attachment uploads).
|
||||
func verifyProof(
|
||||
claims: TokenClaims,
|
||||
method: String,
|
||||
path: String,
|
||||
bodyDigest digest: String,
|
||||
timestamp: String?,
|
||||
nonce: String?,
|
||||
signature: String?,
|
||||
now: Int
|
||||
) throws -> Caller {
|
||||
let proof = try proofParts(
|
||||
claims: claims, timestamp: timestamp, nonce: nonce,
|
||||
signature: signature, now: now)
|
||||
let canonical = Data(
|
||||
"\(method)\n\(path)\n\(proof.timestamp)\n\(proof.nonce)\n\(digest)".utf8)
|
||||
guard proof.deviceKey.isValidSignature(proof.signature, for: canonical) else {
|
||||
throw AuthError.badProof
|
||||
}
|
||||
return proof.caller
|
||||
}
|
||||
|
||||
/// Every part of the proof that can be checked before the body arrives:
|
||||
/// the headers are present, the timestamp is inside the skew window, and
|
||||
/// the key and signature decode. The signature covers the body's SHA-256,
|
||||
/// so verifying it has to wait for the last byte.
|
||||
func checkProofHeaders(
|
||||
claims: TokenClaims,
|
||||
timestamp: String?,
|
||||
nonce: String?,
|
||||
signature: String?,
|
||||
now: Int
|
||||
) throws -> Caller {
|
||||
try proofParts(
|
||||
claims: claims, timestamp: timestamp, nonce: nonce,
|
||||
signature: signature, now: now
|
||||
).caller
|
||||
}
|
||||
|
||||
private func proofParts(
|
||||
claims: TokenClaims,
|
||||
timestamp: String?,
|
||||
nonce: String?,
|
||||
signature: String?,
|
||||
now: Int
|
||||
) throws -> (
|
||||
caller: Caller, deviceKey: Curve25519.Signing.PublicKey,
|
||||
signature: Data, timestamp: String, nonce: String
|
||||
) {
|
||||
guard let timestamp, let signature, let nonce, !nonce.isEmpty else {
|
||||
throw AuthError.missingProof
|
||||
}
|
||||
@@ -141,13 +197,8 @@ struct TokenVerifier: Sendable {
|
||||
let sigBytes = Data(base64Encoded: signature)
|
||||
else { throw AuthError.badProof }
|
||||
|
||||
let digest = SHA256.hash(data: body).map { String(format: "%02x", $0) }.joined()
|
||||
let canonical = Data("\(method)\n\(path)\n\(timestamp)\n\(nonce)\n\(digest)".utf8)
|
||||
guard deviceKey.isValidSignature(sigBytes, for: canonical) else {
|
||||
throw AuthError.badProof
|
||||
}
|
||||
|
||||
return Caller(userID: claims.sub, deviceID: claims.dev, identityKey: keyBytes)
|
||||
let caller = Caller(userID: claims.sub, deviceID: claims.dev, identityKey: keyBytes)
|
||||
return (caller, deviceKey, sigBytes, timestamp, nonce)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -131,7 +131,7 @@ struct AttachmentRangeTests {
|
||||
let body = Data(contents.utf8)
|
||||
let response = try await h.routes.uploadAttachment(
|
||||
h.request(.POST, "/e2e/attachments", as: d, body: body, now: now),
|
||||
body: body, now: now)
|
||||
now: now)
|
||||
struct Up: Decodable { let id: String }
|
||||
return try JSONDecoder().decode(
|
||||
Up.self, from: Data((response.body.string ?? "").utf8)
|
||||
|
||||
@@ -22,7 +22,7 @@ struct AttachmentRoutesTests {
|
||||
) async throws -> Response {
|
||||
try await h.routes.uploadAttachment(
|
||||
h.request(.POST, "/e2e/attachments", as: d, body: body, now: now),
|
||||
body: body, now: now)
|
||||
now: now)
|
||||
}
|
||||
|
||||
private func token(of res: Response) throws -> String {
|
||||
@@ -158,7 +158,7 @@ struct AttachmentRoutesTests {
|
||||
let body = blob(64)
|
||||
let req = h.anonymousRequest(.POST, "/e2e/attachments", body: body)
|
||||
await #expect(throws: WireError.self) {
|
||||
_ = try await h.routes.uploadAttachment(req, body: body, now: now)
|
||||
_ = try await h.routes.uploadAttachment(req, now: now)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,452 @@
|
||||
import AsyncHTTPClient
|
||||
import Crypto
|
||||
import Fluent
|
||||
import Foundation
|
||||
import NIOCore
|
||||
import Testing
|
||||
import Vapor
|
||||
|
||||
@testable import FlamenetRelay
|
||||
|
||||
/// Uploads streamed to storage and downloads streamed from it (relay review
|
||||
/// 2026-10-14): nothing read before auth, nothing stored that did not arrive
|
||||
/// whole, and no blob held in memory on the way out.
|
||||
@Suite("attachment streaming")
|
||||
struct AttachmentStreamingTests {
|
||||
|
||||
private let now = 1_700_000_000
|
||||
|
||||
// MARK: - Helpers
|
||||
|
||||
/// Records whether anybody asked it for a byte.
|
||||
struct UntouchedBody: AsyncSequence, Sendable {
|
||||
typealias Element = ByteBuffer
|
||||
let touched = Flag()
|
||||
|
||||
final class Flag: @unchecked Sendable {
|
||||
private let lock = NSLock()
|
||||
private var _value = false
|
||||
var value: Bool { lock.withLock { _value } }
|
||||
func set() { lock.withLock { _value = true } }
|
||||
}
|
||||
|
||||
struct AsyncIterator: AsyncIteratorProtocol {
|
||||
let touched: Flag
|
||||
mutating func next() async throws -> ByteBuffer? {
|
||||
touched.set()
|
||||
return nil
|
||||
}
|
||||
}
|
||||
func makeAsyncIterator() -> AsyncIterator { AsyncIterator(touched: touched) }
|
||||
}
|
||||
|
||||
/// `data` in pieces of `size`, as a socket delivers it.
|
||||
private func pieces(_ data: Data, size: Int) -> AsyncStream<ByteBuffer> {
|
||||
AsyncStream { continuation in
|
||||
var offset = 0
|
||||
while offset < data.count {
|
||||
let end = min(offset + size, data.count)
|
||||
continuation.yield(ByteBuffer(data: data.subdata(in: offset..<end)))
|
||||
offset = end
|
||||
}
|
||||
continuation.finish()
|
||||
}
|
||||
}
|
||||
|
||||
private func s3(_ bucket: FakeBucket, fallback: String) -> S3AttachmentStore {
|
||||
S3AttachmentStore(
|
||||
s3: .init(
|
||||
endpoint: "https://s3.example.com", bucket: "blobs",
|
||||
region: "us-west-004", accessKey: "AKIA", secretKey: "secret",
|
||||
prefix: "attachments/"),
|
||||
transport: bucket,
|
||||
localFallback: DirectoryAttachmentStore(directory: fallback))
|
||||
}
|
||||
|
||||
/// The harness's routes over a different store.
|
||||
private func routes(_ h: RelayHarness, store: any AttachmentStore) -> FlamenetRelay.Routes {
|
||||
let r = h.routes
|
||||
return FlamenetRelay.Routes(
|
||||
config: r.config, verifier: r.verifier, limits: r.limits,
|
||||
sendMetrics: r.sendMetrics, federation: r.federation,
|
||||
federationSender: r.federationSender, replay: r.replay,
|
||||
attachmentStore: store, attachmentQuota: r.attachmentQuota)
|
||||
}
|
||||
|
||||
private func device(_ h: RelayHarness) async throws -> RelayHarness.TestDevice {
|
||||
let d = h.device()
|
||||
try await h.register(d, on: h.db)
|
||||
return d
|
||||
}
|
||||
|
||||
private func blob(_ n: Int) -> Data {
|
||||
Data((0..<n).map { UInt8(truncatingIfNeeded: $0 &* 31) })
|
||||
}
|
||||
|
||||
private func code(_ error: (any Error)?) -> String? { (error as? WireError)?.code }
|
||||
|
||||
// MARK: - Before the body
|
||||
|
||||
@Test("an unauthenticated upload is refused without reading the body")
|
||||
func unauthenticatedReadsNothing() async throws {
|
||||
try await withRelay { h in
|
||||
let d = try await device(h)
|
||||
// No credentials at all.
|
||||
let anonymous = UntouchedBody()
|
||||
await #expect(throws: WireError.self) {
|
||||
_ = try await h.routes.uploadAttachment(
|
||||
h.anonymousRequest(.POST, "/e2e/attachments"),
|
||||
body: anonymous, length: 64, now: now)
|
||||
}
|
||||
#expect(!anonymous.touched.value)
|
||||
|
||||
// A good token with the proof headers stripped.
|
||||
let request = h.request(.POST, "/e2e/attachments", as: d, body: blob(64), now: now)
|
||||
request.headers.remove(name: FlamenetRelay.Routes.Headers.signature)
|
||||
let stripped = UntouchedBody()
|
||||
await #expect(throws: WireError.self) {
|
||||
_ = try await h.routes.uploadAttachment(
|
||||
request, body: stripped, length: 64, now: now)
|
||||
}
|
||||
#expect(!stripped.touched.value)
|
||||
|
||||
// An expired token.
|
||||
let expired = h.request(
|
||||
.POST, "/e2e/attachments", as: d, body: blob(64), now: now,
|
||||
token: h.token(for: d, now: now - 7200, expiresIn: 60))
|
||||
let late = UntouchedBody()
|
||||
let err = await #expect(throws: WireError.self) {
|
||||
_ = try await h.routes.uploadAttachment(expired, body: late, length: 64, now: now)
|
||||
}
|
||||
#expect((err as? WireError)?.data.status == 401)
|
||||
#expect(!late.touched.value)
|
||||
}
|
||||
}
|
||||
|
||||
@Test("an oversize or missing Content-Length is refused without reading the body")
|
||||
func oversizeReadsNothing() async throws {
|
||||
try await withRelay(env: ["FN_ATTACHMENT_MAX_BYTES": "1024"]) { h in
|
||||
let d = try await device(h)
|
||||
let big = UntouchedBody()
|
||||
let err = await #expect(throws: WireError.self) {
|
||||
_ = try await h.routes.uploadAttachment(
|
||||
h.request(.POST, "/e2e/attachments", as: d, now: now),
|
||||
body: big, length: 1025, now: now)
|
||||
}
|
||||
#expect(code(err) == "too_large")
|
||||
#expect(!big.touched.value)
|
||||
|
||||
let unsized = UntouchedBody()
|
||||
let missing = await #expect(throws: WireError.self) {
|
||||
_ = try await h.routes.uploadAttachment(
|
||||
h.request(.POST, "/e2e/attachments", as: d, now: now),
|
||||
body: unsized, length: nil, now: now)
|
||||
}
|
||||
#expect(code(missing) == "length_required")
|
||||
#expect(!unsized.touched.value)
|
||||
#expect(try await Attachment.query(on: h.db).count() == 0)
|
||||
}
|
||||
}
|
||||
|
||||
/// The route reads the header, and a streamed request has no other
|
||||
/// length.
|
||||
@Test("the declared length comes from Content-Length")
|
||||
func declaredLength() async throws {
|
||||
try await withRelay { h in
|
||||
let request = h.anonymousRequest(.POST, "/e2e/attachments", body: blob(10))
|
||||
#expect(FlamenetRelay.Routes.declaredLength(of: request) == 10)
|
||||
request.headers.replaceOrAdd(name: .contentLength, value: "4096")
|
||||
#expect(FlamenetRelay.Routes.declaredLength(of: request) == 4096)
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: - The body
|
||||
|
||||
@Test("a streamed upload lands whole in the bucket, signed UNSIGNED-PAYLOAD")
|
||||
func streamedUploadToBucket() async throws {
|
||||
try await withRelay { h in
|
||||
let d = try await device(h)
|
||||
let bucket = FakeBucket()
|
||||
let r = routes(h, store: s3(bucket, fallback: h.attachmentDir))
|
||||
let contents = blob(100_000)
|
||||
|
||||
let response = try await r.uploadAttachment(
|
||||
h.request(.POST, "/e2e/attachments", as: d, body: contents, now: now),
|
||||
body: pieces(contents, size: 4096), length: contents.count, now: now)
|
||||
#expect(response.status == .ok)
|
||||
|
||||
#expect(bucket.objects.count == 1)
|
||||
#expect(bucket.objects.values.first == contents)
|
||||
let put = try #require(bucket.requests.first { $0.method == .PUT })
|
||||
#expect(put.headers.first(name: "x-amz-content-sha256") == "UNSIGNED-PAYLOAD")
|
||||
#expect(
|
||||
put.headers.first(name: "authorization")?.contains("x-amz-content-sha256") == true)
|
||||
let row = try #require(try await Attachment.query(on: h.db).first())
|
||||
#expect(row.bytes == contents.count)
|
||||
}
|
||||
}
|
||||
|
||||
@Test("an upload that ends short leaves no object, no file and no row")
|
||||
func shortUploadLeavesNothing() async throws {
|
||||
try await withRelay { h in
|
||||
let d = try await device(h)
|
||||
let contents = blob(2000)
|
||||
let bucket = FakeBucket()
|
||||
let stores: [any AttachmentStore] = [
|
||||
s3(bucket, fallback: h.attachmentDir),
|
||||
DirectoryAttachmentStore(directory: h.attachmentDir),
|
||||
]
|
||||
for store in stores {
|
||||
let err = await #expect(throws: WireError.self) {
|
||||
_ = try await routes(h, store: store).uploadAttachment(
|
||||
h.request(.POST, "/e2e/attachments", as: d, body: contents, now: now),
|
||||
body: pieces(contents.prefix(1500), size: 500),
|
||||
length: contents.count, now: now)
|
||||
}
|
||||
#expect(code(err) == "incomplete_body")
|
||||
}
|
||||
#expect(bucket.objects.isEmpty)
|
||||
#expect(bucket.requests.contains { $0.method == .DELETE })
|
||||
let left = (try? FileManager.default.contentsOfDirectory(atPath: h.attachmentDir)) ?? []
|
||||
#expect(left.isEmpty, "left on disk: \(left)")
|
||||
#expect(try await Attachment.query(on: h.db).count() == 0)
|
||||
}
|
||||
}
|
||||
|
||||
@Test("a body that runs past its Content-Length is refused and not kept")
|
||||
func longUploadLeavesNothing() async throws {
|
||||
try await withRelay { h in
|
||||
let d = try await device(h)
|
||||
let contents = blob(2000)
|
||||
let err = await #expect(throws: WireError.self) {
|
||||
_ = try await h.routes.uploadAttachment(
|
||||
h.request(.POST, "/e2e/attachments", as: d, body: contents, now: now),
|
||||
body: pieces(contents, size: 500), length: 1500, now: now)
|
||||
}
|
||||
#expect(code(err) == "incomplete_body")
|
||||
let left = (try? FileManager.default.contentsOfDirectory(atPath: h.attachmentDir)) ?? []
|
||||
#expect(left.isEmpty, "left on disk: \(left)")
|
||||
}
|
||||
}
|
||||
|
||||
/// The proof signs the body's digest, which exists only after the last
|
||||
/// byte. A body other than the one signed is stored, then refused and
|
||||
/// deleted.
|
||||
@Test("a body that does not match the proof is deleted and refused")
|
||||
func proofMismatchDeletes() async throws {
|
||||
try await withRelay { h in
|
||||
let d = try await device(h)
|
||||
let bucket = FakeBucket()
|
||||
let r = routes(h, store: s3(bucket, fallback: h.attachmentDir))
|
||||
let signed = blob(4096)
|
||||
var sent = signed
|
||||
sent[100] ^= 0xFF
|
||||
|
||||
let err = await #expect(throws: WireError.self) {
|
||||
_ = try await r.uploadAttachment(
|
||||
h.request(.POST, "/e2e/attachments", as: d, body: signed, now: now),
|
||||
body: pieces(sent, size: 1024), length: sent.count, now: now)
|
||||
}
|
||||
#expect((err as? WireError)?.data.status == 401)
|
||||
#expect(bucket.requests.contains { $0.method == .PUT })
|
||||
#expect(bucket.objects.isEmpty)
|
||||
#expect(try await Attachment.query(on: h.db).count() == 0)
|
||||
}
|
||||
}
|
||||
|
||||
@Test("a replayed upload is refused after the body and not kept")
|
||||
func replayedUploadDeletes() async throws {
|
||||
try await withRelay { h in
|
||||
let d = try await device(h)
|
||||
let contents = blob(256)
|
||||
let request = h.request(.POST, "/e2e/attachments", as: d, body: contents, now: now)
|
||||
_ = try await h.routes.uploadAttachment(
|
||||
request, body: pieces(contents, size: 64), length: contents.count, now: now)
|
||||
let err = await #expect(throws: WireError.self) {
|
||||
_ = try await h.routes.uploadAttachment(
|
||||
request, body: pieces(contents, size: 64), length: contents.count, now: now)
|
||||
}
|
||||
#expect(code(err) == "proof_replayed")
|
||||
#expect(try await Attachment.query(on: h.db).count() == 1)
|
||||
let files = try FileManager.default.contentsOfDirectory(atPath: h.attachmentDir)
|
||||
#expect(files.count == 1)
|
||||
}
|
||||
}
|
||||
|
||||
/// The production route over a real socket: `body: .stream`, the length
|
||||
/// from the header, and the bytes on disk exactly as sent.
|
||||
@Test("an upload over a real connection streams to disk, and a refused one stores nothing")
|
||||
func uploadOverTheWire() async throws {
|
||||
try await withRelay { h in
|
||||
let d = try await device(h)
|
||||
// Production's error shape; the harness does not run `configure`.
|
||||
h.app.middleware.use(WireErrorMiddleware())
|
||||
try register(routes: h.routes, on: h.app)
|
||||
try await h.app.server.start(address: .hostname("127.0.0.1", port: 0))
|
||||
let port = try #require(h.app.http.server.shared.localAddress?.port)
|
||||
let contents = blob(3 * 1024 * 1024)
|
||||
|
||||
func post(_ headers: HTTPHeaders) async throws -> (HTTPResponseStatus, Data) {
|
||||
var request = HTTPClientRequest(url: "http://127.0.0.1:\(port)/e2e/attachments")
|
||||
request.method = .POST
|
||||
request.headers = headers
|
||||
request.body = .bytes(ByteBuffer(data: contents))
|
||||
let response = try await HTTPClient.shared.execute(request, timeout: .seconds(30))
|
||||
let body = try await response.body.collect(upTo: 1 << 16)
|
||||
return (response.status, Data(buffer: body))
|
||||
}
|
||||
|
||||
let (refused, _) = try await post([:])
|
||||
#expect(refused == .unauthorized, "\(refused)")
|
||||
|
||||
let signed = h.request(
|
||||
.POST, "/e2e/attachments", as: d, body: contents, now: unixNow())
|
||||
let (status, body) = try await post(signed.headers)
|
||||
await h.app.server.shutdown()
|
||||
|
||||
#expect(status == .ok, "\(String(decoding: body, as: UTF8.self))")
|
||||
struct Made: Decodable { let id: String }
|
||||
let id = try JSONDecoder().decode(Made.self, from: body).id
|
||||
#expect(
|
||||
FileManager.default.contents(atPath: attachmentPath(h.attachmentDir, id))
|
||||
== contents)
|
||||
#expect(try await Attachment.query(on: h.db).count() == 1)
|
||||
#expect(
|
||||
try FileManager.default.contentsOfDirectory(atPath: h.attachmentDir) == [
|
||||
"\(id).bin"
|
||||
])
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: - Downloads
|
||||
|
||||
@Test("a range is asked of the bucket and comes back as a 206 slice")
|
||||
func rangeThroughBucket() async throws {
|
||||
try await withRelay { h in
|
||||
let d = try await device(h)
|
||||
let bucket = FakeBucket()
|
||||
let r = routes(h, store: s3(bucket, fallback: h.attachmentDir))
|
||||
let contents = blob(5000)
|
||||
let up = try await r.uploadAttachment(
|
||||
h.request(.POST, "/e2e/attachments", as: d, body: contents, now: now),
|
||||
now: now)
|
||||
struct Made: Decodable { let id: String }
|
||||
let id = try JSONDecoder().decode(Made.self, from: Data(up.body.string!.utf8)).id
|
||||
|
||||
let request = h.request(.GET, "/e2e/attachments/\(id)", as: d, now: now)
|
||||
request.headers.add(name: .range, value: "bytes=1000-1999")
|
||||
let response = try await r.downloadAttachment(request, token: id, now: now)
|
||||
#expect(response.status == .partialContent)
|
||||
#expect(response.headers.first(name: .contentRange) == "bytes 1000-1999/5000")
|
||||
#expect(response.body.count == 1000)
|
||||
let body = try await response.body.collect(on: h.app.eventLoopGroup.next()).get()
|
||||
#expect(body.map { Data(buffer: $0) } == contents.subdata(in: 1000..<2000))
|
||||
let get = try #require(bucket.requests.last { $0.method == .GET })
|
||||
#expect(get.headers.first(name: "range") == "bytes=1000-1999")
|
||||
|
||||
// Vapor runs a body's stream a second time when writing it to the
|
||||
// client fails. A bucket body read twice traps the process.
|
||||
_ = try? await response.body.collect(on: h.app.eventLoopGroup.next()).get()
|
||||
}
|
||||
}
|
||||
|
||||
/// The bucket's body is held at chunk 4 of 256 until the client has
|
||||
/// received bytes. A relay that collected the object first would never
|
||||
/// send any, and the test would time out.
|
||||
///
|
||||
/// The timeout is long because the full suite can stall the cooperative
|
||||
/// thread pool for 30 seconds at a time (`blockingShutdown` waits on a
|
||||
/// semaphore). Alone, this test takes under a second.
|
||||
@Test("a download reaches the client while the bucket is still sending")
|
||||
func downloadStreams() async throws {
|
||||
try await withRelay { h in
|
||||
let chunkSize = 64 * 1024
|
||||
let chunks = 256 // 16 MB
|
||||
let hold = 4
|
||||
let (gate, open) = AsyncStream<Void>.makeStream()
|
||||
let produced = Counter()
|
||||
let token = randomHexToken()
|
||||
let bucket = FakeBucket(onGet: { _ in
|
||||
HTTPClientResponse(
|
||||
status: .ok,
|
||||
headers: ["content-length": "\(chunkSize * chunks)"],
|
||||
body: .stream(
|
||||
Held(
|
||||
chunk: ByteBuffer(repeating: 0x5A, count: chunkSize),
|
||||
count: chunks, holdAt: hold, gate: gate, produced: produced)))
|
||||
})
|
||||
let store = s3(bucket, fallback: h.attachmentDir)
|
||||
h.app.get("blob") { req in
|
||||
try await store.respond(token, request: req, range: nil)!
|
||||
}
|
||||
try await h.app.server.start(address: .hostname("127.0.0.1", port: 0))
|
||||
let port = try #require(h.app.http.server.shared.localAddress?.port)
|
||||
|
||||
let outcome = try await withThrowingTaskGroup(of: (Int, Int)?.self) { group in
|
||||
group.addTask {
|
||||
let response = try await HTTPClient.shared.execute(
|
||||
HTTPClientRequest(url: "http://127.0.0.1:\(port)/blob"),
|
||||
timeout: .seconds(180))
|
||||
var received = 0
|
||||
var heldWhenFirstBytesArrived = -1
|
||||
for try await piece in response.body {
|
||||
if heldWhenFirstBytesArrived < 0 {
|
||||
heldWhenFirstBytesArrived = produced.value
|
||||
open.yield(())
|
||||
}
|
||||
received += piece.readableBytes
|
||||
}
|
||||
return (received, heldWhenFirstBytesArrived)
|
||||
}
|
||||
group.addTask {
|
||||
try await Task.sleep(for: .seconds(120))
|
||||
return nil
|
||||
}
|
||||
let first = try await group.next()!
|
||||
group.cancelAll()
|
||||
return first
|
||||
}
|
||||
await h.app.server.shutdown()
|
||||
|
||||
let (received, producedBeforeFirstByte) = try #require(
|
||||
outcome, "no bytes reached the client while the bucket was held")
|
||||
#expect(producedBeforeFirstByte <= hold)
|
||||
#expect(received == chunkSize * chunks)
|
||||
#expect(produced.value == chunks)
|
||||
}
|
||||
}
|
||||
|
||||
final class Counter: @unchecked Sendable {
|
||||
private let lock = NSLock()
|
||||
private var n = 0
|
||||
var value: Int { lock.withLock { n } }
|
||||
func increment() { lock.withLock { n += 1 } }
|
||||
}
|
||||
|
||||
/// `count` copies of `chunk`, stopping before index `holdAt` until `gate`
|
||||
/// yields.
|
||||
struct Held: AsyncSequence, Sendable {
|
||||
typealias Element = ByteBuffer
|
||||
let chunk: ByteBuffer
|
||||
let count: Int
|
||||
let holdAt: Int
|
||||
let gate: AsyncStream<Void>
|
||||
let produced: Counter
|
||||
|
||||
struct AsyncIterator: AsyncIteratorProtocol {
|
||||
let source: Held
|
||||
var index = 0
|
||||
mutating func next() async -> ByteBuffer? {
|
||||
guard index < source.count else { return nil }
|
||||
if index == source.holdAt {
|
||||
var waiting = source.gate.makeAsyncIterator()
|
||||
_ = await waiting.next()
|
||||
}
|
||||
index += 1
|
||||
source.produced.increment()
|
||||
return source.chunk
|
||||
}
|
||||
}
|
||||
func makeAsyncIterator() -> AsyncIterator { AsyncIterator(source: self) }
|
||||
}
|
||||
}
|
||||
@@ -169,7 +169,7 @@ struct CapacityTests {
|
||||
endpoint: "https://example.invalid", bucket: "b",
|
||||
region: "r", accessKey: "k", secretKey: "s",
|
||||
prefix: "attachments/"),
|
||||
client: h.app.client,
|
||||
transport: HTTPClientS3Transport(client: h.app.http.client.shared),
|
||||
localFallback: DirectoryAttachmentStore(directory: h.attachmentDir))
|
||||
let check = await StatusChecks.storage(s3)
|
||||
#expect(check.state != .answering)
|
||||
|
||||
@@ -0,0 +1,88 @@
|
||||
import AsyncHTTPClient
|
||||
import Foundation
|
||||
import NIOCore
|
||||
import Vapor
|
||||
|
||||
@testable import FlamenetRelay
|
||||
|
||||
/// An S3 bucket in memory, behind the same transport seam production uses.
|
||||
///
|
||||
/// A PUT is kept only once its whole body has arrived: a body that throws part
|
||||
/// way leaves nothing, which is what a real bucket does with a PUT whose
|
||||
/// connection drops. `status` answers every request with that status instead,
|
||||
/// and `failing` makes every request fail to connect.
|
||||
final class FakeBucket: S3Transport, @unchecked Sendable {
|
||||
struct Sent {
|
||||
let method: HTTPMethod
|
||||
let url: String
|
||||
let headers: HTTPHeaders
|
||||
}
|
||||
struct Unreachable: Error {}
|
||||
|
||||
private let lock = NSLock()
|
||||
private var _objects: [String: Data] = [:]
|
||||
private var _requests: [Sent] = []
|
||||
private let status: HTTPResponseStatus?
|
||||
private let failing: Bool
|
||||
/// Answers a GET in place of the stored object, when it returns non-nil.
|
||||
let onGet: (@Sendable (String) -> HTTPClientResponse?)?
|
||||
|
||||
init(
|
||||
status: HTTPResponseStatus? = nil, failing: Bool = false,
|
||||
onGet: (@Sendable (String) -> HTTPClientResponse?)? = nil
|
||||
) {
|
||||
self.status = status
|
||||
self.failing = failing
|
||||
self.onGet = onGet
|
||||
}
|
||||
|
||||
var requests: [Sent] { lock.withLock { _requests } }
|
||||
var objects: [String: Data] { lock.withLock { _objects } }
|
||||
|
||||
func execute(_ request: HTTPClientRequest) async throws -> HTTPClientResponse {
|
||||
lock.withLock {
|
||||
_requests.append(
|
||||
.init(method: request.method, url: request.url, headers: request.headers))
|
||||
}
|
||||
if failing { throw Unreachable() }
|
||||
if let status { return HTTPClientResponse(status: status) }
|
||||
let key = request.url
|
||||
|
||||
switch request.method {
|
||||
case .PUT:
|
||||
var data = Data()
|
||||
if let body = request.body {
|
||||
for try await chunk in body { data.append(contentsOf: chunk.readableBytesView) }
|
||||
}
|
||||
lock.withLock { _objects[key] = data }
|
||||
return HTTPClientResponse(status: .ok)
|
||||
case .GET:
|
||||
if let answer = onGet?(key) { return answer }
|
||||
guard let data = objects[key] else { return HTTPClientResponse(status: .notFound) }
|
||||
guard let range = request.headers.first(name: "range"),
|
||||
case .partial(let first, let last) = AttachmentRange.decide(
|
||||
header: range, total: data.count)
|
||||
else {
|
||||
return HTTPClientResponse(
|
||||
status: .ok, headers: ["content-length": "\(data.count)"],
|
||||
body: .bytes(ByteBuffer(data: data)))
|
||||
}
|
||||
let slice = data.subdata(in: first..<(last + 1))
|
||||
return HTTPClientResponse(
|
||||
status: .partialContent,
|
||||
headers: [
|
||||
"content-length": "\(slice.count)",
|
||||
"content-range": AttachmentRange.contentRange(
|
||||
first: first, last: last, total: data.count),
|
||||
],
|
||||
body: .bytes(ByteBuffer(data: slice)))
|
||||
case .HEAD:
|
||||
return HTTPClientResponse(status: objects[key] == nil ? .notFound : .ok)
|
||||
case .DELETE:
|
||||
lock.withLock { _objects[key] = nil }
|
||||
return HTTPClientResponse(status: .noContent)
|
||||
default:
|
||||
return HTTPClientResponse(status: .methodNotAllowed)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -40,7 +40,7 @@ extension EndToEndTests {
|
||||
let picture = Data(String(repeating: "p", count: 64 * 1024).utf8)
|
||||
let upload = try await h.routes.uploadAttachment(
|
||||
h.request(.POST, "/e2e/attachments", as: ada, body: picture, now: now),
|
||||
body: picture, now: now)
|
||||
now: now)
|
||||
struct Up: Decodable { let id: String }
|
||||
let token = try JSONDecoder().decode(
|
||||
Up.self, from: Data(text(upload).utf8)
|
||||
|
||||
@@ -146,13 +146,13 @@ struct PushAndS3Tests {
|
||||
|
||||
// MARK: - Object storage
|
||||
|
||||
private func s3(_ client: RecordingClient, fallback: String) -> S3AttachmentStore {
|
||||
private func s3(_ bucket: FakeBucket, fallback: String) -> S3AttachmentStore {
|
||||
S3AttachmentStore(
|
||||
s3: .init(
|
||||
endpoint: "https://s3.example.com", bucket: "blobs",
|
||||
region: "us-west-004", accessKey: "AKIA", secretKey: "secret",
|
||||
prefix: "attachments/"),
|
||||
client: client,
|
||||
transport: bucket,
|
||||
localFallback: DirectoryAttachmentStore(directory: fallback))
|
||||
}
|
||||
|
||||
@@ -163,16 +163,15 @@ struct PushAndS3Tests {
|
||||
@Test("object storage reports reachable when the bucket answers as a bucket")
|
||||
func s3ReachableOn404() async throws {
|
||||
try await withRelay { h in
|
||||
let loop = EmbeddedEventLoop()
|
||||
for status in [HTTPStatus.notFound, .ok, .noContent] {
|
||||
let client = RecordingClient(eventLoop: loop, status: status)
|
||||
let state = await s3(client, fallback: h.attachmentDir).reachability()
|
||||
let bucket = FakeBucket(status: status)
|
||||
let state = await s3(bucket, fallback: h.attachmentDir).reachability()
|
||||
guard case .reachable = state else {
|
||||
Issue.record("\(status.code) should read as reachable, got \(state)")
|
||||
continue
|
||||
}
|
||||
// Signed, and a HEAD: the probe must not read or store a thing.
|
||||
let req = try #require(client.requests.first)
|
||||
let req = try #require(bucket.requests.first)
|
||||
#expect(req.method == .HEAD)
|
||||
#expect(req.headers.first(name: "Authorization")?.contains("AWS4-HMAC-SHA256") == true)
|
||||
}
|
||||
@@ -184,10 +183,9 @@ struct PushAndS3Tests {
|
||||
@Test("a bucket that refuses or does not answer reads as down")
|
||||
func s3UnreachableOnRefusal() async throws {
|
||||
try await withRelay { h in
|
||||
let loop = EmbeddedEventLoop()
|
||||
for status in [HTTPStatus.forbidden, .unauthorized, .internalServerError] {
|
||||
let state = await s3(
|
||||
RecordingClient(eventLoop: loop, status: status),
|
||||
FakeBucket(status: status),
|
||||
fallback: h.attachmentDir
|
||||
).reachability()
|
||||
guard case .unreachable = state else {
|
||||
@@ -196,7 +194,7 @@ struct PushAndS3Tests {
|
||||
}
|
||||
}
|
||||
let dead = await s3(
|
||||
RecordingClient(eventLoop: loop, failing: true),
|
||||
FakeBucket(failing: true),
|
||||
fallback: h.attachmentDir
|
||||
).reachability()
|
||||
guard case .unreachable = dead else {
|
||||
@@ -209,15 +207,14 @@ struct PushAndS3Tests {
|
||||
@Test("a write is a signed PUT to the bucket path")
|
||||
func s3Write() async throws {
|
||||
try await withRelay { h in
|
||||
let loop = EmbeddedEventLoop()
|
||||
let client = RecordingClient(eventLoop: loop)
|
||||
let bucket = FakeBucket()
|
||||
let token = randomHexToken()
|
||||
try await s3(client, fallback: h.attachmentDir)
|
||||
try await s3(bucket, fallback: h.attachmentDir)
|
||||
.write(token, data: Data(repeating: 1, count: 64))
|
||||
|
||||
let req = try #require(client.requests.first)
|
||||
let req = try #require(bucket.requests.first)
|
||||
#expect(req.method == .PUT)
|
||||
#expect(req.url.string == "https://s3.example.com/blobs/attachments/\(token).bin")
|
||||
#expect(req.url == "https://s3.example.com/blobs/attachments/\(token).bin")
|
||||
// SigV4: the request must actually carry a signature and a payload
|
||||
// hash, or the bucket answers 403 and blobs silently never store.
|
||||
#expect(req.headers.first(name: "Authorization")?.contains("AWS4-HMAC-SHA256") == true)
|
||||
@@ -230,10 +227,9 @@ struct PushAndS3Tests {
|
||||
@Test("a refused write is an error, not a silent success")
|
||||
func s3WriteFailure() async throws {
|
||||
try await withRelay { h in
|
||||
let loop = EmbeddedEventLoop()
|
||||
let client = RecordingClient(eventLoop: loop, status: .forbidden)
|
||||
let bucket = FakeBucket(status: .forbidden)
|
||||
await #expect(throws: (any Error).self) {
|
||||
try await s3(client, fallback: h.attachmentDir)
|
||||
try await s3(bucket, fallback: h.attachmentDir)
|
||||
.write(randomHexToken(), data: Data(repeating: 1, count: 8))
|
||||
}
|
||||
}
|
||||
@@ -244,18 +240,17 @@ struct PushAndS3Tests {
|
||||
@Test("a blob missing from the bucket falls back to local disk")
|
||||
func s3FallsBackToDisk() async throws {
|
||||
try await withRelay { h in
|
||||
let loop = EmbeddedEventLoop()
|
||||
let token = randomHexToken()
|
||||
// The pre-switch blob, still on disk.
|
||||
try await DirectoryAttachmentStore(directory: h.attachmentDir)
|
||||
.write(token, data: Data(repeating: 7, count: 32))
|
||||
|
||||
let client = RecordingClient(eventLoop: loop, status: .notFound)
|
||||
let bucket = FakeBucket(status: .notFound)
|
||||
let req = Request(
|
||||
application: h.app, method: .GET,
|
||||
url: URI(path: "/e2e/attachments/\(token)"),
|
||||
on: h.app.eventLoopGroup.next())
|
||||
let response = try await s3(client, fallback: h.attachmentDir)
|
||||
let response = try await s3(bucket, fallback: h.attachmentDir)
|
||||
.respond(token, request: req, range: nil)
|
||||
#expect(response?.status == .ok)
|
||||
}
|
||||
@@ -264,14 +259,13 @@ struct PushAndS3Tests {
|
||||
@Test("a bucket error on read is surfaced rather than treated as absent")
|
||||
func s3ReadFailure() async throws {
|
||||
try await withRelay { h in
|
||||
let loop = EmbeddedEventLoop()
|
||||
let client = RecordingClient(eventLoop: loop, status: .internalServerError)
|
||||
let bucket = FakeBucket(status: .internalServerError)
|
||||
let req = Request(
|
||||
application: h.app, method: .GET,
|
||||
url: URI(path: "/e2e/attachments/x"),
|
||||
on: h.app.eventLoopGroup.next())
|
||||
await #expect(throws: (any Error).self) {
|
||||
_ = try await s3(client, fallback: h.attachmentDir)
|
||||
_ = try await s3(bucket, fallback: h.attachmentDir)
|
||||
.respond(randomHexToken(), request: req, range: nil)
|
||||
}
|
||||
}
|
||||
@@ -283,10 +277,9 @@ struct PushAndS3Tests {
|
||||
@Test("delete accepts both deleted and already-gone")
|
||||
func s3DeleteAcceptsBoth() async throws {
|
||||
try await withRelay { h in
|
||||
let loop = EmbeddedEventLoop()
|
||||
for status in [HTTPStatus.noContent, .ok, .notFound] {
|
||||
let client = RecordingClient(eventLoop: loop, status: status)
|
||||
try await s3(client, fallback: h.attachmentDir).delete(randomHexToken())
|
||||
let bucket = FakeBucket(status: status)
|
||||
try await s3(bucket, fallback: h.attachmentDir).delete(randomHexToken())
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -294,10 +287,9 @@ struct PushAndS3Tests {
|
||||
@Test("a bucket error on delete is surfaced")
|
||||
func s3DeleteFailure() async throws {
|
||||
try await withRelay { h in
|
||||
let loop = EmbeddedEventLoop()
|
||||
let client = RecordingClient(eventLoop: loop, status: .forbidden)
|
||||
let bucket = FakeBucket(status: .forbidden)
|
||||
await #expect(throws: (any Error).self) {
|
||||
try await s3(client, fallback: h.attachmentDir).delete(randomHexToken())
|
||||
try await s3(bucket, fallback: h.attachmentDir).delete(randomHexToken())
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -308,7 +300,6 @@ struct PushAndS3Tests {
|
||||
@Test("deleting removes the local fallback copy as well")
|
||||
func s3DeleteClearsDisk() async throws {
|
||||
try await withRelay { h in
|
||||
let loop = EmbeddedEventLoop()
|
||||
let token = randomHexToken()
|
||||
try await DirectoryAttachmentStore(directory: h.attachmentDir)
|
||||
.write(token, data: Data(repeating: 3, count: 16))
|
||||
@@ -317,7 +308,7 @@ struct PushAndS3Tests {
|
||||
atPath: attachmentPath(h.attachmentDir, token)))
|
||||
|
||||
try await s3(
|
||||
RecordingClient(eventLoop: loop, status: .noContent),
|
||||
FakeBucket(status: .noContent),
|
||||
fallback: h.attachmentDir
|
||||
).delete(token)
|
||||
#expect(
|
||||
|
||||
Reference in New Issue
Block a user