Guard against false positive block attacks on filtered blockstore

Peergos · Sep 25, 2024 · b86d804 · b86d804
1 parent adefb10
commit b86d804
Show file tree

Hide file tree

Showing 4 changed files with 47 additions and 11 deletions.
diff --git a/src/main/java/org/peergos/blockstore/CidInfiniFilter.java b/src/main/java/org/peergos/blockstore/CidInfiniFilter.java
@@ -35,13 +35,18 @@ public static CidInfiniFilter build(Blockstore bs) {
     public static CidInfiniFilter build(Blockstore bs, double falsePositiveRate) {
         List<Cid> refs = bs.refs(false).join();
         int nBlocks = refs.size()*5/4; //  increase by 25% to avoid expansion during build
+        CidInfiniFilter infini = build(nBlocks, falsePositiveRate);
+        refs.forEach(c -> infini.add(c));
+        return infini;
+    }
+
+    public static CidInfiniFilter build(int nBlocks, double falsePositiveRate) {
         int nextPowerOfTwo = Math.max(17, (int) (1 + Math.log(nBlocks) / Math.log(2)));
         double expansionAlpha = 0.8;
         int bitsPerEntry = (int)(4 - Math.log(falsePositiveRate / expansionAlpha) / Math.log(2) + 1);
         LOG.info("Using infini filter of initial size " + ((double)(bitsPerEntry * (1 << nextPowerOfTwo) / 8) / 1024 / 1024) + " MiB");
         ChainedInfiniFilter infini = new ChainedInfiniFilter(nextPowerOfTwo, bitsPerEntry);
         infini.set_expand_autonomously(true);
-        refs.forEach(c -> infini.insert(c.toBytes(), true));
         return new CidInfiniFilter(infini);
     }
 }
diff --git a/src/main/java/org/peergos/blockstore/FilteredBlockstore.java b/src/main/java/org/peergos/blockstore/FilteredBlockstore.java
@@ -8,27 +8,48 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.*;
 
 public class FilteredBlockstore implements Blockstore {
 
     private final Blockstore blocks;
-    private final Filter filter;
+    private final Filter present;
+    private volatile Filter absent;
+    private final AtomicLong absentCount = new AtomicLong(0);
 
-    public FilteredBlockstore(Blockstore blocks, Filter filter) {
+    public FilteredBlockstore(Blockstore blocks, Filter present) {
         this.blocks = blocks;
-        this.filter = filter;
+        this.present = present;
+        this.absent = buildAbsentFilter();
     }
 
     public CompletableFuture<Boolean> bloomAdd(Cid cid) {
-        filter.add(cid);
+        present.add(cid);
         return CompletableFuture.completedFuture(true);
     }
 
+    private static Filter buildAbsentFilter() {
+        return CidInfiniFilter.build(1_000, 0.001);
+    }
+
+    private void addAbsentBlock(Cid c) {
+        if (absentCount.get() > 10_000) {
+            absentCount.set(0);
+            absent = buildAbsentFilter();
+        }
+        absentCount.incrementAndGet();
+        absent.add(c);
+    }
+
     @Override
     public CompletableFuture<Boolean> has(Cid c) {
-        if (filter.has(c))
-            return blocks.has(c);
+        if (present.has(c) && ! absent.has(c))
+            return blocks.has(c).thenApply(res -> {
+                if (! res)
+                    addAbsentBlock(c);
+                return false;
+            });
         return CompletableFuture.completedFuture(false);
     }
 
@@ -40,15 +61,19 @@ public CompletableFuture<Boolean> hasAny(Multihash h) {
 
     @Override
     public CompletableFuture<Optional<byte[]>> get(Cid c) {
-        if (filter.has(c))
-            return blocks.get(c);
+        if (present.has(c) && ! absent.has(c)) {
+            return blocks.get(c).exceptionally(t -> {
+                addAbsentBlock(c);
+                return Optional.empty();
+            });
+        }
         return CompletableFuture.completedFuture(Optional.empty());
     }
 
     @Override
     public CompletableFuture<Cid> put(byte[] block, Cid.Codec codec) {
         return blocks.put(block, codec)
-                .thenApply(filter::add);
+                .thenApply(present::add);
     }
 
     @Override

diff --git a/src/main/java/org/peergos/blockstore/s3/S3Blockstore.java b/src/main/java/org/peergos/blockstore/s3/S3Blockstore.java
@@ -349,7 +349,7 @@ private CompletableFuture<Optional<byte[]>> getWithoutRetry(Cid cid) {
                 LOG.log(Level.WARNING, msg, e);
             }
             failedBlockGets.inc();
-            throw new RuntimeException(e.getMessage(), e);
+            return Futures.errored(e);
         } finally {
             readTimer.observeDuration();
         }

diff --git a/src/main/java/org/peergos/util/Futures.java b/src/main/java/org/peergos/util/Futures.java
@@ -19,4 +19,10 @@ private static <T> T logAndThrow(Throwable t, Optional<String> message) {
         t.printStackTrace();
         throw new RuntimeException(t.getMessage(), t);
     }
+
+    public static <T> CompletableFuture<T> errored(Throwable t) {
+        CompletableFuture<T> err = new CompletableFuture<>();
+        err.completeExceptionally(t);
+        return err;
+    }
 }