From 2276fee60c0f58e1d66aeaae27b118d056c5080f Mon Sep 17 00:00:00 2001 From: LunarSkyOSS Date: Sat, 10 Oct 2026 07:51:38 +0200 Subject: [PATCH] Refactor complex ObjectStore methods --- README.md | 2 +- src/cloud/lunarsky/store/ClusterNode.java | 76 +++++--- src/cloud/lunarsky/store/ClusterStore.java | 214 ++++++++++++--------- src/cloud/lunarsky/store/DiskStore.java | 125 ++++++------ src/cloud/lunarsky/store/Main.java | 213 +++++++++++--------- src/cloud/lunarsky/store/NodeRegistry.java | 118 +++++++----- src/cloud/lunarsky/store/SigV4.java | 52 +++-- test/cloud/lunarsky/store/HttpTest.java | 207 +++++++++++--------- test/cloud/lunarsky/store/StoreTest.java | 186 ++++++++++-------- 9 files changed, 693 insertions(+), 500 deletions(-) diff --git a/README.md b/README.md index 285fb4b..4884f2f 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ Source: [GitHub](https://github.com/LunarSkyOSS/ObjectStore) · [Gitea mirror](h **Development software. Do not use it for production data.** It is provided as is, without warranty, under the [MIT License](LICENSE). -[![Stage: development](assets/badge-stage.svg)](#limits-and-safety) [![License: MIT](assets/badge-license.svg)](LICENSE) [![Runtime: Java 21](assets/badge-java.svg)](Dockerfile) [![S3 API: partial support](assets/badge-api.svg)](#s3-api-support-checklist) +[![Stage: development](assets/badge-stage.svg)](#limits-and-safety) [![License: MIT](assets/badge-license.svg)](LICENSE) [![Runtime: Java 21](assets/badge-java.svg)](Dockerfile) [![S3 API: partial support](assets/badge-api.svg)](#s3-api-support-checklist) [![CodeFactor](https://www.codefactor.io/repository/github/lunarskyoss/objectstore/badge)](https://www.codefactor.io/repository/github/lunarskyoss/objectstore) ## Contents diff --git a/src/cloud/lunarsky/store/ClusterNode.java b/src/cloud/lunarsky/store/ClusterNode.java index 2b2b90a..22d67be 100644 --- a/src/cloud/lunarsky/store/ClusterNode.java +++ b/src/cloud/lunarsky/store/ClusterNode.java @@ -78,40 +78,14 @@ public final class ClusterNode implements AutoCloseable { respond(exchange, 200, "ok"); return; } - byte[] supplied = exchange.getRequestHeaders().getFirst("X-Cluster-Token") == null - ? new byte[0] : exchange.getRequestHeaders().getFirst("X-Cluster-Token") - .getBytes(java.nio.charset.StandardCharsets.UTF_8); - if (!MessageDigest.isEqual(token, supplied)) { - respond(exchange, 403, "Forbidden"); - return; - } + if (!authorized(exchange)) return; if (path.equals("/identity") && exchange.getRequestMethod().equals("GET")) { respond(exchange, 200, identity.nodeId() + " " + identity.hostId()); return; } - if (!identity.nodeId().toString().equals(exchange.getRequestHeaders().getFirst("X-Cluster-Expected-Node"))) { - respond(exchange, 409, "Wrong storage node"); - return; - } - if (exchange.getRequestMethod().equals("PUT") && - "true".equals(exchange.getRequestHeaders().getFirst("X-Cluster-Repair"))) { - String suppliedRepair = exchange.getRequestHeaders().getFirst("X-Cluster-Repair-Token"); - byte[] suppliedBytes = suppliedRepair == null ? new byte[0] - : suppliedRepair.getBytes(java.nio.charset.StandardCharsets.UTF_8); - if (!MessageDigest.isEqual(repairToken, suppliedBytes)) { - respond(exchange, 403, "Repair authority required"); - return; - } - } - if (!path.matches("/segments/[0-9a-f-]{36}")) { - respond(exchange, 404, "Not found"); - return; - } - String id = path.substring("/segments/".length()); - if (!UUID.fromString(id).toString().equals(id)) { - respond(exchange, 400, "Invalid segment ID"); - return; - } + if (!expectedNodeAndRepairAuthorized(exchange)) return; + String id = segmentId(exchange, path); + if (id == null) return; switch (exchange.getRequestMethod()) { case "PUT" -> put(exchange, segmentPath(id, true)); case "GET" -> get(exchange, segmentPath(id, false)); @@ -127,6 +101,48 @@ public final class ClusterNode implements AutoCloseable { } } + private boolean authorized(HttpExchange exchange) throws IOException { + byte[] supplied = exchange.getRequestHeaders().getFirst("X-Cluster-Token") == null + ? new byte[0] : exchange.getRequestHeaders().getFirst("X-Cluster-Token") + .getBytes(java.nio.charset.StandardCharsets.UTF_8); + if (!MessageDigest.isEqual(token, supplied)) { + respond(exchange, 403, "Forbidden"); + return false; + } + return true; + } + + private boolean expectedNodeAndRepairAuthorized(HttpExchange exchange) throws IOException { + if (!identity.nodeId().toString().equals(exchange.getRequestHeaders().getFirst("X-Cluster-Expected-Node"))) { + respond(exchange, 409, "Wrong storage node"); + return false; + } + if (exchange.getRequestMethod().equals("PUT") && + "true".equals(exchange.getRequestHeaders().getFirst("X-Cluster-Repair"))) { + String suppliedRepair = exchange.getRequestHeaders().getFirst("X-Cluster-Repair-Token"); + byte[] suppliedBytes = suppliedRepair == null ? new byte[0] + : suppliedRepair.getBytes(java.nio.charset.StandardCharsets.UTF_8); + if (!MessageDigest.isEqual(repairToken, suppliedBytes)) { + respond(exchange, 403, "Repair authority required"); + return false; + } + } + return true; + } + + private static String segmentId(HttpExchange exchange, String path) throws IOException { + if (!path.matches("/segments/[0-9a-f-]{36}")) { + respond(exchange, 404, "Not found"); + return null; + } + String id = path.substring("/segments/".length()); + if (!UUID.fromString(id).toString().equals(id)) { + respond(exchange, 400, "Invalid segment ID"); + return null; + } + return id; + } + private synchronized Path segmentPath(String id, boolean createShard) throws IOException { Path shard = segments.resolve(id.substring(0, 2)); if (createShard && !Files.isDirectory(shard)) { diff --git a/src/cloud/lunarsky/store/ClusterStore.java b/src/cloud/lunarsky/store/ClusterStore.java index 843a241..be5130f 100644 --- a/src/cloud/lunarsky/store/ClusterStore.java +++ b/src/cloud/lunarsky/store/ClusterStore.java @@ -52,6 +52,23 @@ final class ClusterStore implements ObjectStorage { @Override public Metadata put(String bucket, String key, InputStream input, long length, String expectedHash, String checksum, boolean createOnly, String contentType) throws IOException { + validatePut(bucket, length, contentType); + MessageDigest md5 = digest("MD5"); + Path staged = Files.createTempFile("objectstore-cluster-", ".pending"); + List segments; + byte[] fullHash; + try { + fullHash = stageInput(staged, input, length, expectedHash, checksum, md5); + checkCapacity(bucket, key, length, createOnly); + segments = uploadSegments(staged, length); + } finally { Files.deleteIfExists(staged); } + Metadata metadata = new Metadata(length, Instant.now().toEpochMilli(), + HexFormat.of().formatHex(md5.digest()), fullHash, bucket, key, contentType); + persistObject(metadata, segments, createOnly); + return metadata; + } + + private void validatePut(String bucket, long length, String contentType) { if (!configuredBucket.equals(bucket)) throw new StoreException(404, "NoSuchBucket", "Bucket not found"); if (length < 0) throw new StoreException(411, "MissingContentLength", "Content-Length is required"); if (length > maxObject) throw new StoreException(413, "EntityTooLarge", "Object exceeds the configured size limit"); @@ -59,74 +76,85 @@ final class ClusterStore implements ObjectStorage { throw new StoreException(503, "SlowDown", "Fewer than two storage hosts are available"); if (contentType.getBytes(java.nio.charset.StandardCharsets.UTF_8).length > 255) throw new StoreException(400, "InvalidArgument", "Content-Type is too long"); - MessageDigest sha = digest("SHA-256"), md5 = digest("MD5"); + } + + private byte[] stageInput(Path staged, InputStream input, long length, String expectedHash, + String checksum, MessageDigest md5) throws IOException { + MessageDigest sha = digest("SHA-256"); + try (OutputStream output = Files.newOutputStream(staged)) { + byte[] buffer = new byte[65536]; + long remaining = length; + while (remaining > 0) { + int count = input.read(buffer, 0, (int) Math.min(buffer.length, remaining)); + if (count < 0) throw new StoreException(400, "IncompleteBody", "Payload length does not match Content-Length"); + if (count == 0) continue; + sha.update(buffer, 0, count); md5.update(buffer, 0, count); + output.write(buffer, 0, count); + remaining -= count; + } + } + if (input.read() != -1) throw new StoreException(413, "EntityTooLarge", "Payload exceeds declared size"); + byte[] fullHash = sha.digest(); + if (!HexFormat.of().formatHex(fullHash).equals(expectedHash)) + throw new StoreException(400, "XAmzContentSHA256Mismatch", "Payload hash mismatch"); + if (checksum != null && !Base64.getEncoder().encodeToString(fullHash).equals(checksum)) + throw new StoreException(400, "BadDigest", "SHA-256 checksum mismatch"); + return fullHash; + } + + private void checkCapacity(String bucket, String key, long length, boolean createOnly) throws IOException { + try (Connection connection = connect()) { + long previous = currentLength(connection, bucket, key); + if (createOnly && previous >= 0) + throw new StoreException(412, "PreconditionFailed", "Object already exists"); + try (PreparedStatement query = connection.prepareStatement("SELECT used_bytes FROM cluster_usage WHERE bucket=?")) { + query.setString(1, bucket); + try (ResultSet result = query.executeQuery()) { + if (!result.next()) throw new SQLException("Bucket quota row is missing"); + if (result.getLong(1) - Math.max(0, previous) > maxTotal - length) + throw new StoreException(507, "InsufficientStorage", "Store capacity limit reached"); + } + } + } catch (SQLException error) { throw databaseError(error); } + } + + private List uploadSegments(Path staged, long length) throws IOException { List segments = new ArrayList<>(); - byte[] fullHash; - Path staged = Files.createTempFile("objectstore-cluster-", ".pending"); - try { - try (OutputStream output = Files.newOutputStream(staged)) { - byte[] buffer = new byte[65536]; - long remaining = length; - while (remaining > 0) { - int count = input.read(buffer, 0, (int) Math.min(buffer.length, remaining)); - if (count < 0) throw new StoreException(400, "IncompleteBody", "Payload length does not match Content-Length"); - if (count == 0) continue; - sha.update(buffer, 0, count); md5.update(buffer, 0, count); - output.write(buffer, 0, count); - remaining -= count; - } - } - if (input.read() != -1) throw new StoreException(413, "EntityTooLarge", "Payload exceeds declared size"); - fullHash = sha.digest(); - if (!HexFormat.of().formatHex(fullHash).equals(expectedHash)) - throw new StoreException(400, "XAmzContentSHA256Mismatch", "Payload hash mismatch"); - if (checksum != null && !Base64.getEncoder().encodeToString(fullHash).equals(checksum)) - throw new StoreException(400, "BadDigest", "SHA-256 checksum mismatch"); - try (Connection connection = connect()) { - long previous = currentLength(connection, bucket, key); - if (createOnly && previous >= 0) - throw new StoreException(412, "PreconditionFailed", "Object already exists"); - try (PreparedStatement query = connection.prepareStatement("SELECT used_bytes FROM cluster_usage WHERE bucket=?")) { - query.setString(1, bucket); - try (ResultSet result = query.executeQuery()) { - if (!result.next()) throw new SQLException("Bucket quota row is missing"); - if (result.getLong(1) - Math.max(0, previous) > maxTotal - length) - throw new StoreException(507, "InsufficientStorage", "Store capacity limit reached"); + try (InputStream stagedInput = Files.newInputStream(staged)) { + long remaining = length; + while (remaining > 0) { + int wanted = (int) Math.min(ClusterNode.MAX_SEGMENT, remaining); + byte[] bytes = stagedInput.readNBytes(wanted); + if (bytes.length != wanted) throw new IOException("Staged object was truncated"); + byte[] segmentHash = SigV4.hash(bytes); + UUID id = UUID.randomUUID(); + List replicas = new ArrayList<>(); + Set acceptedHosts = new HashSet<>(); + for (int index : PlacementPolicy.candidates(id, nodes, testNodeDomains)) { + UUID host = nodes.faultDomain(index, testNodeDomains); + if (acceptedHosts.contains(host)) continue; + try { + nodes.put(index, id, bytes, segmentHash); + replicas.add(nodes.node(index).id()); + acceptedHosts.add(host); + if (acceptedHosts.size() == 3) break; + } catch (IOException error) { + System.err.println("Cluster node " + nodes.node(index).id() + + " did not accept segment " + id + ": " + error.getMessage()); } } - } catch (SQLException error) { throw databaseError(error); } - try (InputStream stagedInput = Files.newInputStream(staged)) { - long remaining = length; - while (remaining > 0) { - int wanted = (int) Math.min(ClusterNode.MAX_SEGMENT, remaining); - byte[] bytes = stagedInput.readNBytes(wanted); - if (bytes.length != wanted) throw new IOException("Staged object was truncated"); - byte[] segmentHash = SigV4.hash(bytes); - UUID id = UUID.randomUUID(); - List replicas = new ArrayList<>(); - Set acceptedHosts = new HashSet<>(); - for (int index : PlacementPolicy.candidates(id, nodes, testNodeDomains)) { - UUID host = nodes.faultDomain(index, testNodeDomains); - if (acceptedHosts.contains(host)) continue; - try { - nodes.put(index, id, bytes, segmentHash); - replicas.add(nodes.node(index).id()); - acceptedHosts.add(host); - if (acceptedHosts.size() == 3) break; - } catch (IOException error) { - System.err.println("Cluster node " + nodes.node(index).id() + - " did not accept segment " + id + ": " + error.getMessage()); - } - } - if (acceptedHosts.size() < 2) - throw new StoreException(503, "SlowDown", "Fewer than two storage hosts accepted the segment"); - segments.add(new Segment(id, wanted, segmentHash, List.copyOf(replicas))); - remaining -= wanted; - } + if (acceptedHosts.size() < 2) + throw new StoreException(503, "SlowDown", "Fewer than two storage hosts accepted the segment"); + segments.add(new Segment(id, wanted, segmentHash, List.copyOf(replicas))); + remaining -= wanted; } - } finally { Files.deleteIfExists(staged); } - Metadata metadata = new Metadata(length, Instant.now().toEpochMilli(), - HexFormat.of().formatHex(md5.digest()), fullHash, bucket, key, contentType); + } + return segments; + } + + private void persistObject(Metadata metadata, List segments, boolean createOnly) throws IOException { + String bucket = metadata.bucket(), key = metadata.key(); + long length = metadata.length(); UUID generation = UUID.randomUUID(); try (Connection connection = connect()) { connection.setAutoCommit(false); @@ -159,7 +187,6 @@ final class ClusterStore implements ObjectStorage { delete.setString(1, bucket); delete.setString(2, key); delete.executeUpdate(); } connection.commit(); - return metadata; } catch (SQLException | RuntimeException error) { connection.rollback(); if (error instanceof SQLException sql) throw databaseError(sql); @@ -241,45 +268,54 @@ final class ClusterStore implements ObjectStorage { } @Override public ListPage list(String bucket, String prefix, String delimiter, int maxKeys, String after) throws IOException { - List entries = new ArrayList<>(); - List prefixes = new ArrayList<>(); - if (maxKeys == 0) return new ListPage(entries, prefixes, null, false); - String lastKey = null, activePrefix = null; - boolean truncated = false; + if (maxKeys == 0) return new ListPage(new ArrayList<>(), new ArrayList<>(), null, false); try (Connection connection = connect()) { connection.setAutoCommit(false); + ListPage page; try (PreparedStatement query = connection.prepareStatement( "SELECT object_key, length, modified, etag, sha256, content_type FROM cluster_objects WHERE bucket=? AND object_key>=? ORDER BY object_key")) { query.setString(1, bucket); query.setString(2, after != null && after.compareTo(prefix) > 0 ? after : prefix); query.setFetchSize(128); try (ResultSet result = query.executeQuery()) { - while (result.next()) { - String key = result.getString(1); - if (!key.startsWith(prefix)) break; - if (after != null && key.compareTo(after) <= 0) continue; - String group = null; - if (!delimiter.isEmpty()) { - int at = key.indexOf(delimiter, prefix.length()); - if (at >= 0) group = key.substring(0, at + delimiter.length()); - } - if (group != null && group.equals(activePrefix)) { lastKey = key; continue; } - if (entries.size() + prefixes.size() >= maxKeys) { truncated = true; break; } - if (group != null) { prefixes.add(group); activePrefix = group; } - else { - entries.add(new ListedObject(key, new Metadata(result.getLong(2), result.getLong(3), - result.getString(4), result.getBytes(5), bucket, key, result.getString(6)))); - activePrefix = null; - } - lastKey = key; - } + page = readListPage(result, bucket, prefix, delimiter, maxKeys, after); } } connection.commit(); + return page; } catch (SQLException error) { throw databaseError(error); } + } + + private static ListPage readListPage(ResultSet result, String bucket, String prefix, String delimiter, + int maxKeys, String after) throws SQLException { + List entries = new ArrayList<>(); + List prefixes = new ArrayList<>(); + String lastKey = null, activePrefix = null; + boolean truncated = false; + while (result.next()) { + String key = result.getString(1); + if (!key.startsWith(prefix)) break; + if (after != null && key.compareTo(after) <= 0) continue; + String group = commonPrefix(key, prefix, delimiter); + if (group != null && group.equals(activePrefix)) { lastKey = key; continue; } + if (entries.size() + prefixes.size() >= maxKeys) { truncated = true; break; } + if (group != null) { prefixes.add(group); activePrefix = group; } + else { + entries.add(new ListedObject(key, new Metadata(result.getLong(2), result.getLong(3), + result.getString(4), result.getBytes(5), bucket, key, result.getString(6)))); + activePrefix = null; + } + lastKey = key; + } return new ListPage(entries, prefixes, truncated ? lastKey : null, truncated); } + private static String commonPrefix(String key, String prefix, String delimiter) { + if (delimiter.isEmpty()) return null; + int at = key.indexOf(delimiter, prefix.length()); + return at < 0 ? null : key.substring(0, at + delimiter.length()); + } + private long lockUsage(Connection connection, String bucket) throws SQLException { try (PreparedStatement query = connection.prepareStatement("SELECT used_bytes FROM cluster_usage WHERE bucket=? FOR UPDATE")) { query.setString(1, bucket); diff --git a/src/cloud/lunarsky/store/DiskStore.java b/src/cloud/lunarsky/store/DiskStore.java index fd7d7f3..d5bf450 100644 --- a/src/cloud/lunarsky/store/DiskStore.java +++ b/src/cloud/lunarsky/store/DiskStore.java @@ -117,67 +117,82 @@ final class DiskStore implements ObjectStorage { byte[] bucketBytes = bucket.getBytes(StandardCharsets.UTF_8); byte[] keyBytes = key.getBytes(StandardCharsets.UTF_8); byte[] typeBytes = contentType.getBytes(StandardCharsets.UTF_8); - if (bucketBytes.length > 63 || keyBytes.length > 1024 || typeBytes.length > 255) - throw new StoreException(400, "InvalidArgument", "Object metadata is too long"); - int headerLength = HEADER_V2 + bucketBytes.length + keyBytes.length + typeBytes.length; + validateMetadataLengths(bucketBytes, keyBytes, typeBytes); Path destination = object(bucket, key), pending = Files.createTempFile(temporary, "upload-", ".part"); try { - MessageDigest sha = digest("SHA-256"), md5 = digest("MD5"); - long count = 0; - try (OutputStream out = Files.newOutputStream(pending)) { - out.write(new byte[headerLength]); - byte[] buffer = new byte[65536]; int n; - while ((n = input.read(buffer)) != -1) { - count += n; - if (count > length || count > maxObject) - throw new StoreException(413, "EntityTooLarge", "Payload exceeds declared size"); - sha.update(buffer, 0, n); md5.update(buffer, 0, n); out.write(buffer, 0, n); - } - } - if (count != length) throw new StoreException(400, "IncompleteBody", "Payload length does not match Content-Length"); - byte[] hash = sha.digest(), etag = md5.digest(); - if (!MessageDigest.isEqual(hash, HexFormat.of().parseHex(expectedHash))) - throw new StoreException(400, "XAmzContentSHA256Mismatch", "Payload hash mismatch"); - if (checksum != null && !Base64.getEncoder().encodeToString(hash).equals(checksum)) - throw new StoreException(400, "BadDigest", "SHA-256 checksum mismatch"); - long modified = Instant.now().toEpochMilli(); - ByteBuffer header = ByteBuffer.allocate(headerLength).putLong(MAGIC_V2).putLong(count) - .putLong(modified).put(etag).put(hash).putShort((short) bucketBytes.length) - .putShort((short) keyBytes.length).putShort((short) typeBytes.length) - .put(bucketBytes).put(keyBytes).put(typeBytes); - header.flip(); - try (FileChannel file = FileChannel.open(pending, StandardOpenOption.WRITE)) { - while (header.hasRemaining()) file.write(header, header.position()); - file.force(true); - } - Metadata metadata = new Metadata(count, modified, SigV4.hex(etag), hash, bucket, key, contentType); - synchronized (lock(destination)) { - long previous = 0; - boolean existed = Files.exists(destination); - boolean legacy = false; - if (existed) { - if (createOnly) throw new StoreException(412, "PreconditionFailed", "Object already exists"); - try (var in = new DataInputStream(Files.newInputStream(destination))) { - Metadata old = readRecord(in).metadata(); - previous = old.length(); - legacy = old.key() == null; - } - } - synchronized (this) { - if (used - previous + count > maxTotal) - throw new StoreException(507, "InsufficientStorage", "Store capacity limit reached"); - Files.move(pending, destination, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING); - used = used - previous + count; - if (!existed) objectCount++; - if (legacy) legacyCount--; - index.put(indexKey(bucket, key), metadata); - syncDirectory(destination.getParent()); - } - } + Metadata metadata = stagePut(pending, input, length, expectedHash, checksum, + bucket, key, contentType, bucketBytes, keyBytes, typeBytes); + installPending(destination, pending, metadata, createOnly); return metadata; } finally { Files.deleteIfExists(pending); } } + private static void validateMetadataLengths(byte[] bucketBytes, byte[] keyBytes, byte[] typeBytes) { + if (bucketBytes.length > 63 || keyBytes.length > 1024 || typeBytes.length > 255) + throw new StoreException(400, "InvalidArgument", "Object metadata is too long"); + } + + private Metadata stagePut(Path pending, InputStream input, long length, String expectedHash, String checksum, + String bucket, String key, String contentType, + byte[] bucketBytes, byte[] keyBytes, byte[] typeBytes) throws IOException { + int headerLength = HEADER_V2 + bucketBytes.length + keyBytes.length + typeBytes.length; + MessageDigest sha = digest("SHA-256"), md5 = digest("MD5"); + long count = 0; + try (OutputStream out = Files.newOutputStream(pending)) { + out.write(new byte[headerLength]); + byte[] buffer = new byte[65536]; int n; + while ((n = input.read(buffer)) != -1) { + count += n; + if (count > length || count > maxObject) + throw new StoreException(413, "EntityTooLarge", "Payload exceeds declared size"); + sha.update(buffer, 0, n); md5.update(buffer, 0, n); out.write(buffer, 0, n); + } + } + if (count != length) throw new StoreException(400, "IncompleteBody", "Payload length does not match Content-Length"); + byte[] hash = sha.digest(), etag = md5.digest(); + if (!MessageDigest.isEqual(hash, HexFormat.of().parseHex(expectedHash))) + throw new StoreException(400, "XAmzContentSHA256Mismatch", "Payload hash mismatch"); + if (checksum != null && !Base64.getEncoder().encodeToString(hash).equals(checksum)) + throw new StoreException(400, "BadDigest", "SHA-256 checksum mismatch"); + long modified = Instant.now().toEpochMilli(); + ByteBuffer header = ByteBuffer.allocate(headerLength).putLong(MAGIC_V2).putLong(count) + .putLong(modified).put(etag).put(hash).putShort((short) bucketBytes.length) + .putShort((short) keyBytes.length).putShort((short) typeBytes.length) + .put(bucketBytes).put(keyBytes).put(typeBytes); + header.flip(); + try (FileChannel file = FileChannel.open(pending, StandardOpenOption.WRITE)) { + while (header.hasRemaining()) file.write(header, header.position()); + file.force(true); + } + return new Metadata(count, modified, SigV4.hex(etag), hash, bucket, key, contentType); + } + + private void installPending(Path destination, Path pending, Metadata metadata, boolean createOnly) throws IOException { + synchronized (lock(destination)) { + long previous = 0; + boolean existed = Files.exists(destination); + boolean legacy = false; + if (existed) { + if (createOnly) throw new StoreException(412, "PreconditionFailed", "Object already exists"); + try (var in = new DataInputStream(Files.newInputStream(destination))) { + Metadata old = readRecord(in).metadata(); + previous = old.length(); + legacy = old.key() == null; + } + } + synchronized (this) { + if (used - previous + metadata.length() > maxTotal) + throw new StoreException(507, "InsufficientStorage", "Store capacity limit reached"); + Files.move(pending, destination, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING); + used = used - previous + metadata.length(); + if (!existed) objectCount++; + if (legacy) legacyCount--; + index.put(indexKey(metadata.bucket(), metadata.key()), metadata); + syncDirectory(destination.getParent()); + } + } + } + public OpenObject open(String bucket, String key) throws IOException { Path destination = object(bucket, key); synchronized (lock(destination)) { diff --git a/src/cloud/lunarsky/store/Main.java b/src/cloud/lunarsky/store/Main.java index 38e3f00..867798f 100644 --- a/src/cloud/lunarsky/store/Main.java +++ b/src/cloud/lunarsky/store/Main.java @@ -45,87 +45,14 @@ public final class Main { exchange.getResponseHeaders().set("X-Content-Type-Options", "nosniff"); try { if (!admitted) throw new StoreException(503, "SlowDown", "Too many concurrent requests"); - if (exchange.getRequestURI().getRawPath().equals("/health") && exchange.getRequestMethod().equals("GET")) { - byte[] body = "{\"status\":\"ok\",\"service\":\"lunarsky-objectstore\"}".getBytes(StandardCharsets.UTF_8); - exchange.getResponseHeaders().set("Content-Type", "application/json"); - exchange.sendResponseHeaders(200, body.length); - exchange.getResponseBody().write(body); - return; - } - if (exchange.getRequestURI().getRawPath().equals("/ready") && exchange.getRequestMethod().equals("GET")) { - boolean ready = store.ready(); - byte[] body = (ready ? "ready" : "unavailable").getBytes(StandardCharsets.UTF_8); - exchange.getResponseHeaders().set("Content-Type", "text/plain; charset=utf-8"); - exchange.sendResponseHeaders(ready ? 200 : 503, body.length); - exchange.getResponseBody().write(body); - return; - } + if (handleStatus(exchange)) return; String hash = authentication.verify(exchange.getRequestMethod(), exchange.getRequestURI(), exchange.getRequestHeaders()); String path = SigV4.decode(exchange.getRequestURI().getRawPath()); Map query = query(exchange.getRequestURI().getRawQuery()); if (path.equals("/" + bucket) || path.equals("/" + bucket + "/")) { - if (!exchange.getRequestMethod().equals("GET") || !"2".equals(query.get("list-type")) || - !query.keySet().stream().allMatch(java.util.Set.of("list-type", "prefix", "delimiter", "max-keys", - "continuation-token", "start-after", "encoding-type", "x-id")::contains) || - (query.containsKey("x-id") && !"ListObjectsV2".equals(query.get("x-id")))) - unsupported("Bucket operation"); - requireEmptyBody(exchange, hash); - listObjects(exchange, query); - return; - } - String prefix = "/" + bucket + "/"; - if (!path.startsWith(prefix)) throw new StoreException(404, "NoSuchBucket", "Bucket not found"); - String key = path.substring(prefix.length()); - if (key.isEmpty() || key.getBytes(StandardCharsets.UTF_8).length > 1024 || key.indexOf('\0') >= 0) - throw new StoreException(400, "InvalidArgument", "Invalid object key"); - String method = exchange.getRequestMethod(); - boolean multipartRequest = multipartRequest(method, query); - if (!multipartRequest && !query.isEmpty() && !(query.size() == 1 && - ("PutObject".equals(query.get("x-id")) || "GetObject".equals(query.get("x-id")) || - "HeadObject".equals(query.get("x-id")) || "DeleteObject".equals(query.get("x-id"))))) - unsupported("Query operation"); - var headers = exchange.getRequestHeaders(); - for (String name : headers.keySet()) { - String lower = name.toLowerCase(java.util.Locale.ROOT); - if (lower.startsWith("x-amz-") && !java.util.Set.of("x-amz-date", "x-amz-content-sha256", - "x-amz-checksum-sha256", "x-amz-sdk-checksum-algorithm", "x-amz-user-agent").contains(lower)) - unsupported("Amazon header"); - if (lower.startsWith("x-amz-meta-") || lower.startsWith("x-amz-server-side-") || - lower.startsWith("x-amz-copy-") || lower.startsWith("x-amz-acl") || - lower.startsWith("x-amz-grant") || lower.startsWith("x-amz-tagging") || - lower.equals("content-md5")) unsupported("Object metadata, encryption, ACL, copy, tagging or MD5 header"); - if (lower.startsWith("x-amz-checksum-") && !lower.equals("x-amz-checksum-sha256")) - unsupported("Checksum algorithm"); - } - String algorithm = SigV4.single(headers, "x-amz-sdk-checksum-algorithm"); - if (algorithm != null && !algorithm.equals("SHA256")) unsupported("Checksum algorithm"); - if (multipartRequest) { - handleMultipart(exchange, method, query, key, hash); - return; - } - if (!method.equals("PUT")) requireEmptyBody(exchange, hash); - switch (method) { - case "PUT" -> { - String length = SigV4.single(headers, "content-length"), condition = SigV4.single(headers, "if-none-match"); - if (condition != null && !condition.equals("*")) unsupported("Write condition"); - long bytes; - try { bytes = length == null ? -1 : Long.parseLong(length); } - catch (NumberFormatException e) { throw new StoreException(400, "InvalidArgument", "Invalid Content-Length"); } - if (headers.containsKey("content-encoding")) unsupported("Encoded payload"); - String contentType = contentType(headers); - ObjectStorage.Metadata data = store.put(bucket, key, exchange.getRequestBody(), bytes, hash, - SigV4.single(headers, "x-amz-checksum-sha256"), condition != null, contentType); - exchange.getResponseHeaders().set("ETag", "\"" + data.etag() + "\""); - exchange.getResponseHeaders().set("x-amz-checksum-sha256", Base64.getEncoder().encodeToString(data.sha256())); - exchange.sendResponseHeaders(200, -1); - } - case "GET", "HEAD" -> readObject(exchange, key); - case "DELETE" -> { - if (headers.containsKey("if-none-match")) unsupported("Conditional delete"); - store.delete(bucket, key); - exchange.sendResponseHeaders(204, -1); - } - default -> unsupported("HTTP method"); + handleBucket(exchange, query, hash); + } else { + handleObject(exchange, path, query, hash); } } catch (StoreException error) { sendError(exchange, error.status, error.code, error.getMessage(), requestId); } catch (Exception error) { @@ -134,6 +61,100 @@ public final class Main { } finally { if (admitted) slots.release(); exchange.close(); } } + private boolean handleStatus(HttpExchange exchange) throws IOException { + if (!exchange.getRequestMethod().equals("GET")) return false; + String path = exchange.getRequestURI().getRawPath(); + if (path.equals("/health")) { + byte[] body = "{\"status\":\"ok\",\"service\":\"lunarsky-objectstore\"}".getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().set("Content-Type", "application/json"); + exchange.sendResponseHeaders(200, body.length); + exchange.getResponseBody().write(body); + return true; + } + if (!path.equals("/ready")) return false; + boolean ready = store.ready(); + byte[] body = (ready ? "ready" : "unavailable").getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().set("Content-Type", "text/plain; charset=utf-8"); + exchange.sendResponseHeaders(ready ? 200 : 503, body.length); + exchange.getResponseBody().write(body); + return true; + } + + private void handleBucket(HttpExchange exchange, Map query, String hash) throws IOException { + if (!exchange.getRequestMethod().equals("GET") || !"2".equals(query.get("list-type")) || + !query.keySet().stream().allMatch(java.util.Set.of("list-type", "prefix", "delimiter", "max-keys", + "continuation-token", "start-after", "encoding-type", "x-id")::contains) || + (query.containsKey("x-id") && !"ListObjectsV2".equals(query.get("x-id")))) + unsupported("Bucket operation"); + requireEmptyBody(exchange, hash); + listObjects(exchange, query); + } + + private void handleObject(HttpExchange exchange, String path, Map query, String hash) throws IOException { + String prefix = "/" + bucket + "/"; + if (!path.startsWith(prefix)) throw new StoreException(404, "NoSuchBucket", "Bucket not found"); + String key = path.substring(prefix.length()); + if (key.isEmpty() || key.getBytes(StandardCharsets.UTF_8).length > 1024 || key.indexOf('\0') >= 0) + throw new StoreException(400, "InvalidArgument", "Invalid object key"); + String method = exchange.getRequestMethod(); + boolean multipartRequest = multipartRequest(method, query); + if (!multipartRequest && !query.isEmpty() && !(query.size() == 1 && + ("PutObject".equals(query.get("x-id")) || "GetObject".equals(query.get("x-id")) || + "HeadObject".equals(query.get("x-id")) || "DeleteObject".equals(query.get("x-id"))))) + unsupported("Query operation"); + validateObjectHeaders(exchange.getRequestHeaders()); + if (multipartRequest) { + handleMultipart(exchange, method, query, key, hash); + return; + } + if (!method.equals("PUT")) requireEmptyBody(exchange, hash); + switch (method) { + case "PUT" -> putObject(exchange, key, hash); + case "GET", "HEAD" -> readObject(exchange, key); + case "DELETE" -> deleteObject(exchange, key); + default -> unsupported("HTTP method"); + } + } + + private static void validateObjectHeaders(com.sun.net.httpserver.Headers headers) { + for (String name : headers.keySet()) { + String lower = name.toLowerCase(java.util.Locale.ROOT); + if (lower.startsWith("x-amz-") && !java.util.Set.of("x-amz-date", "x-amz-content-sha256", + "x-amz-checksum-sha256", "x-amz-sdk-checksum-algorithm", "x-amz-user-agent").contains(lower)) + unsupported("Amazon header"); + if (lower.startsWith("x-amz-meta-") || lower.startsWith("x-amz-server-side-") || + lower.startsWith("x-amz-copy-") || lower.startsWith("x-amz-acl") || + lower.startsWith("x-amz-grant") || lower.startsWith("x-amz-tagging") || + lower.equals("content-md5")) unsupported("Object metadata, encryption, ACL, copy, tagging or MD5 header"); + if (lower.startsWith("x-amz-checksum-") && !lower.equals("x-amz-checksum-sha256")) + unsupported("Checksum algorithm"); + } + String algorithm = SigV4.single(headers, "x-amz-sdk-checksum-algorithm"); + if (algorithm != null && !algorithm.equals("SHA256")) unsupported("Checksum algorithm"); + } + + private void putObject(HttpExchange exchange, String key, String hash) throws IOException { + var headers = exchange.getRequestHeaders(); + String length = SigV4.single(headers, "content-length"), condition = SigV4.single(headers, "if-none-match"); + if (condition != null && !condition.equals("*")) unsupported("Write condition"); + long bytes; + try { bytes = length == null ? -1 : Long.parseLong(length); } + catch (NumberFormatException e) { throw new StoreException(400, "InvalidArgument", "Invalid Content-Length"); } + if (headers.containsKey("content-encoding")) unsupported("Encoded payload"); + String type = contentType(headers); + ObjectStorage.Metadata data = store.put(bucket, key, exchange.getRequestBody(), bytes, hash, + SigV4.single(headers, "x-amz-checksum-sha256"), condition != null, type); + exchange.getResponseHeaders().set("ETag", "\"" + data.etag() + "\""); + exchange.getResponseHeaders().set("x-amz-checksum-sha256", Base64.getEncoder().encodeToString(data.sha256())); + exchange.sendResponseHeaders(200, -1); + } + + private void deleteObject(HttpExchange exchange, String key) throws IOException { + if (exchange.getRequestHeaders().containsKey("if-none-match")) unsupported("Conditional delete"); + store.delete(bucket, key); + exchange.sendResponseHeaders(204, -1); + } + private static String contentType(com.sun.net.httpserver.Headers headers) { String value = SigV4.single(headers, "content-type"); if (value == null) return "application/octet-stream"; @@ -337,6 +358,24 @@ public final class Main { } private void listObjects(HttpExchange exchange, Map query) throws IOException { + ListRequest request = listRequest(query); + var page = store.list(bucket, request.prefix(), request.delimiter(), request.maxKeys(), request.after()); + StringBuilder xml = new StringBuilder(""); + appendListHeader(xml, query, request, page); + appendListEntries(xml, page, request.encoding()); + if (page.truncated()) xml.append("") + .append(Base64.getUrlEncoder().withoutPadding().encodeToString(page.nextKey().getBytes(StandardCharsets.UTF_8))) + .append(""); + xml.append(""); + byte[] body = xml.toString().getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().set("Content-Type", "application/xml"); + exchange.sendResponseHeaders(200, body.length); + exchange.getResponseBody().write(body); + } + + private record ListRequest(String prefix, String delimiter, String encoding, int maxKeys, String after) { } + + private static ListRequest listRequest(Map query) { String prefix = query.getOrDefault("prefix", ""), delimiter = query.getOrDefault("delimiter", ""); String encoding = query.get("encoding-type"); if (encoding != null && !encoding.equals("url")) unsupported("Encoding type"); @@ -356,8 +395,11 @@ public final class Main { throw new StoreException(400, "InvalidArgument", "Invalid continuation token"); } } - var page = store.list(bucket, prefix, delimiter, maxKeys, after); - StringBuilder xml = new StringBuilder(""); + return new ListRequest(prefix, delimiter, encoding, maxKeys, after); + } + + private void appendListHeader(StringBuilder xml, Map query, ListRequest request, ObjectStorage.ListPage page) { + String prefix = request.prefix(), delimiter = request.delimiter(), encoding = request.encoding(); xml.append("").append(xml(bucket)).append("").append(xml(listKey(prefix, encoding))).append(""); if (!delimiter.isEmpty()) xml.append("").append(xml(listKey(delimiter, encoding))).append(""); if (encoding != null) xml.append("url"); @@ -365,8 +407,11 @@ public final class Main { .append(xml(query.get("continuation-token"))).append(""); if (query.containsKey("start-after")) xml.append("") .append(xml(listKey(query.get("start-after"), encoding))).append(""); - xml.append("").append(page.keyCount()).append("").append(maxKeys) + xml.append("").append(page.keyCount()).append("").append(request.maxKeys()) .append("").append(page.truncated()).append(""); + } + + private static void appendListEntries(StringBuilder xml, ObjectStorage.ListPage page, String encoding) { int objectAt = 0, prefixAt = 0; while (objectAt < page.objects().size() || prefixAt < page.prefixes().size()) { if (objectAt < page.objects().size() && @@ -384,14 +429,6 @@ public final class Main { .append(""); } } - if (page.truncated()) xml.append("") - .append(Base64.getUrlEncoder().withoutPadding().encodeToString(page.nextKey().getBytes(StandardCharsets.UTF_8))) - .append(""); - xml.append(""); - byte[] body = xml.toString().getBytes(StandardCharsets.UTF_8); - exchange.getResponseHeaders().set("Content-Type", "application/xml"); - exchange.sendResponseHeaders(200, body.length); - exchange.getResponseBody().write(body); } private static String listKey(String key, String encoding) { diff --git a/src/cloud/lunarsky/store/NodeRegistry.java b/src/cloud/lunarsky/store/NodeRegistry.java index 7657b7d..b6cfc4f 100644 --- a/src/cloud/lunarsky/store/NodeRegistry.java +++ b/src/cloud/lunarsky/store/NodeRegistry.java @@ -69,56 +69,10 @@ final class NodeRegistry { try (Statement statement = connection.createStatement()) { statement.execute("SELECT pg_advisory_xact_lock(6834071092781)"); } - Map stored = new HashMap<>(); - try (Statement statement = connection.createStatement(); - ResultSet result = statement.executeQuery("SELECT node_id, host_id, endpoint FROM cluster_nodes WHERE state <> 'retired'")) { - while (result.next()) { - URI url = URI.create(result.getString(3)); - stored.put(url.toString(), new NodeClient.Node((UUID) result.getObject(1), - (UUID) result.getObject(2), url)); - } - } - if (stored.isEmpty()) { - try (Statement statement = connection.createStatement(); - ResultSet result = statement.executeQuery("SELECT EXISTS (SELECT 1 FROM cluster_segments)")) { - result.next(); - if (result.getBoolean(1)) throw new IOException("Existing segments have no registered node identities"); - } - for (URI url : urls) { - NodeIdentity identity = NodeClient.probe(url, token); - try (PreparedStatement insert = connection.prepareStatement( - "INSERT INTO cluster_nodes (node_id, host_id, endpoint, state) VALUES (?, ?, ?, 'active')")) { - insert.setObject(1, identity.nodeId()); - insert.setObject(2, identity.hostId()); - insert.setString(3, url.toString()); - insert.executeUpdate(); - } - stored.put(url.toString(), new NodeClient.Node(identity.nodeId(), identity.hostId(), url)); - } - } - List configured = new ArrayList<>(); - Set configuredIds = new HashSet<>(); - for (URI url : urls) { - NodeClient.Node node = stored.get(url.toString()); - if (node == null) throw new IOException("Unregistered storage node URL: " + url); - NodeIdentity actual = null; - try { - actual = NodeClient.probe(url, token); - } catch (IOException offline) { } - if (actual != null && (!actual.nodeId().equals(node.id()) || !actual.hostId().equals(node.hostId()))) - throw new IOException("Storage node identity changed at " + url); - configured.add(node); - configuredIds.add(node.id()); - } - try (Statement statement = connection.createStatement(); - ResultSet result = statement.executeQuery( - "SELECT DISTINCT unnest(s.replica_ids) FROM cluster_segments s JOIN cluster_objects o ON o.generation=s.generation")) { - while (result.next()) { - UUID id = (UUID) result.getObject(1); - if (!configuredIds.contains(id)) - throw new IOException("A live segment refers to a node missing from CLUSTER_NODES: " + id); - } - } + Map stored = registeredNodes(connection); + if (stored.isEmpty()) registerInitialNodes(connection, urls, token, stored); + List configured = configuredNodes(urls, token, stored); + ensureLiveReplicasConfigured(connection, configured); NodeClient nodes = new NodeClient(configured, token, repairToken); connection.commit(); return nodes; @@ -132,4 +86,68 @@ final class NodeRegistry { catch (SQLException error) { throw new IOException("Could not restore metadata connection", error); } } } + + private static Map registeredNodes(Connection connection) throws SQLException { + Map stored = new HashMap<>(); + try (Statement statement = connection.createStatement(); + ResultSet result = statement.executeQuery("SELECT node_id, host_id, endpoint FROM cluster_nodes WHERE state <> 'retired'")) { + while (result.next()) { + URI url = URI.create(result.getString(3)); + stored.put(url.toString(), new NodeClient.Node((UUID) result.getObject(1), + (UUID) result.getObject(2), url)); + } + } + return stored; + } + + private static void registerInitialNodes(Connection connection, List urls, String token, + Map stored) throws SQLException, IOException { + try (Statement statement = connection.createStatement(); + ResultSet result = statement.executeQuery("SELECT EXISTS (SELECT 1 FROM cluster_segments)")) { + result.next(); + if (result.getBoolean(1)) throw new IOException("Existing segments have no registered node identities"); + } + for (URI url : urls) { + NodeIdentity identity = NodeClient.probe(url, token); + try (PreparedStatement insert = connection.prepareStatement( + "INSERT INTO cluster_nodes (node_id, host_id, endpoint, state) VALUES (?, ?, ?, 'active')")) { + insert.setObject(1, identity.nodeId()); + insert.setObject(2, identity.hostId()); + insert.setString(3, url.toString()); + insert.executeUpdate(); + } + stored.put(url.toString(), new NodeClient.Node(identity.nodeId(), identity.hostId(), url)); + } + } + + private static List configuredNodes(List urls, String token, + Map stored) throws IOException { + List configured = new ArrayList<>(); + for (URI url : urls) { + NodeClient.Node node = stored.get(url.toString()); + if (node == null) throw new IOException("Unregistered storage node URL: " + url); + NodeIdentity actual = null; + try { actual = NodeClient.probe(url, token); } + catch (IOException offline) { } + if (actual != null && (!actual.nodeId().equals(node.id()) || !actual.hostId().equals(node.hostId()))) + throw new IOException("Storage node identity changed at " + url); + configured.add(node); + } + return configured; + } + + private static void ensureLiveReplicasConfigured(Connection connection, List configured) + throws SQLException, IOException { + Set configuredIds = new HashSet<>(); + for (NodeClient.Node node : configured) configuredIds.add(node.id()); + try (Statement statement = connection.createStatement(); + ResultSet result = statement.executeQuery( + "SELECT DISTINCT unnest(s.replica_ids) FROM cluster_segments s JOIN cluster_objects o ON o.generation=s.generation")) { + while (result.next()) { + UUID id = (UUID) result.getObject(1); + if (!configuredIds.contains(id)) + throw new IOException("A live segment refers to a node missing from CLUSTER_NODES: " + id); + } + } + } } diff --git a/src/cloud/lunarsky/store/SigV4.java b/src/cloud/lunarsky/store/SigV4.java index 161066f..8cde1d7 100644 --- a/src/cloud/lunarsky/store/SigV4.java +++ b/src/cloud/lunarsky/store/SigV4.java @@ -28,28 +28,61 @@ final class SigV4 { } String verify(String method, URI uri, Headers headers) { + Map fields = authorizationFields(headers); + String[] credential = credentialScope(fields.get("Credential")); + String date = signingDate(headers, credential[1]); + String payload = payloadHash(headers); + String signedHeaders = fields.get("SignedHeaders"); + String canonicalHeaders = canonicalHeaders(headers, signedHeaders); + String canonical = method + "\n" + encode(decode(uri.getRawPath()), true) + "\n" + + canonicalQuery(uri.getRawQuery()) + "\n" + canonicalHeaders + "\n" + signedHeaders + "\n" + payload; + String scope = String.join("/", Arrays.copyOfRange(credential, 1, 5)); + String toSign = "AWS4-HMAC-SHA256\n" + date + "\n" + scope + "\n" + hex(hash(canonical.getBytes(StandardCharsets.UTF_8))); + byte[] signingKey = signingKey(secretKey, credential[1], region); + String signature = fields.get("Signature"); + if (!HEX.matcher(signature).matches() || !MessageDigest.isEqual(hmac(signingKey, toSign), HexFormat.of().parseHex(signature))) denied("Signature mismatch"); + return payload; + } + + private static Map authorizationFields(Headers headers) { String authorization = single(headers, "authorization"); if (authorization == null || !authorization.startsWith("AWS4-HMAC-SHA256 ")) denied("Signed requests are required"); - Map fields = new TreeMap<>(); + Map fields = new TreeMap<>(); for (String part : authorization.substring(17).split(",")) { String[] pair = part.trim().split("=", 2); if (pair.length != 2 || fields.put(pair[0], pair[1]) != null) denied("Invalid authorization header"); } if (!fields.keySet().equals(java.util.Set.of("Credential", "SignedHeaders", "Signature"))) denied("Invalid authorization fields"); - String[] credential = fields.get("Credential").split("/", -1); + return fields; + } + + private String[] credentialScope(String value) { + String[] credential = value.split("/", -1); if (credential.length != 5 || !credential[0].equals(accessKey) || !credential[2].equals(region) || !credential[3].equals("s3") || !credential[4].equals("aws4_request")) denied("Invalid credential scope"); - String date = single(headers, "x-amz-date"), payload = single(headers, "x-amz-content-sha256"); - if (date == null || !credential[1].matches("[0-9]{8}") || !date.matches("[0-9]{8}T[0-9]{6}Z") || !date.startsWith(credential[1])) denied("Invalid signing date"); + return credential; + } + + private String signingDate(Headers headers, String credentialDate) { + String date = single(headers, "x-amz-date"); + if (date == null || !credentialDate.matches("[0-9]{8}") || !date.matches("[0-9]{8}T[0-9]{6}Z") || !date.startsWith(credentialDate)) denied("Invalid signing date"); try { Instant signed = Instant.from(DATE.parse(date)); if (Duration.between(signed, clock.instant()).abs().compareTo(Duration.ofMinutes(5)) > 0) throw new StoreException(403, "RequestTimeTooSkewed", "Request timestamp is outside the permitted window"); } catch (java.time.DateTimeException e) { denied("Invalid signing date"); } + return date; + } + + private static String payloadHash(Headers headers) { + String payload = single(headers, "x-amz-content-sha256"); if (payload == null || !HEX.matcher(payload).matches()) throw new StoreException(400, "NotImplemented", "A hexadecimal SHA-256 payload hash is required; unsigned and chunk-signed payloads are unsupported"); if (headers.containsKey("x-amz-security-token")) denied("Temporary credentials are unsupported"); - String signedHeaders = fields.get("SignedHeaders"); + return payload; + } + + private static String canonicalHeaders(Headers headers, String signedHeaders) { String[] names = signedHeaders.split(";", -1); if (names.length > 32 || !signedHeaders.equals(String.join(";", Arrays.stream(names).distinct().sorted().toList()))) denied("Signed headers must be unique and sorted"); var namesSet = java.util.Set.copyOf(Arrays.asList(names)); @@ -66,14 +99,7 @@ final class SigV4 { if (value == null) denied("Missing signed header"); canonicalHeaders.append(name).append(':').append(value.trim().replaceAll("[\\t ]+", " ")).append('\n'); } - String canonical = method + "\n" + encode(decode(uri.getRawPath()), true) + "\n" - + canonicalQuery(uri.getRawQuery()) + "\n" + canonicalHeaders + "\n" + signedHeaders + "\n" + payload; - String scope = String.join("/", Arrays.copyOfRange(credential, 1, 5)); - String toSign = "AWS4-HMAC-SHA256\n" + date + "\n" + scope + "\n" + hex(hash(canonical.getBytes(StandardCharsets.UTF_8))); - byte[] signingKey = signingKey(secretKey, credential[1], region); - String signature = fields.get("Signature"); - if (!HEX.matcher(signature).matches() || !MessageDigest.isEqual(hmac(signingKey, toSign), HexFormat.of().parseHex(signature))) denied("Signature mismatch"); - return payload; + return canonicalHeaders.toString(); } static String single(Headers headers, String name) { diff --git a/test/cloud/lunarsky/store/HttpTest.java b/test/cloud/lunarsky/store/HttpTest.java index dd7e24c..234d20a 100644 --- a/test/cloud/lunarsky/store/HttpTest.java +++ b/test/cloud/lunarsky/store/HttpTest.java @@ -66,6 +66,114 @@ public final class HttpTest { } } + private static void testObjects(HttpClient client, String base) throws Exception { + status(200, client.send(HttpRequest.newBuilder(URI.create(base + "/health")).GET().build(), + HttpResponse.BodyHandlers.ofByteArray())); + String key = "folder/moon-☾.txt"; + byte[] body = "independent storage test".getBytes(StandardCharsets.UTF_8); + status(403, client.send(HttpRequest.newBuilder(URI.create(base + "/objects/" + key)) + .GET().build(), HttpResponse.BodyHandlers.ofByteArray())); + status(200, client.send(signed(base, "PUT", key, body), + HttpResponse.BodyHandlers.ofByteArray())); + String other = "folder/stars.txt"; + status(200, client.send(signedUri(URI.create(base + "/objects/" + other), "PUT", + "stars".getBytes(StandardCharsets.UTF_8), Map.of("content-type", "text/plain")), + HttpResponse.BodyHandlers.ofByteArray())); + var get = client.send(signed(base, "GET", key, new byte[0]), + HttpResponse.BodyHandlers.ofByteArray()); + status(200, get); + if (!java.util.Arrays.equals(body, get.body())) throw new AssertionError("GET body mismatch"); + if (!"application/octet-stream".equals(get.headers().firstValue("content-type").orElse(""))) + throw new AssertionError("Unexpected content type"); + var typed = client.send(signed(base, "GET", other, new byte[0]), HttpResponse.BodyHandlers.ofByteArray()); + status(200, typed); + if (!"text/plain".equals(typed.headers().firstValue("content-type").orElse(""))) + throw new AssertionError("Stored content type missing"); + var partial = client.send(signedUri(URI.create(base + "/objects/" + other), "GET", + new byte[0], Map.of("range", "bytes=1-3")), HttpResponse.BodyHandlers.ofByteArray()); + status(206, partial); + if (!"tar".equals(new String(partial.body(), StandardCharsets.UTF_8)) || + !"bytes 1-3/5".equals(partial.headers().firstValue("content-range").orElse(""))) + throw new AssertionError("Range response mismatch"); + status(416, client.send(signedUri(URI.create(base + "/objects/" + other), "GET", + new byte[0], Map.of("range", "bytes=20-30")), HttpResponse.BodyHandlers.ofByteArray())); + } + + private static void testListing(HttpClient client, String base) throws Exception { + var listed = client.send(signedUri(URI.create(base + "/objects?list-type=2&prefix=folder%2F"), + "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString()); + if (listed.statusCode() != 200 || !listed.body().contains("folder/stars.txt") || + !listed.body().contains("folder/moon-☾.txt")) + throw new AssertionError("ListObjectsV2 failed: " + listed.body()); + var page = client.send(signedUri(URI.create(base + "/objects?list-type=2&max-keys=1"), + "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString()); + if (page.statusCode() != 200 || !page.body().contains("true")) + throw new AssertionError("List pagination failed: " + page.body()); + String token = page.body().split("")[1].split("")[0]; + var next = client.send(signedUri(URI.create(base + "/objects?list-type=2&continuation-token=" + token), + "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString()); + if (next.statusCode() != 200 || !next.body().contains("folder/stars.txt")) + throw new AssertionError("List continuation failed: " + next.body()); + var grouped = client.send(signedUri(URI.create(base + "/objects?list-type=2&delimiter=%2F"), + "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString()); + if (grouped.statusCode() != 200 || !grouped.body().contains("folder/") || + grouped.body().contains("")) + throw new AssertionError("Delimiter listing failed: " + grouped.body()); + var encoded = client.send(signedUri(URI.create(base + "/objects?list-type=2&encoding-type=url"), + "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString()); + if (encoded.statusCode() != 200 || !encoded.body().contains("folder%2Fmoon-%E2%98%BE.txt")) + throw new AssertionError("Encoded listing failed: " + encoded.body()); + var emptyPage = client.send(signedUri(URI.create(base + "/objects?list-type=2&max-keys=0"), + "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString()); + if (emptyPage.statusCode() != 200 || !emptyPage.body().contains("0")) + throw new AssertionError("Empty list page failed: " + emptyPage.body()); + } + + private static void testMultipart(HttpClient client, String base) throws Exception { + String movie = "folder/video.mp4"; + URI initiate = URI.create(base + "/objects/" + movie + "?uploads="); + var created = client.send(signedUri(initiate, "POST", new byte[0], + Map.of("content-type", "video/mp4")), HttpResponse.BodyHandlers.ofString()); + if (created.statusCode() != 200) throw new AssertionError("Multipart initiation failed: " + created.body()); + String upload = created.body().split("")[1].split("")[0]; + byte[] first = "hello ".getBytes(StandardCharsets.UTF_8); + byte[] second = "world".getBytes(StandardCharsets.UTF_8); + var partOne = client.send(signedUri(URI.create(base + "/objects/" + movie + + "?partNumber=1&uploadId=" + upload), "PUT", first, Map.of()), HttpResponse.BodyHandlers.ofByteArray()); + var partTwo = client.send(signedUri(URI.create(base + "/objects/" + movie + + "?partNumber=2&uploadId=" + upload), "PUT", second, Map.of()), HttpResponse.BodyHandlers.ofByteArray()); + status(200, partOne); status(200, partTwo); + String completion = "1" + + partOne.headers().firstValue("etag").orElseThrow() + + "2" + + partTwo.headers().firstValue("etag").orElseThrow() + + ""; + status(200, client.send(signedUri(URI.create(base + "/objects/" + movie + "?uploadId=" + upload), + "POST", completion.getBytes(StandardCharsets.UTF_8), Map.of()), HttpResponse.BodyHandlers.ofByteArray())); + var assembled = client.send(signed(base, "GET", movie, new byte[0]), HttpResponse.BodyHandlers.ofByteArray()); + status(200, assembled); + if (!"hello world".equals(new String(assembled.body(), StandardCharsets.UTF_8)) || + !"video/mp4".equals(assembled.headers().firstValue("content-type").orElse(""))) + throw new AssertionError("Completed multipart object mismatch"); + var abandoned = client.send(signedUri(URI.create(base + "/objects/abandoned?uploads="), + "POST", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString()); + String abandonedId = abandoned.body().split("")[1].split("")[0]; + status(204, client.send(signedUri(URI.create(base + "/objects/abandoned?uploadId=" + abandonedId), + "DELETE", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofByteArray())); + } + + private static void testDelete(HttpClient client, String base) throws Exception { + String key = "folder/moon-☾.txt"; + var head = client.send(signed(base, "HEAD", key, new byte[0]), + HttpResponse.BodyHandlers.ofByteArray()); + status(200, head); + if (head.body().length != 0) throw new AssertionError("HEAD returned a body"); + status(204, client.send(signed(base, "DELETE", key, new byte[0]), + HttpResponse.BodyHandlers.ofByteArray())); + status(404, client.send(signed(base, "GET", key, new byte[0]), + HttpResponse.BodyHandlers.ofByteArray())); + } + public static void main(String[] args) throws Exception { Path root = Files.createTempDirectory("store-http-test-"); var executor = Executors.newVirtualThreadPerTaskExecutor(); @@ -79,101 +187,10 @@ public final class HttpTest { server.start(); String base = "http://127.0.0.1:" + server.getAddress().getPort(); HttpClient client = HttpClient.newHttpClient(); - status(200, client.send(HttpRequest.newBuilder(URI.create(base + "/health")).GET().build(), - HttpResponse.BodyHandlers.ofByteArray())); - String key = "folder/moon-☾.txt"; - byte[] body = "independent storage test".getBytes(StandardCharsets.UTF_8); - status(403, client.send(HttpRequest.newBuilder(URI.create(base + "/objects/" + key)) - .GET().build(), HttpResponse.BodyHandlers.ofByteArray())); - status(200, client.send(signed(base, "PUT", key, body), - HttpResponse.BodyHandlers.ofByteArray())); - String other = "folder/stars.txt"; - status(200, client.send(signedUri(URI.create(base + "/objects/" + other), "PUT", - "stars".getBytes(StandardCharsets.UTF_8), Map.of("content-type", "text/plain")), - HttpResponse.BodyHandlers.ofByteArray())); - var get = client.send(signed(base, "GET", key, new byte[0]), - HttpResponse.BodyHandlers.ofByteArray()); - status(200, get); - if (!java.util.Arrays.equals(body, get.body())) throw new AssertionError("GET body mismatch"); - if (!"application/octet-stream".equals(get.headers().firstValue("content-type").orElse(""))) - throw new AssertionError("Unexpected content type"); - var typed = client.send(signed(base, "GET", other, new byte[0]), HttpResponse.BodyHandlers.ofByteArray()); - status(200, typed); - if (!"text/plain".equals(typed.headers().firstValue("content-type").orElse(""))) - throw new AssertionError("Stored content type missing"); - var partial = client.send(signedUri(URI.create(base + "/objects/" + other), "GET", - new byte[0], Map.of("range", "bytes=1-3")), HttpResponse.BodyHandlers.ofByteArray()); - status(206, partial); - if (!"tar".equals(new String(partial.body(), StandardCharsets.UTF_8)) || - !"bytes 1-3/5".equals(partial.headers().firstValue("content-range").orElse(""))) - throw new AssertionError("Range response mismatch"); - status(416, client.send(signedUri(URI.create(base + "/objects/" + other), "GET", - new byte[0], Map.of("range", "bytes=20-30")), HttpResponse.BodyHandlers.ofByteArray())); - var listed = client.send(signedUri(URI.create(base + "/objects?list-type=2&prefix=folder%2F"), - "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString()); - if (listed.statusCode() != 200 || !listed.body().contains("folder/stars.txt") || - !listed.body().contains("folder/moon-☾.txt")) - throw new AssertionError("ListObjectsV2 failed: " + listed.body()); - var page = client.send(signedUri(URI.create(base + "/objects?list-type=2&max-keys=1"), - "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString()); - if (page.statusCode() != 200 || !page.body().contains("true")) - throw new AssertionError("List pagination failed: " + page.body()); - String token = page.body().split("")[1].split("")[0]; - var next = client.send(signedUri(URI.create(base + "/objects?list-type=2&continuation-token=" + token), - "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString()); - if (next.statusCode() != 200 || !next.body().contains("folder/stars.txt")) - throw new AssertionError("List continuation failed: " + next.body()); - var grouped = client.send(signedUri(URI.create(base + "/objects?list-type=2&delimiter=%2F"), - "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString()); - if (grouped.statusCode() != 200 || !grouped.body().contains("folder/") || - grouped.body().contains("")) - throw new AssertionError("Delimiter listing failed: " + grouped.body()); - var encoded = client.send(signedUri(URI.create(base + "/objects?list-type=2&encoding-type=url"), - "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString()); - if (encoded.statusCode() != 200 || !encoded.body().contains("folder%2Fmoon-%E2%98%BE.txt")) - throw new AssertionError("Encoded listing failed: " + encoded.body()); - var emptyPage = client.send(signedUri(URI.create(base + "/objects?list-type=2&max-keys=0"), - "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString()); - if (emptyPage.statusCode() != 200 || !emptyPage.body().contains("0")) - throw new AssertionError("Empty list page failed: " + emptyPage.body()); - String movie = "folder/video.mp4"; - URI initiate = URI.create(base + "/objects/" + movie + "?uploads="); - var created = client.send(signedUri(initiate, "POST", new byte[0], - Map.of("content-type", "video/mp4")), HttpResponse.BodyHandlers.ofString()); - if (created.statusCode() != 200) throw new AssertionError("Multipart initiation failed: " + created.body()); - String upload = created.body().split("")[1].split("")[0]; - byte[] first = "hello ".getBytes(StandardCharsets.UTF_8); - byte[] second = "world".getBytes(StandardCharsets.UTF_8); - var partOne = client.send(signedUri(URI.create(base + "/objects/" + movie + - "?partNumber=1&uploadId=" + upload), "PUT", first, Map.of()), HttpResponse.BodyHandlers.ofByteArray()); - var partTwo = client.send(signedUri(URI.create(base + "/objects/" + movie + - "?partNumber=2&uploadId=" + upload), "PUT", second, Map.of()), HttpResponse.BodyHandlers.ofByteArray()); - status(200, partOne); status(200, partTwo); - String completion = "1" + - partOne.headers().firstValue("etag").orElseThrow() + - "2" + - partTwo.headers().firstValue("etag").orElseThrow() + - ""; - status(200, client.send(signedUri(URI.create(base + "/objects/" + movie + "?uploadId=" + upload), - "POST", completion.getBytes(StandardCharsets.UTF_8), Map.of()), HttpResponse.BodyHandlers.ofByteArray())); - var assembled = client.send(signed(base, "GET", movie, new byte[0]), HttpResponse.BodyHandlers.ofByteArray()); - status(200, assembled); - if (!"hello world".equals(new String(assembled.body(), StandardCharsets.UTF_8)) || - !"video/mp4".equals(assembled.headers().firstValue("content-type").orElse(""))) - throw new AssertionError("Completed multipart object mismatch"); - var abandoned = client.send(signedUri(URI.create(base + "/objects/abandoned?uploads="), - "POST", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString()); - String abandonedId = abandoned.body().split("")[1].split("")[0]; - status(204, client.send(signedUri(URI.create(base + "/objects/abandoned?uploadId=" + abandonedId), - "DELETE", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofByteArray())); - var head = client.send(signed(base, "HEAD", key, new byte[0]), - HttpResponse.BodyHandlers.ofByteArray()); - status(200, head); - if (head.body().length != 0) throw new AssertionError("HEAD returned a body"); - status(204, client.send(signed(base, "DELETE", key, new byte[0]), - HttpResponse.BodyHandlers.ofByteArray())); - status(404, client.send(signed(base, "GET", key, new byte[0]), - HttpResponse.BodyHandlers.ofByteArray())); + testObjects(client, base); + testListing(client, base); + testMultipart(client, base); + testDelete(client, base); System.out.println("HTTP tests passed: health, authentication, PUT, GET, HEAD, DELETE, MIME, ranges, listing, multipart"); } finally { server.stop(0); diff --git a/test/cloud/lunarsky/store/StoreTest.java b/test/cloud/lunarsky/store/StoreTest.java index 3b6b3dc..9b1b94e 100644 --- a/test/cloud/lunarsky/store/StoreTest.java +++ b/test/cloud/lunarsky/store/StoreTest.java @@ -14,7 +14,7 @@ public final class StoreTest { static ObjectStorage.Metadata put(DiskStore store,String key,byte[] body,boolean only)throws Exception{ return store.put("test",key,new ByteArrayInputStream(body),body.length,SigV4.hex(SigV4.hash(body)),null,only,"application/octet-stream"); } - public static void main(String[] args)throws Exception{ + private static void testSignature() throws Exception { var headers=new com.sun.net.httpserver.Headers(); headers.set("host","examplebucket.s3.amazonaws.com");headers.set("range","bytes=0-9"); headers.set("x-amz-date","20130524T000000Z"); @@ -29,88 +29,116 @@ public final class StoreTest { fails(403,()->new SigV4("AKIAIOSFODNN7EXAMPLE","wrong","us-east-1",java.time.Clock.systemUTC()).verify("GET",uri,headers)); headers.add("host","duplicate");fails(403,()->auth.verify("GET",uri,headers)); System.out.println("SigV4 official vector and tampering tests passed"); - Path root=Files.createTempDirectory("store-test-"); - try{ - try(var store=new DiskStore(root,8,10)){ - byte[] body={1,2,3,4,5,6}; - put(store,"../nested/☾",body,true); - try(var obj=store.open("test","../nested/☾")){if(!Arrays.equals(body,obj.stream().readAllBytes()))throw new AssertionError("Roundtrip");} - fails(412,()->put(store,"../nested/☾",new byte[]{9},true)); - fails(507,()->put(store,"second",body,true)); - fails(413,()->put(store,"large",new byte[9],true)); - fails(400,()->store.put("test","bad",new ByteArrayInputStream(body),6,"0".repeat(64),null,true,"application/octet-stream")); - fails(400,()->store.put("test","short",new ByteArrayInputStream(body),7,SigV4.hex(SigV4.hash(body)),null,true,"application/octet-stream")); - fails(404,()->store.open("test","bad")); - put(store,"../nested/☾",new byte[]{9},false); - put(store,"empty",new byte[0],true); - if(store.list("test","","",100,null).objects().size()!=2)throw new AssertionError("List index"); - if(store.objectCount()!=2||store.legacyObjects()!=0)throw new AssertionError("Object counts"); - try { - new DiskStore(root,8,10); - throw new AssertionError("A second process opened the same data directory"); - } catch(IOException expected) { - if(!expected.getMessage().contains("already in use"))throw expected; - } - } - try(var restarted=new DiskStore(root,8,10)){ - try(var obj=restarted.open("test","../nested/☾")){if(obj.stream().read()!=9)throw new AssertionError("Persistence");} - if(restarted.list("test","","",100,null).objects().size()!=2)throw new AssertionError("Index persistence"); - restarted.delete("test","../nested/☾");restarted.delete("test","../nested/☾"); - fails(404,()->restarted.open("test","../nested/☾")); - var uploads=new MultipartStore(restarted); - String upload=uploads.create("test","from-parts","text/plain"); - byte[] part={1,2,3}; - uploads.putPart(upload,"test","from-parts",1,new ByteArrayInputStream(part),part.length, - SigV4.hex(SigV4.hash(part)),null); - Files.writeString(root.resolve("pending-upload-id"),upload); - } - try(var resumed=new DiskStore(root,8,10)){ - Path unfinished=root.resolve("multipart/.creating-00000000-0000-0000-0000-000000000000"); - Files.createDirectory(unfinished); - Files.write(unfinished.resolve("manifest"),new byte[]{1,2,3}); - var uploads=new MultipartStore(resumed); - if(Files.exists(unfinished))throw new AssertionError("Unfinished multipart creation survived restart"); - String upload=Files.readString(root.resolve("pending-upload-id")); - byte[] part={1,2,3}; - String etag=SigV4.hex(MessageDigest.getInstance("MD5").digest(part)); - uploads.complete(upload,"test","from-parts",List.of(new MultipartStorage.Part(1,etag))); - try(var obj=resumed.open("test","from-parts")){ - if(!Arrays.equals(part,obj.stream().readAllBytes())||!obj.metadata().contentType().equals("text/plain")) - throw new AssertionError("Multipart restart"); - } - resumed.delete("test","from-parts"); - } - byte[] old={4,5,6}; - String oldId=SigV4.hex(SigV4.hash("test/legacy".getBytes(java.nio.charset.StandardCharsets.UTF_8))); - Path oldPath=root.resolve("objects").resolve(oldId.substring(0,2)).resolve(oldId); - Files.createDirectories(oldPath.getParent()); - ByteBuffer oldRecord=ByteBuffer.allocate(72+old.length).putLong(0x4c534f424a303031L) - .putLong(old.length).putLong(123456789L) - .put(MessageDigest.getInstance("MD5").digest(old)).put(SigV4.hash(old)).put(old); - Files.write(oldPath,oldRecord.array()); - try(var migrated=new DiskStore(root,8,10)){ - if(migrated.objectCount()!=2||migrated.legacyObjects()!=1)throw new AssertionError("Legacy counts"); - try(var obj=migrated.open("test","legacy")){ - if(!Arrays.equals(old,obj.stream().readAllBytes()))throw new AssertionError("Legacy read"); - } - if(migrated.list("test","","",100,null).objects().stream().anyMatch(entry->entry.key().equals("legacy"))) - throw new AssertionError("Legacy object appeared without a stored key"); - put(migrated,"legacy",old,false); - if(migrated.objectCount()!=2||migrated.legacyObjects()!=0)throw new AssertionError("Legacy count after overwrite"); - if(migrated.list("test","","",100,null).objects().stream().noneMatch(entry->entry.key().equals("legacy"))) - throw new AssertionError("Legacy overwrite was not indexed"); - } - Files.delete(root.resolve("pending-upload-id")); - String id=SigV4.hex(SigV4.hash("test/empty".getBytes(java.nio.charset.StandardCharsets.UTF_8))); - Files.write(root.resolve("objects").resolve(id.substring(0,2)).resolve(id),new byte[]{1},StandardOpenOption.APPEND); + } + + private static void testInitialStore(Path root) throws Exception { + try(var store=new DiskStore(root,8,10)){ + byte[] body={1,2,3,4,5,6}; + put(store,"../nested/☾",body,true); + try(var obj=store.open("test","../nested/☾")){if(!Arrays.equals(body,obj.stream().readAllBytes()))throw new AssertionError("Roundtrip");} + fails(412,()->put(store,"../nested/☾",new byte[]{9},true)); + fails(507,()->put(store,"second",body,true)); + fails(413,()->put(store,"large",new byte[9],true)); + fails(400,()->store.put("test","bad",new ByteArrayInputStream(body),6,"0".repeat(64),null,true,"application/octet-stream")); + fails(400,()->store.put("test","short",new ByteArrayInputStream(body),7,SigV4.hex(SigV4.hash(body)),null,true,"application/octet-stream")); + fails(404,()->store.open("test","bad")); + put(store,"../nested/☾",new byte[]{9},false); + put(store,"empty",new byte[0],true); + if(store.list("test","","",100,null).objects().size()!=2)throw new AssertionError("List index"); + if(store.objectCount()!=2||store.legacyObjects()!=0)throw new AssertionError("Object counts"); try { new DiskStore(root,8,10); - throw new AssertionError("A damaged object record was accepted"); + throw new AssertionError("A second process opened the same data directory"); } catch(IOException expected) { - if(!expected.getMessage().contains("object record"))throw expected; + if(!expected.getMessage().contains("already in use"))throw expected; } - try(var pending=Files.list(root.resolve("pending"))){if(pending.count()!=0)throw new AssertionError("Pending cleanup");} + } + } + + private static void testRestart(Path root) throws Exception { + try(var restarted=new DiskStore(root,8,10)){ + try(var obj=restarted.open("test","../nested/☾")){if(obj.stream().read()!=9)throw new AssertionError("Persistence");} + if(restarted.list("test","","",100,null).objects().size()!=2)throw new AssertionError("Index persistence"); + restarted.delete("test","../nested/☾");restarted.delete("test","../nested/☾"); + fails(404,()->restarted.open("test","../nested/☾")); + var uploads=new MultipartStore(restarted); + String upload=uploads.create("test","from-parts","text/plain"); + byte[] part={1,2,3}; + uploads.putPart(upload,"test","from-parts",1,new ByteArrayInputStream(part),part.length, + SigV4.hex(SigV4.hash(part)),null); + Files.writeString(root.resolve("pending-upload-id"),upload); + } + } + + private static void testMultipartRecovery(Path root) throws Exception { + try(var resumed=new DiskStore(root,8,10)){ + Path unfinished=root.resolve("multipart/.creating-00000000-0000-0000-0000-000000000000"); + Files.createDirectory(unfinished); + Files.write(unfinished.resolve("manifest"),new byte[]{1,2,3}); + var uploads=new MultipartStore(resumed); + if(Files.exists(unfinished))throw new AssertionError("Unfinished multipart creation survived restart"); + String upload=Files.readString(root.resolve("pending-upload-id")); + byte[] part={1,2,3}; + String etag=SigV4.hex(MessageDigest.getInstance("MD5").digest(part)); + uploads.complete(upload,"test","from-parts",List.of(new MultipartStorage.Part(1,etag))); + try(var obj=resumed.open("test","from-parts")){ + if(!Arrays.equals(part,obj.stream().readAllBytes())||!obj.metadata().contentType().equals("text/plain")) + throw new AssertionError("Multipart restart"); + } + resumed.delete("test","from-parts"); + } + } + + private static void testLegacyRecord(Path root) throws Exception { + byte[] old={4,5,6}; + String oldId=SigV4.hex(SigV4.hash("test/legacy".getBytes(java.nio.charset.StandardCharsets.UTF_8))); + Path oldPath=root.resolve("objects").resolve(oldId.substring(0,2)).resolve(oldId); + Files.createDirectories(oldPath.getParent()); + ByteBuffer oldRecord=ByteBuffer.allocate(72+old.length).putLong(0x4c534f424a303031L) + .putLong(old.length).putLong(123456789L) + .put(MessageDigest.getInstance("MD5").digest(old)).put(SigV4.hash(old)).put(old); + Files.write(oldPath,oldRecord.array()); + try(var migrated=new DiskStore(root,8,10)){ + if(migrated.objectCount()!=2||migrated.legacyObjects()!=1)throw new AssertionError("Legacy counts"); + try(var obj=migrated.open("test","legacy")){ + if(!Arrays.equals(old,obj.stream().readAllBytes()))throw new AssertionError("Legacy read"); + } + if(migrated.list("test","","",100,null).objects().stream().anyMatch(entry->entry.key().equals("legacy"))) + throw new AssertionError("Legacy object appeared without a stored key"); + put(migrated,"legacy",old,false); + if(migrated.objectCount()!=2||migrated.legacyObjects()!=0)throw new AssertionError("Legacy count after overwrite"); + if(migrated.list("test","","",100,null).objects().stream().noneMatch(entry->entry.key().equals("legacy"))) + throw new AssertionError("Legacy overwrite was not indexed"); + } + } + + private static void testCorruption(Path root) throws Exception { + Files.delete(root.resolve("pending-upload-id")); + String id=SigV4.hex(SigV4.hash("test/empty".getBytes(java.nio.charset.StandardCharsets.UTF_8))); + Files.write(root.resolve("objects").resolve(id.substring(0,2)).resolve(id),new byte[]{1},StandardOpenOption.APPEND); + try { + new DiskStore(root,8,10); + throw new AssertionError("A damaged object record was accepted"); + } catch(IOException expected) { + if(!expected.getMessage().contains("object record"))throw expected; + } + try(var pending=Files.list(root.resolve("pending"))){if(pending.count()!=0)throw new AssertionError("Pending cleanup");} + } + + public static void main(String[] args) throws Exception { + testSignature(); + Path root = Files.createTempDirectory("store-test-"); + try { + testInitialStore(root); + testRestart(root); + testMultipartRecovery(root); + testLegacyRecord(root); + testCorruption(root); System.out.println("Java storage tests passed: roundtrip, quota, indexing, persistence, multipart recovery, legacy reads, locking, corruption, delete"); - }finally{try(var paths=Files.walk(root)){for(var p:paths.sorted(java.util.Comparator.reverseOrder()).toList())Files.delete(p);}} + } finally { + try (var paths = Files.walk(root)) { + for (var path : paths.sorted(java.util.Comparator.reverseOrder()).toList()) Files.delete(path); + } + } } }