rabin-aes: new chunker with provable fingerprinting resistance

Add a "UHF-then-PRF" content-defined chunker following the Chk-PHTE construction of "Breaking and Fixing Content-Defined Chunking" (Truong et al., eprint 2025/558): a rolling Rabin fingerprint over GF(2)[x]/P(x) - with P a secret, random, irreducible polynomial of degree 63 - hashes the last 64 bytes (a universal hash family), and AES-128 with a secret key is applied to the digest; the cut decision only looks at the AES output. Observed chunk boundaries are thus pseudo-random and yield no usable equations about the chunking secrets, unlike chunkers cutting directly on (keyed) rolling-hash bits. Both secrets are derived from the id key (domain "rabin-aes"). The C scan kernel has three bit-identical paths: batched OpenSSL EVP AES-128-ECB (portable), arm64 crypto-extension intrinsics and x86-64 AES-NI (runtime detected), interleaving the AES work with the serial rolling-hash chain. Batching ahead of the cut decision is possible because the digest stream does not depend on where cuts happen. Throughput on an Apple M-series (1 GiB random, 19,23,21,2): 489 MB/s (arm64 hw path), 334 MB/s (EVP path); chunk-size distribution, shift resilience and dedup behavior are identical to fastcdc/buzhash64. Normalized chunking, param shape rabin-aes,min,max,mask,nc_level. Co-Authored-By: Claude Fable 5 <[email protected]>
2026-10-12 02:11:36 +00:00 · 2026-08-04 15:43:33 +02:00
parent f8dee0c3ff
commit b7dcf30bd4
15 changed files with 1188 additions and 4 deletions
@@ -11,6 +11,7 @@ src/borg/item.c
 src/borg/chunkers/buzhash.c
 src/borg/chunkers/buzhash64.c
 src/borg/chunkers/fastcdc.c
+src/borg/chunkers/rabin_aes.c
 src/borg/chunkers/reader.c
 src/borg/checksums.c
 src/borg/platform/darwin.c
@@ -253,6 +253,9 @@ New features:
    XORing a 32bit seed into the table), so chunk cut points are much harder to
    predict without the key (better resistance against fingerprinting attacks).
  - buzhash64: add normalized chunking (better chunk size distribution, less clamping)
+  - rabin-aes: new chunker with cryptographically sound resistance against
+    chunk-size fingerprinting (UHF-then-PRF: secret Rabin polynomial + AES-128,
+    following eprint 2025/558); uses AES hw acceleration or OpenSSL, ~450+ MB/s
 - borg keys:

  - locate the borg key automatically in the key directory or in the repository, #9743
@@ -408,6 +408,10 @@ Borg has these chunkers (the default is "fastcdc"):
 - "buzhash64": similar to "buzhash", but improved 64bit implementation
 - "fastcdc": variable, content-defined blocksize, uses the window-less, keyed
  Gear rolling hash (FastCDC_); faster than buzhash, same deduplication.
+- "rabin-aes": variable, content-defined blocksize; a rolling Rabin fingerprint
+  (secret polynomial) post-processed with AES-128, so the cut decision only
+  depends on the AES output ("UHF-then-PRF" construction). Strongest available
+  protection against chunk-size fingerprinting attacks.

 For some more general usage hints see also ``--chunker-params``.

@@ -514,6 +518,31 @@ distribution and reduces clamping at the min./max. chunk size.
 This is the default chunker (``fastcdc,19,23,21,2``), also used for the item
 metadata stream (with a finer granularity, ``fastcdc,15,19,17,2``).

+"rabin-aes" chunker
+++++++++++++++++++
+
+A "UHF-then-PRF" content-defined chunker, following the provably secure
+construction of `Breaking and Fixing Content-Defined Chunking
+<https://eprint.iacr.org/2025/558>`_ (Truong et al., 2025): a rolling Rabin
+fingerprint over GF(2)[x]/P(x) - with P a secret, random, irreducible
+polynomial of degree 63 - compresses the last 64 bytes into a digest
+(a universal hash), and AES-128 with a secret key is applied to that digest.
+The cut decision only looks at the AES output, so observed chunk boundaries
+are pseudo-random and do not provide usable equations about the chunking
+secrets, unlike chunkers that cut directly on (keyed) rolling hash bits.
+Both secrets are derived from the repository key material.
+
+This is the recommended chunker when resistance against chunk-size
+fingerprinting attacks matters most. It is slower than "fastcdc" (one AES
+block encryption per scanned byte), but still fast in absolute terms: the
+implementation batches the AES work through OpenSSL or uses AES hardware
+instructions (arm64 crypto extensions / x86-64 AES-NI) where available.
+
+``borg create --chunker-params rabin-aes,CHUNK_MIN_EXP,CHUNK_MAX_EXP,HASH_MASK_BITS,NC_LEVEL``
+
+The window size is fixed at 64 bytes. NC_LEVEL is the normalized chunking
+level (0 disables it); 2 is a good default. E.g.: ``rabin-aes,19,23,21,2``.
+
 .. _cache:

 The files cache
@@ -131,8 +131,8 @@ def chunk_stats(algo, data, min_exp, max_exp, mask_bits, win, nc_level=0, normal
    if algo == "buzhash64":
        params.append(nc_level)  # nc_level is a positional param
        kw["normal_size"] = normal_size
-    elif algo == "fastcdc":
-        params = [min_exp, max_exp, mask_bits, nc_level]  # fastcdc is window-less
+    elif algo in ("fastcdc", "rabin-aes"):
+        params = [min_exp, max_exp, mask_bits, nc_level]  # window-less param shape
        kw["normal_size"] = normal_size
    chunker = get_chunker(algo, *params, **kw)
    sizes = []
@@ -288,11 +288,11 @@ def main():
    print(f"shift test: {args.shift_edits} edits   repeats: {args.repeat}")
    print("-" * 118)

-    # build (algo, nc_level) variants; for buzhash64/fastcdc also run the requested NC level
+    # build (algo, nc_level) variants; for buzhash64/fastcdc/rabin-aes also run the requested NC level
    variants = []
    for algo in args.algo:
        variants.append((algo, 0))
-        if algo in ("buzhash64", "fastcdc") and args.nc_level > 0:
+        if algo in ("buzhash64", "fastcdc", "rabin-aes") and args.nc_level > 0:
            variants.append((algo, args.nc_level))

    for algo, nc in variants:
@@ -57,6 +57,8 @@ crypto_legacy_ll_source = "src/borg/legacy/crypto/low_level.pyx"
 buzhash_source = "src/borg/chunkers/buzhash.pyx"
 buzhash64_source = "src/borg/chunkers/buzhash64.pyx"
 fastcdc_source = "src/borg/chunkers/fastcdc.pyx"
+rabin_aes_source = "src/borg/chunkers/rabin_aes.pyx"
+rabin_aes_impl_source = "src/borg/chunkers/rabin_aes_impl.c"
 reader_source = "src/borg/chunkers/reader.pyx"
 hashindex_source = "src/borg/hashindex.pyx"
 item_source = "src/borg/item.pyx"
@@ -75,6 +77,7 @@ cython_sources = [
    buzhash_source,
    buzhash64_source,
    fastcdc_source,
+    rabin_aes_source,
    reader_source,
    hashindex_source,
    item_source,
@@ -166,6 +169,13 @@ if not on_rtd:
        dict(sources=[crypto_legacy_ll_source]), crypto_ext_lib, dict(extra_compile_args=cflags)
    )

+    # rabin-aes uses OpenSSL (EVP AES-128-ECB) in its C scan kernel
+    rabin_aes_ext_kwargs = members_appended(
+        dict(sources=[rabin_aes_source, rabin_aes_impl_source], include_dirs=["src/borg/chunkers"]),
+        crypto_ext_lib,
+        dict(extra_compile_args=cflags),
+    )
+
    compress_ext_kwargs = members_appended(
        dict(sources=[compress_source]),
        lib_ext_kwargs(pc, "BORG_LIBLZ4_PREFIX", "lz4", "liblz4", ">= 1.7.0"),
@@ -192,6 +202,7 @@ if not on_rtd:
        Extension("borg.chunkers.buzhash", [buzhash_source], extra_compile_args=cflags),
        Extension("borg.chunkers.buzhash64", [buzhash64_source], extra_compile_args=cflags),
        Extension("borg.chunkers.fastcdc", [fastcdc_source], extra_compile_args=cflags),
+        Extension("borg.chunkers.rabin_aes", **rabin_aes_ext_kwargs),
        Extension("borg.chunkers.reader", [reader_source], extra_compile_args=cflags),
    ]

@@ -195,6 +195,13 @@ class BenchmarkMixIn:
            ),
            # fastcdc (window-less keyed gear hash); gear table creation is slow, keep it in setup
            ("fastcdc,19,23,21,2", "ch = get_chunker('fastcdc', 19, 23, 21, 2, sparse=False)", "chunkit(ch)", locals()),
+            # rabin-aes (UHF-then-PRF); polynomial sampling / table creation is slow, keep it in setup
+            (
+                "rabin-aes,19,23,21,2",
+                "ch = get_chunker('rabin-aes', 19, 23, 21, 2, sparse=False)",
+                "chunkit(ch)",
+                locals(),
+            ),
            ("fixed,1048576", "ch = get_chunker('fixed', 1048576, sparse=False)", "chunkit(ch)", locals()),
        ]:
            dt = timeit(func, setup, number=number_default, globals=vars)
@@ -714,6 +714,7 @@ class CompletionMixIn:
            "buzhash,19,23,21,4095",
            "buzhash64,19,23,21,4095,2",
            "fastcdc,19,23,21,2",
+            "rabin-aes,19,23,21,2",
        ]
        chunker_params_choices_str = " ".join(chunker_params_choices)

@@ -1,6 +1,7 @@
 from .buzhash import Chunker
 from .buzhash64 import ChunkerBuzHash64
 from .fastcdc import ChunkerFastCDC
+from .rabin_aes import ChunkerRabinAES
 from .failing import ChunkerFailing
 from .fixed import ChunkerFixed
 from .reader import *  # noqa
@@ -28,6 +29,14 @@ def get_chunker(algo, *params, **kw):
            key.derive_key(salt=b"", domain=b"fastcdc", size=32, from_id_key=True) if key is not None else b"\0" * 32
        )
        return ChunkerFastCDC(fc_key, *params, normal_size=kw.get("normal_size", 0), sparse=sparse)
+    if algo == "rabin-aes":
+        # UHF-then-PRF chunker (secret Rabin polynomial + AES-128), derived from
+        # the id key (own domain). params is (chunk_min_exp, chunk_max_exp,
+        # hash_mask_bits, nc_level) - no window param (fixed 64-byte window).
+        ra_key = (
+            key.derive_key(salt=b"", domain=b"rabin-aes", size=32, from_id_key=True) if key is not None else b"\0" * 32
+        )
+        return ChunkerRabinAES(ra_key, *params, normal_size=kw.get("normal_size", 0), sparse=sparse)
    if algo == "fixed":
        return ChunkerFixed(*params, sparse=sparse)
    if algo == "fail":
@@ -0,0 +1,22 @@
+from collections.abc import Iterator
+from typing import BinaryIO
+
+from .reader import fmap_entry
+
+def rabin_aes_get_polynomial(key: bytes) -> int: ...
+def rabin_aes_get_tables(key: bytes) -> tuple[list[int], list[int]]: ...
+def _is_irreducible(p: int) -> bool: ...
+
+class ChunkerRabinAES:
+    kernel: str
+    def __init__(
+        self,
+        key: bytes,
+        chunk_min_exp: int,
+        chunk_max_exp: int,
+        hash_mask_bits: int,
+        nc_level: int = 0,
+        normal_size: int = 0,
+        sparse: bool = False,
+    ) -> None: ...
+    def chunkify(self, fd: BinaryIO = None, fh: int = -1, fmap: list[fmap_entry] = None) -> Iterator: ...
@@ -0,0 +1,420 @@
+# cython: language_level=3
+
+import os
+import cython
+import time
+
+from cpython.bytes cimport PyBytes_AsString
+from libc.stdint cimport uint8_t, uint64_t, int64_t
+from libc.stdlib cimport malloc, free
+from libc.string cimport memcpy, memmove, memset
+
+from ..crypto.low_level import CSPRNG
+
+from ..constants import CH_DATA, CH_ALLOC, CH_HOLE, zeros
+from .reader import FileReader, Chunk
+
+# rabin-aes: a "Chk-PHTE" style content-defined chunker (UHF-then-PRF).
+#
+# Motivated by "Breaking and Fixing Content-Defined Chunking" (Truong, Merz,
+# Scarlata, Guenther, Paterson; eprint 2025/558) and Alexeev/Percival/Zhang
+# (arXiv 2504.02095): keyed chunkers built by hiding a secret inside a rolling
+# hash (secret buzhash/gear tables) admit key-recovery attacks from known
+# plaintext, because every observed cut is an equation about the secret.
+# The provably sound construction separates the two jobs:
+#
+#   1. a rolling *universal hash function* (UHF) compresses the last w bytes
+#      into a short digest; here: a Rabin fingerprint over GF(2)[x]/P(x) with
+#      a secret random irreducible polynomial P of degree 63, over a 64-byte
+#      window (the same UHF family restic uses, with a bigger polynomial).
+#   2. a *PRF* (AES-128 with a secret key) is applied to the digest; the cut
+#      decision looks only at the PRF output (low bits of the first 8
+#      ciphertext bytes, little-endian): cut iff (AES(digest) & mask) == 0.
+#
+# An attacker who observes cut positions sees only PRF outputs; the digest and
+# P remain hidden behind AES. Two windows colliding under the UHF is the only
+# residual leakage (they necessarily get equal decisions); for a degree-63
+# random irreducible P the collision probability of two distinct 64-byte
+# windows is <= 511/2^62 ~= 2^-53.
+#
+# Security notes (honest limits):
+#  * The FSWC-RoR proof bound (eprint 2025/558, Thm 4.5 analogue) degrades
+#    with the square of the number of processed positions; with a 63-bit
+#    digest it stays meaningful up to roughly tens of GiB per chunker key.
+#    Beyond that, no *attack* is known - but the guarantee is heuristic, like
+#    everything else in this area except a wider-digest construction.
+#  * As with any deterministic chunker, equal chunks still produce equal
+#    lengths (deduplication requires this); see the paper's discussion.
+#
+# Performance: the digest stream is independent of where cuts happen (each
+# digest is a pure function of the last 64 bytes), so digests can be computed
+# ahead and encrypted in batches. The C kernel (rabin_aes_impl.c) does this
+# either via OpenSSL EVP AES-128-ECB batches (portable) or, where available,
+# via AES instructions (arm64 crypto extension / x86-64 AES-NI) interleaved
+# with the rolling-hash chain. All paths produce bit-identical cuts.
+
+cdef extern from "rabin_aes_impl.h":
+    ctypedef struct RA_CTX:
+        pass
+    RA_CTX *ra_new(const uint64_t *out_tbl, const uint64_t *red_tbl, const uint8_t *aes_key, int force_sw)
+    void ra_free(RA_CTX *ctx)
+    const char *ra_kind(const RA_CTX *ctx)
+    uint64_t ra_digest64(const RA_CTX *ctx, const uint8_t *q)
+    int64_t ra_scan(RA_CTX *ctx, const uint8_t *p, size_t n, uint64_t *digest, uint64_t mask)
+
+
+# --- GF(2)[x] polynomial helpers (pure Python, used at chunker setup only) ---
+
+_DEG = 63  # degree of the secret irreducible polynomial P
+_WINDOW_SIZE = 64  # bytes; window polynomial degree is 8 * 64 - 1 = 511
+
+# spread table for GF(2) squaring: 8 bits -> 16 bits with zeros interleaved
+_SPREAD8 = []
+for _b in range(256):
+    _s = 0
+    for _i in range(8):
+        if _b & (1 << _i):
+            _s |= 1 << (2 * _i)
+    _SPREAD8.append(_s)
+
+
+def _poly_mod(v, p):
+    """v mod p in GF(2)[x]; p must have its top bit set at degree deg(p)."""
+    dp = p.bit_length() - 1
+    while v.bit_length() - 1 >= dp:
+        v ^= p << (v.bit_length() - 1 - dp)
+    return v
+
+
+def _poly_sq_mod(a, p):
+    """a^2 mod p in GF(2)[x] (square = interleave zeros between bits)."""
+    sq = 0
+    shift = 0
+    while a:
+        sq |= _SPREAD8[a & 0xFF] << shift
+        a >>= 8
+        shift += 16
+    return _poly_mod(sq, p)
+
+
+def _poly_gcd(a, b):
+    """gcd(a, b) in GF(2)[x]."""
+    while b:
+        a, b = b, _poly_mod(a, b)
+    return a
+
+
+def _is_irreducible(p):
+    """Rabin's irreducibility test for deg(p) == 63 (prime factors of 63: 3, 7).
+
+    p is irreducible iff x^(2^63) == x (mod p) and gcd(x^(2^(63/q)) - x, p) == 1
+    for q in {3, 7}, i.e. for the exponents 21 and 9.
+    """
+    assert p.bit_length() - 1 == _DEG
+    x = 2  # the polynomial "x"
+    r = x
+    saved = {}
+    for i in range(1, _DEG + 1):
+        r = _poly_sq_mod(r, p)
+        if i in (9, 21):
+            saved[i] = r
+    if r != x:
+        return False
+    for e in (21, 9):
+        if _poly_gcd(p, saved[e] ^ x) != 1:
+            return False
+    return True
+
+
+def _sample_polynomial(rng):
+    """Sample a random irreducible polynomial of degree 63 from the CSPRNG.
+
+    The polynomial has its degree-63 and constant coefficients set; the 62
+    middle coefficients are random. About 1 in 63 candidates is irreducible.
+    """
+    while True:
+        v = int.from_bytes(rng.random_bytes(8), "little")
+        p = (1 << _DEG) | (v & ((1 << _DEG) - 1)) | 1
+        if _is_irreducible(p):
+            return p
+
+
+def _build_tables(p):
+    """Build the two rolling tables for P.
+
+    out_tbl[b] = poly(b) * x^504 mod P   (remove the byte leaving the window;
+                                          504 = 8 * (window_size - 1))
+    red_tbl[t] = poly(t) * x^63 mod P    (reduce the 8 bits shifted above bit 62)
+    """
+    out_tbl = [_poly_mod(b << (8 * (_WINDOW_SIZE - 1)), p) for b in range(256)]
+    red_tbl = [_poly_mod(t << _DEG, p) for t in range(256)]
+    return out_tbl, red_tbl
+
+
+def _derive(bytes key):
+    """Derive (aes_key, P, out_tbl, red_tbl) deterministically from a 256-bit key.
+
+    Frozen derivation order (changing it changes all cut points and thus breaks
+    deduplication against existing repos): first the 16-byte AES key, then P
+    (8 bytes per candidate until irreducible).
+    """
+    rng = CSPRNG(key)
+    aes_key = rng.random_bytes(16)
+    p = _sample_polynomial(rng)
+    out_tbl, red_tbl = _build_tables(p)
+    return aes_key, p, out_tbl, red_tbl
+
+
+cdef class ChunkerRabinAES:
+    """
+    Content-Defined Chunker, variable chunk sizes, UHF-then-PRF cut decision.
+
+    Rolling Rabin fingerprint (secret irreducible polynomial, 64-byte window)
+    post-processed with AES-128; the cut decision only sees the AES output,
+    so observed chunk boundaries do not leak usable information about the
+    chunking secret (see module docstring).
+    """
+    cdef uint64_t chunk_mask
+    cdef uint64_t mask_s, mask_l  # normalized chunking: strict / loose masks
+    cdef size_t normal_size       # chunk length at which we switch mask_s -> mask_l
+    cdef int nc_level             # normalized chunking level (0 = disabled)
+    cdef RA_CTX* ctx
+    cdef uint8_t* data
+    cdef object _fd
+    cdef int fh
+    cdef int done, eof
+    cdef size_t min_size, buf_size, remaining, position, last
+    cdef long long bytes_read, bytes_yielded
+    cdef readonly float chunking_time
+    cdef object file_reader
+    cdef size_t reader_block_size
+    cdef bint sparse
+
+    def __cinit__(self, bytes key, int chunk_min_exp, int chunk_max_exp, int hash_mask_bits, int nc_level=0, size_t normal_size=0, bint sparse=False):
+        cdef uint64_t c_out[256]
+        cdef uint64_t c_red[256]
+        cdef int i
+        self.ctx = NULL
+        self.data = NULL
+        min_size = 1 << chunk_min_exp
+        max_size = 1 << chunk_max_exp
+        assert max_size <= len(zeros)
+        assert min_size + 1 <= max_size, "too small max_size"
+        # the rolling window needs 64 bytes of in-chunk history at the first
+        # possible cut position (min_size), so the minimum chunk size must be
+        # at least the window size:
+        assert chunk_min_exp >= 6, "chunk_min_exp must be >= 6 (2^6 = 64 = window size)"
+
+        self.chunk_mask = (1ULL << hash_mask_bits) - 1
+        self.min_size = min_size
+        # Normalized chunking, identical structure to the fastcdc chunker (see there);
+        # the AES output is uniform, so contiguous low-bit masks are fine.
+        assert nc_level >= 0
+        assert hash_mask_bits - nc_level >= 1, "nc_level too large for hash_mask_bits"
+        assert hash_mask_bits + nc_level <= 48, "nc_level too large for hash_mask_bits"
+        self.nc_level = nc_level
+        if nc_level:
+            self.mask_s = (1ULL << (hash_mask_bits + nc_level)) - 1
+            self.mask_l = (1ULL << (hash_mask_bits - nc_level)) - 1
+            self.normal_size = normal_size if normal_size else ((1ULL << hash_mask_bits) - (1ULL << (hash_mask_bits - nc_level)))
+        else:
+            self.mask_s = self.chunk_mask
+            self.mask_l = self.chunk_mask
+            self.normal_size = 0
+
+        aes_key, p, out_tbl, red_tbl = _derive(key)
+        for i in range(256):
+            c_out[i] = out_tbl[i]
+            c_red[i] = red_tbl[i]
+        force_sw = os.environ.get("BORG_RABIN_AES_FORCE_EVP", "") not in ("", "0")
+        self.ctx = ra_new(c_out, c_red, aes_key, 1 if force_sw else 0)
+        if self.ctx == NULL:
+            raise MemoryError("Failed to set up rabin-aes kernel")
+
+        self.buf_size = max_size
+        self.data = <uint8_t*>malloc(self.buf_size)
+        if self.data == NULL:
+            raise MemoryError("Failed to allocate chunker buffer")
+        self.fh = -1
+        self.done = 0
+        self.eof = 0
+        self.remaining = 0
+        self.position = 0
+        self.last = 0
+        self.bytes_read = 0
+        self.bytes_yielded = 0
+        self._fd = None
+        self.chunking_time = 0.0
+        self.reader_block_size = 1024 * 1024
+        self.sparse = sparse
+
+    def __dealloc__(self):
+        if self.ctx != NULL:
+            ra_free(self.ctx)
+            self.ctx = NULL
+        if self.data != NULL:
+            free(self.data)
+            self.data = NULL
+
+    @property
+    def kernel(self):
+        """Which scan kernel this chunker uses: 'aes-arm64', 'aes-ni' or 'evp'."""
+        return (<bytes>ra_kind(self.ctx)).decode("ascii")
+
+    cdef int fill(self) except 0:
+        """Fill the chunker's buffer with more data."""
+        cdef ssize_t n
+        cdef object chunk
+
+        memmove(self.data, self.data + self.last, self.position + self.remaining - self.last)
+        self.position -= self.last
+        self.last = 0
+        n = self.buf_size - self.position - self.remaining
+
+        if self.eof or n == 0:
+            return 1
+
+        chunk = self.file_reader.read(n)
+        n = chunk.meta["size"]
+
+        if n > 0:
+            if chunk.meta["allocation"] == CH_DATA:
+                memcpy(self.data + self.position + self.remaining, <const unsigned char*>PyBytes_AsString(chunk.data), n)
+            else:
+                memset(self.data + self.position + self.remaining, 0, n)
+            self.remaining += n
+            self.bytes_read += n
+        else:
+            self.eof = 1
+        return 1
+
+    cdef object process(self) except *:
+        """Process the chunker's buffer and return the next chunk."""
+        cdef uint64_t digest = 0, mask, mask_s = self.mask_s, mask_l = self.mask_l
+        cdef int nc_level = self.nc_level
+        cdef size_t n, old_last, min_size = self.min_size
+        cdef size_t normal_size = self.normal_size, normal_pos, chunk_len, span, did
+        cdef int64_t r
+        cdef uint8_t* p
+        cdef uint8_t* stop
+
+        if self.done:
+            if self.bytes_read == self.bytes_yielded:
+                raise StopIteration
+            else:
+                raise Exception("chunkifier byte count mismatch")
+
+        # ensure at least min_size + 1 bytes are buffered, or we are at eof
+        while self.remaining < min_size + 1 and not self.eof:
+            self.fill()
+
+        # at eof with only a remainder (< min_size + 1): emit it as the final chunk
+        if self.eof and self.remaining < min_size + 1:
+            self.done = 1
+            if self.remaining:
+                old_last = self.last
+                self.position += self.remaining
+                self.last = self.position
+                n = self.last - old_last
+                self.remaining = 0
+                self.bytes_yielded += n
+                return memoryview((self.data + old_last)[:n])
+            else:
+                if self.bytes_read == self.bytes_yielded:
+                    raise StopIteration
+                else:
+                    raise Exception("chunkifier byte count mismatch")
+
+        # skip the sub-minimum region (no cut allowed below min_size), then scan.
+        # warm up the rolling digest over the 64 bytes preceding the first
+        # scan position (they are within the current chunk, since min_size >= 64).
+        self.position += min_size
+        self.remaining -= min_size
+        digest = ra_digest64(self.ctx, self.data + self.position - 64)
+
+        while True:
+            chunk_len = self.position - self.last
+            mask = mask_s if (nc_level and chunk_len < normal_size) else mask_l
+
+            if self.remaining == 0:
+                if self.eof:
+                    break  # cut at end of data
+                self.fill()
+                if self.remaining == 0:
+                    break  # buffer full -> chunk reached max_size -> forced cut
+                continue
+
+            p = self.data + self.position
+            stop = p + self.remaining
+            if nc_level and chunk_len < normal_size:
+                # do not scan past the strict->loose transition; re-evaluate the mask there
+                normal_pos = self.last + normal_size
+                if (self.data + normal_pos) < stop:
+                    stop = self.data + normal_pos
+
+            span = stop - p
+            r = ra_scan(self.ctx, p, span, &digest, mask)
+            if r == -2:
+                raise Exception("rabin-aes: OpenSSL EVP encryption failed")
+            if r >= 0:
+                did = <size_t>r + 1  # cut right after the position that matched
+                self.position += did
+                self.remaining -= did
+                break
+            else:
+                self.position += span
+                self.remaining -= span
+
+        old_last = self.last
+        self.last = self.position
+        n = self.last - old_last
+        self.bytes_yielded += n
+        return memoryview((self.data + old_last)[:n])
+
+    def chunkify(self, fd, fh=-1, fmap=None):
+        """
+        Cut a file into chunks.
+
+        :param fd: Python file object
+        :param fh: OS-level file handle (if available),
+                   defaults to -1 which means not to use OS-level fd.
+        :param fmap: a file map, same format as generated by sparsemap
+        """
+        self._fd = fd
+        self.fh = fh
+        self.file_reader = FileReader(fd=fd, fh=fh, read_size=self.reader_block_size, sparse=self.sparse, fmap=fmap)
+        self.done = 0
+        self.remaining = 0
+        self.bytes_read = 0
+        self.bytes_yielded = 0
+        self.position = 0
+        self.last = 0
+        self.eof = 0
+        return self
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        started_chunking = time.monotonic()
+        data = self.process()
+        got = len(data)
+        if zeros.startswith(data):
+            data = None
+            allocation = CH_ALLOC
+        else:
+            allocation = CH_DATA
+        self.chunking_time += time.monotonic() - started_chunking
+        return Chunk(data, size=got, allocation=allocation)
+
+
+def rabin_aes_get_polynomial(bytes key):
+    """Get the secret irreducible polynomial P derived from <key> (for tests)."""
+    aes_key, p, out_tbl, red_tbl = _derive(key)
+    return p
+
+
+def rabin_aes_get_tables(bytes key):
+    """Get the (out_tbl, red_tbl) rolling tables derived from <key> (for tests)."""
+    aes_key, p, out_tbl, red_tbl = _derive(key)
+    return out_tbl, red_tbl
@@ -0,0 +1,391 @@
+/* rabin-aes chunker scan kernel, see rabin_aes_impl.h and rabin_aes.pyx. */
+
+#include <stdlib.h>
+#include <string.h>
+
+#include <openssl/evp.h>
+
+#include "rabin_aes_impl.h"
+
+#define M55 ((((uint64_t)1) << 55) - 1)
+
+struct RA_CTX {
+    uint64_t out_tbl[256];
+    uint64_t red_tbl[256];
+    uint8_t rk[11][16]; /* AES-128 round keys, for the hardware paths */
+    EVP_CIPHER_CTX *evp;
+    int use_hw;
+};
+
+/* --- portable helpers ------------------------------------------------- */
+
+static inline void store_le64(uint8_t *b, uint64_t v)
+{
+    for (int j = 0; j < 8; j++)
+        b[j] = (uint8_t)(v >> (8 * j));
+}
+
+static inline uint64_t load_le64(const uint8_t *b)
+{
+    uint64_t v = 0;
+    for (int j = 0; j < 8; j++)
+        v |= ((uint64_t)b[j]) << (8 * j);
+    return v;
+}
+
+/* Advance the digest by one byte: remove the byte leaving the 64-byte window
+ * (its coefficient is x^504), multiply by x^8, add the incoming byte, reduce.
+ * The digest is the mod-P remainder, so it always fits in 63 bits. */
+static inline uint64_t ra_roll(const RA_CTX *c, uint64_t d, uint8_t byte_out, uint8_t byte_in)
+{
+    d ^= c->out_tbl[byte_out];
+    return (((d & M55) << 8) | byte_in) ^ c->red_tbl[d >> 55];
+}
+
+uint64_t ra_digest64(const RA_CTX *c, const uint8_t *q)
+{
+    uint64_t d = 0;
+    for (int j = 0; j < 64; j++)
+        d = (((d & M55) << 8) | q[j]) ^ c->red_tbl[d >> 55];
+    return d;
+}
+
+/* --- AES-128 key expansion (encryption-only, standard FIPS-197) -------- */
+
+static const uint8_t sbox[256] = {
+    0x63, 0x7c, 0x77, 0x7b, 0xf2, 0x6b, 0x6f, 0xc5, 0x30, 0x01, 0x67, 0x2b, 0xfe, 0xd7, 0xab, 0x76,
+    0xca, 0x82, 0xc9, 0x7d, 0xfa, 0x59, 0x47, 0xf0, 0xad, 0xd4, 0xa2, 0xaf, 0x9c, 0xa4, 0x72, 0xc0,
+    0xb7, 0xfd, 0x93, 0x26, 0x36, 0x3f, 0xf7, 0xcc, 0x34, 0xa5, 0xe5, 0xf1, 0x71, 0xd8, 0x31, 0x15,
+    0x04, 0xc7, 0x23, 0xc3, 0x18, 0x96, 0x05, 0x9a, 0x07, 0x12, 0x80, 0xe2, 0xeb, 0x27, 0xb2, 0x75,
+    0x09, 0x83, 0x2c, 0x1a, 0x1b, 0x6e, 0x5a, 0xa0, 0x52, 0x3b, 0xd6, 0xb3, 0x29, 0xe3, 0x2f, 0x84,
+    0x53, 0xd1, 0x00, 0xed, 0x20, 0xfc, 0xb1, 0x5b, 0x6a, 0xcb, 0xbe, 0x39, 0x4a, 0x4c, 0x58, 0xcf,
+    0xd0, 0xef, 0xaa, 0xfb, 0x43, 0x4d, 0x33, 0x85, 0x45, 0xf9, 0x02, 0x7f, 0x50, 0x3c, 0x9f, 0xa8,
+    0x51, 0xa3, 0x40, 0x8f, 0x92, 0x9d, 0x38, 0xf5, 0xbc, 0xb6, 0xda, 0x21, 0x10, 0xff, 0xf3, 0xd2,
+    0xcd, 0x0c, 0x13, 0xec, 0x5f, 0x97, 0x44, 0x17, 0xc4, 0xa7, 0x7e, 0x3d, 0x64, 0x5d, 0x19, 0x73,
+    0x60, 0x81, 0x4f, 0xdc, 0x22, 0x2a, 0x90, 0x88, 0x46, 0xee, 0xb8, 0x14, 0xde, 0x5e, 0x0b, 0xdb,
+    0xe0, 0x32, 0x3a, 0x0a, 0x49, 0x06, 0x24, 0x5c, 0xc2, 0xd3, 0xac, 0x62, 0x91, 0x95, 0xe4, 0x79,
+    0xe7, 0xc8, 0x37, 0x6d, 0x8d, 0xd5, 0x4e, 0xa9, 0x6c, 0x56, 0xf4, 0xea, 0x65, 0x7a, 0xae, 0x08,
+    0xba, 0x78, 0x25, 0x2e, 0x1c, 0xa6, 0xb4, 0xc6, 0xe8, 0xdd, 0x74, 0x1f, 0x4b, 0xbd, 0x8b, 0x8a,
+    0x70, 0x3e, 0xb5, 0x66, 0x48, 0x03, 0xf6, 0x0e, 0x61, 0x35, 0x57, 0xb9, 0x86, 0xc1, 0x1d, 0x9e,
+    0xe1, 0xf8, 0x98, 0x11, 0x69, 0xd9, 0x8e, 0x94, 0x9b, 0x1e, 0x87, 0xe9, 0xce, 0x55, 0x28, 0xdf,
+    0x8c, 0xa1, 0x89, 0x0d, 0xbf, 0xe6, 0x42, 0x68, 0x41, 0x99, 0x2d, 0x0f, 0xb0, 0x54, 0xbb, 0x16,
+};
+
+static void aes128_expand(const uint8_t key[16], uint8_t rk[11][16])
+{
+    uint8_t rcon = 1;
+    memcpy(rk[0], key, 16);
+    for (int i = 1; i <= 10; i++) {
+        rk[i][0] = (uint8_t)(rk[i - 1][0] ^ sbox[rk[i - 1][13]] ^ rcon);
+        rk[i][1] = (uint8_t)(rk[i - 1][1] ^ sbox[rk[i - 1][14]]);
+        rk[i][2] = (uint8_t)(rk[i - 1][2] ^ sbox[rk[i - 1][15]]);
+        rk[i][3] = (uint8_t)(rk[i - 1][3] ^ sbox[rk[i - 1][12]]);
+        for (int j = 4; j < 16; j++)
+            rk[i][j] = (uint8_t)(rk[i - 1][j] ^ rk[i][j - 4]);
+        rcon = (uint8_t)((rcon << 1) ^ ((rcon >> 7) * 0x1b));
+    }
+}
+
+/* --- portable path: batched AES via OpenSSL EVP ------------------------ */
+
+/* sub-batch size (blocks); 512 * 16 B = 8 KiB in/out buffers on the stack */
+#define RA_SB 512
+
+static int64_t ra_scan_evp(RA_CTX *c, const uint8_t *p, size_t n, uint64_t *digest, uint64_t mask)
+{
+    uint8_t inb[RA_SB * 16];
+    uint8_t outb[RA_SB * 16];
+    uint64_t dig[RA_SB];
+    uint64_t d = *digest;
+    size_t base = 0;
+    int outlen;
+
+    memset(inb, 0, sizeof(inb)); /* bytes 8..15 of each block stay zero */
+    while (base < n) {
+        const uint8_t *q = p + base;
+        const uint8_t *q_out = q - 64;
+        size_t m = n - base;
+        if (m > RA_SB)
+            m = RA_SB;
+        for (size_t i = 0; i < m; i++) {
+            d = ra_roll(c, d, q_out[i], q[i]);
+            dig[i] = d;
+            store_le64(inb + i * 16, d);
+        }
+        outlen = 0;
+        if (!EVP_EncryptUpdate(c->evp, outb, &outlen, inb, (int)(m * 16)) || outlen != (int)(m * 16))
+            return -2; /* OpenSSL failure; caller raises */
+        for (size_t i = 0; i < m; i++) {
+            if ((load_le64(outb + i * 16) & mask) == 0) {
+                *digest = dig[i];
+                return (int64_t)(base + i);
+            }
+        }
+        base += m;
+    }
+    *digest = d;
+    return -1;
+}
+
+/* --- hardware path: arm64 crypto extension ----------------------------- */
+
+#if defined(__aarch64__) && defined(__ARM_FEATURE_AES)
+#define RA_HAVE_HW 1
+#define RA_KIND_HW "aes-arm64"
+
+#include <arm_neon.h>
+#if defined(__linux__)
+#include <sys/auxv.h>
+#ifndef HWCAP_AES
+#define HWCAP_AES (1 << 3)
+#endif
+#endif
+
+static int ra_hw_available(void)
+{
+#if defined(__linux__)
+    return (getauxval(AT_HWCAP) & HWCAP_AES) != 0;
+#else
+    return 1; /* compiler targeted +aes; on Apple Silicon it is always there */
+#endif
+}
+
+static inline uint8x16_t ra_aes1_neon(const uint8x16_t k[11], uint8x16_t b)
+{
+    for (int r = 0; r < 9; r++)
+        b = vaesmcq_u8(vaeseq_u8(b, k[r]));
+    return veorq_u8(vaeseq_u8(b, k[9]), k[10]);
+}
+
+static int64_t ra_scan_hw(RA_CTX *c, const uint8_t *p, size_t n, uint64_t *digest, uint64_t mask)
+{
+    uint8x16_t k[11];
+    uint64_t d = *digest;
+    const uint8_t *q_out = p - 64;
+    size_t i = 0;
+
+    for (int j = 0; j < 11; j++)
+        k[j] = vld1q_u8(c->rk[j]);
+
+    while (i + 8 <= n) {
+        uint64_t dg[8];
+        uint8x16_t b0, b1, b2, b3, b4, b5, b6, b7;
+
+        /* serial Rabin chain for the next 8 positions */
+        d = ra_roll(c, d, q_out[i + 0], p[i + 0]); dg[0] = d;
+        d = ra_roll(c, d, q_out[i + 1], p[i + 1]); dg[1] = d;
+        d = ra_roll(c, d, q_out[i + 2], p[i + 2]); dg[2] = d;
+        d = ra_roll(c, d, q_out[i + 3], p[i + 3]); dg[3] = d;
+        d = ra_roll(c, d, q_out[i + 4], p[i + 4]); dg[4] = d;
+        d = ra_roll(c, d, q_out[i + 5], p[i + 5]); dg[5] = d;
+        d = ra_roll(c, d, q_out[i + 6], p[i + 6]); dg[6] = d;
+        d = ra_roll(c, d, q_out[i + 7], p[i + 7]); dg[7] = d;
+
+        /* 8 independent AES-128 encryptions, interleaved to fill the AES pipes */
+        b0 = vreinterpretq_u8_u64(vcombine_u64(vcreate_u64(dg[0]), vcreate_u64(0)));
+        b1 = vreinterpretq_u8_u64(vcombine_u64(vcreate_u64(dg[1]), vcreate_u64(0)));
+        b2 = vreinterpretq_u8_u64(vcombine_u64(vcreate_u64(dg[2]), vcreate_u64(0)));
+        b3 = vreinterpretq_u8_u64(vcombine_u64(vcreate_u64(dg[3]), vcreate_u64(0)));
+        b4 = vreinterpretq_u8_u64(vcombine_u64(vcreate_u64(dg[4]), vcreate_u64(0)));
+        b5 = vreinterpretq_u8_u64(vcombine_u64(vcreate_u64(dg[5]), vcreate_u64(0)));
+        b6 = vreinterpretq_u8_u64(vcombine_u64(vcreate_u64(dg[6]), vcreate_u64(0)));
+        b7 = vreinterpretq_u8_u64(vcombine_u64(vcreate_u64(dg[7]), vcreate_u64(0)));
+        for (int r = 0; r < 9; r++) {
+            b0 = vaesmcq_u8(vaeseq_u8(b0, k[r]));
+            b1 = vaesmcq_u8(vaeseq_u8(b1, k[r]));
+            b2 = vaesmcq_u8(vaeseq_u8(b2, k[r]));
+            b3 = vaesmcq_u8(vaeseq_u8(b3, k[r]));
+            b4 = vaesmcq_u8(vaeseq_u8(b4, k[r]));
+            b5 = vaesmcq_u8(vaeseq_u8(b5, k[r]));
+            b6 = vaesmcq_u8(vaeseq_u8(b6, k[r]));
+            b7 = vaesmcq_u8(vaeseq_u8(b7, k[r]));
+        }
+        b0 = veorq_u8(vaeseq_u8(b0, k[9]), k[10]);
+        b1 = veorq_u8(vaeseq_u8(b1, k[9]), k[10]);
+        b2 = veorq_u8(vaeseq_u8(b2, k[9]), k[10]);
+        b3 = veorq_u8(vaeseq_u8(b3, k[9]), k[10]);
+        b4 = veorq_u8(vaeseq_u8(b4, k[9]), k[10]);
+        b5 = veorq_u8(vaeseq_u8(b5, k[9]), k[10]);
+        b6 = veorq_u8(vaeseq_u8(b6, k[9]), k[10]);
+        b7 = veorq_u8(vaeseq_u8(b7, k[9]), k[10]);
+
+        {
+            uint64_t c0 = vgetq_lane_u64(vreinterpretq_u64_u8(b0), 0);
+            uint64_t c1 = vgetq_lane_u64(vreinterpretq_u64_u8(b1), 0);
+            uint64_t c2 = vgetq_lane_u64(vreinterpretq_u64_u8(b2), 0);
+            uint64_t c3 = vgetq_lane_u64(vreinterpretq_u64_u8(b3), 0);
+            uint64_t c4 = vgetq_lane_u64(vreinterpretq_u64_u8(b4), 0);
+            uint64_t c5 = vgetq_lane_u64(vreinterpretq_u64_u8(b5), 0);
+            uint64_t c6 = vgetq_lane_u64(vreinterpretq_u64_u8(b6), 0);
+            uint64_t c7 = vgetq_lane_u64(vreinterpretq_u64_u8(b7), 0);
+            if ((((c0 & mask) == 0) | ((c1 & mask) == 0) | ((c2 & mask) == 0) | ((c3 & mask) == 0) |
+                 ((c4 & mask) == 0) | ((c5 & mask) == 0) | ((c6 & mask) == 0) | ((c7 & mask) == 0))) {
+                uint64_t cs[8] = {c0, c1, c2, c3, c4, c5, c6, c7};
+                for (int j = 0; j < 8; j++) {
+                    if ((cs[j] & mask) == 0) {
+                        *digest = dg[j];
+                        return (int64_t)(i + j);
+                    }
+                }
+            }
+        }
+        i += 8;
+    }
+    /* tail: single blocks */
+    while (i < n) {
+        uint8x16_t b;
+        d = ra_roll(c, d, q_out[i], p[i]);
+        b = vreinterpretq_u8_u64(vcombine_u64(vcreate_u64(d), vcreate_u64(0)));
+        b = ra_aes1_neon(k, b);
+        if ((vgetq_lane_u64(vreinterpretq_u64_u8(b), 0) & mask) == 0) {
+            *digest = d;
+            return (int64_t)i;
+        }
+        i++;
+    }
+    *digest = d;
+    return -1;
+}
+
+/* --- hardware path: x86-64 AES-NI -------------------------------------- */
+
+#elif (defined(__x86_64__) || defined(_M_X64)) && (defined(__GNUC__) || defined(__clang__))
+#define RA_HAVE_HW 1
+#define RA_KIND_HW "aes-ni"
+
+#include <immintrin.h>
+
+static int ra_hw_available(void)
+{
+    return __builtin_cpu_supports("aes") && __builtin_cpu_supports("sse2");
+}
+
+__attribute__((target("aes,sse2"))) static int64_t
+ra_scan_hw(RA_CTX *c, const uint8_t *p, size_t n, uint64_t *digest, uint64_t mask)
+{
+    __m128i k[11];
+    uint64_t d = *digest;
+    const uint8_t *q_out = p - 64;
+    size_t i = 0;
+
+    for (int j = 0; j < 11; j++)
+        k[j] = _mm_loadu_si128((const __m128i *)c->rk[j]);
+
+    while (i + 4 <= n) {
+        uint64_t dg[4];
+        __m128i b0, b1, b2, b3;
+
+        d = ra_roll(c, d, q_out[i + 0], p[i + 0]); dg[0] = d;
+        d = ra_roll(c, d, q_out[i + 1], p[i + 1]); dg[1] = d;
+        d = ra_roll(c, d, q_out[i + 2], p[i + 2]); dg[2] = d;
+        d = ra_roll(c, d, q_out[i + 3], p[i + 3]); dg[3] = d;
+
+        b0 = _mm_xor_si128(_mm_set_epi64x(0, (long long)dg[0]), k[0]);
+        b1 = _mm_xor_si128(_mm_set_epi64x(0, (long long)dg[1]), k[0]);
+        b2 = _mm_xor_si128(_mm_set_epi64x(0, (long long)dg[2]), k[0]);
+        b3 = _mm_xor_si128(_mm_set_epi64x(0, (long long)dg[3]), k[0]);
+        for (int r = 1; r < 10; r++) {
+            b0 = _mm_aesenc_si128(b0, k[r]);
+            b1 = _mm_aesenc_si128(b1, k[r]);
+            b2 = _mm_aesenc_si128(b2, k[r]);
+            b3 = _mm_aesenc_si128(b3, k[r]);
+        }
+        b0 = _mm_aesenclast_si128(b0, k[10]);
+        b1 = _mm_aesenclast_si128(b1, k[10]);
+        b2 = _mm_aesenclast_si128(b2, k[10]);
+        b3 = _mm_aesenclast_si128(b3, k[10]);
+
+        {
+            uint64_t cs[4];
+            cs[0] = (uint64_t)_mm_cvtsi128_si64(b0);
+            cs[1] = (uint64_t)_mm_cvtsi128_si64(b1);
+            cs[2] = (uint64_t)_mm_cvtsi128_si64(b2);
+            cs[3] = (uint64_t)_mm_cvtsi128_si64(b3);
+            for (int j = 0; j < 4; j++) {
+                if ((cs[j] & mask) == 0) {
+                    *digest = dg[j];
+                    return (int64_t)(i + j);
+                }
+            }
+        }
+        i += 4;
+    }
+    while (i < n) {
+        __m128i b;
+        d = ra_roll(c, d, q_out[i], p[i]);
+        b = _mm_xor_si128(_mm_set_epi64x(0, (long long)d), k[0]);
+        for (int r = 1; r < 10; r++)
+            b = _mm_aesenc_si128(b, k[r]);
+        b = _mm_aesenclast_si128(b, k[10]);
+        if (((uint64_t)_mm_cvtsi128_si64(b) & mask) == 0) {
+            *digest = d;
+            return (int64_t)i;
+        }
+        i++;
+    }
+    *digest = d;
+    return -1;
+}
+
+#else
+#define RA_HAVE_HW 0
+#endif
+
+/* --- context management and dispatch ----------------------------------- */
+
+RA_CTX *ra_new(const uint64_t out_tbl[256], const uint64_t red_tbl[256],
+               const uint8_t aes_key[16], int force_sw)
+{
+    RA_CTX *c = calloc(1, sizeof(RA_CTX));
+    if (c == NULL)
+        return NULL;
+    memcpy(c->out_tbl, out_tbl, sizeof(c->out_tbl));
+    memcpy(c->red_tbl, red_tbl, sizeof(c->red_tbl));
+    aes128_expand(aes_key, c->rk);
+#if RA_HAVE_HW
+    c->use_hw = !force_sw && ra_hw_available();
+#else
+    (void)force_sw;
+    c->use_hw = 0;
+#endif
+    c->evp = EVP_CIPHER_CTX_new();
+    if (c->evp == NULL) {
+        free(c);
+        return NULL;
+    }
+    if (!EVP_EncryptInit_ex(c->evp, EVP_aes_128_ecb(), NULL, aes_key, NULL) ||
+        !EVP_CIPHER_CTX_set_padding(c->evp, 0)) {
+        EVP_CIPHER_CTX_free(c->evp);
+        free(c);
+        return NULL;
+    }
+    return c;
+}
+
+void ra_free(RA_CTX *c)
+{
+    if (c != NULL) {
+        if (c->evp != NULL)
+            EVP_CIPHER_CTX_free(c->evp);
+        free(c);
+    }
+}
+
+const char *ra_kind(const RA_CTX *c)
+{
+#if RA_HAVE_HW
+    if (c->use_hw)
+        return RA_KIND_HW;
+#endif
+    (void)c;
+    return "evp";
+}
+
+int64_t ra_scan(RA_CTX *c, const uint8_t *p, size_t n, uint64_t *digest, uint64_t mask)
+{
+#if RA_HAVE_HW
+    if (c->use_hw)
+        return ra_scan_hw(c, p, n, digest, mask);
+#endif
+    return ra_scan_evp(c, p, n, digest, mask);
+}
@@ -0,0 +1,50 @@
+/* rabin-aes chunker scan kernel: rolling Rabin fingerprint (UHF) + AES-128 (PRF).
+ *
+ * See rabin_aes.pyx for the scheme description. This C layer only implements the
+ * performance-critical inner scan: roll the Rabin digest one byte at a time,
+ * encrypt each digest with AES-128, and cut where the ciphertext's low bits
+ * (little-endian uint64 of the first 8 ciphertext bytes) are all zero under mask.
+ *
+ * Two code paths produce bit-identical results:
+ *  - a portable path batching digests through OpenSSL EVP AES-128-ECB,
+ *  - a hardware path using AES instructions directly (arm64 with the crypto
+ *    extension, e.g. Apple Silicon; x86-64 with AES-NI), which interleaves the
+ *    serial Rabin chain with pipelined AES so the AES work is (mostly) hidden
+ *    behind the Rabin table-lookup latency chain.
+ */
+
+#ifndef BORG_RABIN_AES_IMPL_H
+#define BORG_RABIN_AES_IMPL_H
+
+#include <stdint.h>
+#include <stddef.h>
+
+typedef struct RA_CTX RA_CTX;
+
+/* Create a kernel context.
+ * out_tbl[b] = poly(b) * x^504 mod P  (removal of the byte leaving the 64-byte window)
+ * red_tbl[t] = poly(t) * x^63  mod P  (reduction of the 8 bits shifted above bit 62)
+ * aes_key: 16 bytes (AES-128).
+ * force_sw: nonzero forces the portable OpenSSL path (for tests/benchmarks).
+ * Returns NULL on allocation/OpenSSL failure. */
+RA_CTX *ra_new(const uint64_t out_tbl[256], const uint64_t red_tbl[256],
+               const uint8_t aes_key[16], int force_sw);
+
+void ra_free(RA_CTX *ctx);
+
+/* Which path this context uses: "aes-arm64", "aes-ni" or "evp". */
+const char *ra_kind(const RA_CTX *ctx);
+
+/* Full (non-rolling) Rabin digest of the 64 bytes at q: the window warm-up at
+ * the start of each chunk scan. */
+uint64_t ra_digest64(const RA_CTX *ctx, const uint8_t *q);
+
+/* Scan up to n positions. For position i (0 <= i < n), the rolling digest is
+ * advanced to cover the 64-byte window p[i-63..i] (reading the outgoing byte
+ * at p[i-64], which the caller must guarantee to be accessible) and the cut
+ * condition is evaluated on AES(digest).
+ * Returns the first i whose cut condition matched (digest is left at position i),
+ * or -1 if none matched (digest is left at position n-1). */
+int64_t ra_scan(RA_CTX *ctx, const uint8_t *p, size_t n, uint64_t *digest, uint64_t mask);
+
+#endif
@@ -133,6 +133,7 @@ MAX_SEGMENT_DIR_INDEX = 2**32 - 1
 CH_BUZHASH = "buzhash"
 CH_BUZHASH64 = "buzhash64"
 CH_FASTCDC = "fastcdc"
+CH_RABIN_AES = "rabin-aes"
 CH_FIXED = "fixed"
 CH_FAIL = "fail"

@@ -150,6 +151,8 @@ NC_LEVEL = 2  # FastCDC-style normalized chunking: tightens chunk-size distribut
 FASTCDC_PARAMS = (CH_FASTCDC, CHUNK_MIN_EXP, CHUNK_MAX_EXP, HASH_MASK_BITS, NC_LEVEL)
 BUZHASH_PARAMS = (CH_BUZHASH, CHUNK_MIN_EXP, CHUNK_MAX_EXP, HASH_MASK_BITS, HASH_WINDOW_SIZE)
 BUZHASH64_PARAMS = (CH_BUZHASH64, CHUNK_MIN_EXP, CHUNK_MAX_EXP, HASH_MASK_BITS, HASH_WINDOW_SIZE, NC_LEVEL)
+# rabin-aes has a fixed 64-byte window, so it has no window_size parameter either.
+RABIN_AES_PARAMS = (CH_RABIN_AES, CHUNK_MIN_EXP, CHUNK_MAX_EXP, HASH_MASK_BITS, NC_LEVEL)
 CHUNKER_PARAMS = FASTCDC_PARAMS  # the default chunker for file content data

 # chunker params for the items metadata stream, finer granularity
@@ -358,6 +358,32 @@ def ChunkerParams(s):
                "required: 0 <= nc_level and 1 <= chunk_mask - nc_level and chunk_mask + nc_level <= 48"
            )
        return CH_FASTCDC, chunk_min, chunk_max, chunk_mask, nc_level
+    if algo == CH_RABIN_AES:
+        # rabin-aes, chunk_min, chunk_max, chunk_mask, nc_level
+        # rabin-aes has a fixed 64-byte window, so there is no window_size field.
+        # nc_level is required; use nc_level 0 to disable normalized chunking.
+        if count != 5:
+            raise ArgumentTypeError(
+                "rabin-aes chunker params must be: rabin-aes,chunk_min,chunk_max,chunk_mask,nc_level"
+            )
+        chunk_min, chunk_max, chunk_mask = (int(p) for p in params[1:4])
+        nc_level = int(params[4])
+        if not (chunk_min <= chunk_mask <= chunk_max):
+            raise ArgumentTypeError("required: chunk_min <= chunk_mask <= chunk_max")
+        if chunk_min < 6:
+            # the rolling window needs 64 bytes of in-chunk history at the first possible
+            # cut position, so the min. chunk size must be at least the 64-byte window size
+            # (this also matches the general lower bound, see comment in 'fixed' algo check).
+            raise ArgumentTypeError("min. chunk size exponent must not be less than 6 (2^6 = 64B min. chunk size)")
+        if chunk_max > 23:
+            raise ArgumentTypeError("max. chunk size exponent must not be more than 23 (2^23 = 8MiB max. chunk size)")
+        if chunk_min >= chunk_max:
+            raise ArgumentTypeError("required: chunk_min < chunk_max")
+        if not (0 <= nc_level and chunk_mask - nc_level >= 1 and chunk_mask + nc_level <= 48):
+            raise ArgumentTypeError(
+                "required: 0 <= nc_level and 1 <= chunk_mask - nc_level and chunk_mask + nc_level <= 48"
+            )
+        return CH_RABIN_AES, chunk_min, chunk_max, chunk_mask, nc_level
    # this must stay last as it deals with old-style compat mode (no algorithm, 4 numeric params, buzhash);
    # the isdigit check keeps misspelled/incomplete algo specs out of the compat branch.
    if (algo == CH_BUZHASH and count == 5) or (
@@ -0,0 +1,211 @@
+from hashlib import sha256
+from io import BytesIO
+import os
+import random
+
+import pytest
+
+from . import cf, cf_expand
+from ...chunkers import ChunkerRabinAES, get_chunker
+from ...chunkers.rabin_aes import rabin_aes_get_polynomial, rabin_aes_get_tables, _is_irreducible
+from ...constants import *  # NOQA
+from ...helpers import hex_to_bin
+
+
+# from os.urandom(32)
+key0 = hex_to_bin("ad9f89095817f0566337dc9ee292fcd59b70f054a8200151f1df5f21704824da")
+key1 = hex_to_bin("f1088c7e9e6ae83557ad1558ff36c44a369ea719d1081c29684f52ffccb72cb8")
+
+
+def H(data):
+    return sha256(data).digest()
+
+
+def test_chunkpoints_rabin_aes_unchanged():
+    def twist(size):
+        x = 1
+        a = bytearray(size)
+        for i in range(size):
+            x = (x * 1103515245 + 12345) & 0x7FFFFFFF
+            a[i] = x & 0xFF
+        return a
+
+    data = twist(100000)
+
+    runs = []
+    for nc_level in (0, 2, 3):
+        for minexp in (6, 7, 11, 12):  # rabin-aes requires minexp >= 6 (64-byte window)
+            for maxexp in (15, 17):
+                if minexp >= maxexp:
+                    continue
+                for maskbits in (7, 10, 12):
+                    if maskbits - nc_level < 1:  # nc_level needs room below the base mask bits
+                        continue
+                    for key in (key0, key1):
+                        fh = BytesIO(data)
+                        chunker = ChunkerRabinAES(key, minexp, maxexp, maskbits, nc_level)
+                        chunks = [H(c) for c in cf(chunker.chunkify(fh, -1))]
+                        runs.append(H(b"".join(chunks)))
+
+    # The "correct" hash below matches the existing chunker behavior.
+    # Future chunker optimizations must not change this, or existing repos will bloat.
+    overall_hash = H(b"".join(runs))
+    print(overall_hash.hex())
+    assert overall_hash == hex_to_bin("83c89b4d89f7b653e53efd6b42607da01dccdb44d9c73d58f7fe03f30b8e785d")
+
+
+def test_rabin_aes_kernels_identical():
+    # the OpenSSL EVP batch path and the AES hardware instruction path (if available
+    # on this platform) must produce identical cut points.
+    data = os.urandom(4 * 1024 * 1024)
+
+    def sizes(chunker):
+        return [c.meta["size"] for c in chunker.chunkify(BytesIO(data))]
+
+    default = ChunkerRabinAES(key0, 10, 16, 14, 2)
+    sizes_default = sizes(default)
+    os.environ["BORG_RABIN_AES_FORCE_EVP"] = "1"
+    try:
+        forced = ChunkerRabinAES(key0, 10, 16, 14, 2)
+        assert forced.kernel == "evp"
+        sizes_evp = sizes(forced)
+    finally:
+        del os.environ["BORG_RABIN_AES_FORCE_EVP"]
+    assert sizes_default == sizes_evp
+    # whatever kernel was selected by default, it must be a known one
+    assert default.kernel in ("aes-arm64", "aes-ni", "evp")
+
+
+def test_rabin_aes_chunksize_distribution():
+    data = os.urandom(1048576)
+    min_exp, max_exp, mask, nc_level = 10, 16, 14, 2  # chunk size target 16 KiB, clip at 1 KiB and 64 KiB
+    chunker = ChunkerRabinAES(key0, min_exp, max_exp, mask, nc_level)
+    f = BytesIO(data)
+    chunks = cf(chunker.chunkify(f))
+    del chunks[-1]  # get rid of the last chunk, it can be smaller than 2**min_exp
+    chunk_sizes = [len(chunk) for chunk in chunks]
+    chunks_count = len(chunks)
+    min_chunksize_observed = min(chunk_sizes)
+    max_chunksize_observed = max(chunk_sizes)
+    min_count = sum(int(size == 2**min_exp) for size in chunk_sizes)
+    max_count = sum(int(size == 2**max_exp) for size in chunk_sizes)
+    print(
+        f"count: {chunks_count} min: {min_chunksize_observed} max: {max_chunksize_observed} "
+        f"min count: {min_count} max count: {max_count}"
+    )
+    # usually there will about 64 chunks
+    assert 32 < chunks_count < 128
+    # chunks always must be between min and max (clipping must work):
+    assert min_chunksize_observed >= 2**min_exp
+    assert max_chunksize_observed <= 2**max_exp
+    # most chunks should be cut due to the hash triggering, not due to clipping at min/max size:
+    assert min_count < 10
+    assert max_count < 10
+
+
+def test_rabin_aes_shift_resilience():
+    # content-defined cuts must survive a prefix insertion (this also validates that the
+    # rolling digest update and the per-chunk window warm-up agree with each other).
+    data = os.urandom(4 * 1024 * 1024)
+
+    def chunk_hashes(data):
+        chunker = ChunkerRabinAES(key0, 10, 16, 14, 2)
+        return [H(c) for c in cf_expand(chunker.chunkify(BytesIO(data)))]
+
+    h1 = set(chunk_hashes(data))
+    h2 = chunk_hashes(b"PREFIX_SHIFTS_EVERYTHING" + data)
+    survived = sum(1 for h in h2 if h in h1)
+    assert survived / len(h2) > 0.9
+
+
+def test_rabin_aes_polynomial():
+    # P must be a degree-63 irreducible polynomial, deterministically derived from the key
+    p0 = rabin_aes_get_polynomial(key0)
+    assert p0.bit_length() - 1 == 63
+    assert p0 & 1  # constant term set (otherwise divisible by x)
+    assert _is_irreducible(p0)
+    assert p0 == rabin_aes_get_polynomial(key0)
+    p1 = rabin_aes_get_polynomial(key1)
+    assert p0 != p1
+
+    out_tbl, red_tbl = rabin_aes_get_tables(key0)
+    assert len(out_tbl) == 256 and len(red_tbl) == 256
+    for v in out_tbl + red_tbl:
+        assert isinstance(v, int)
+        assert 0 <= v < 2**63  # all table entries are mod-P remainders
+    # spot check the tables against the polynomial: entry b is (b << shift) mod P
+    for b in (0, 1, 2, 128, 255):
+        v = b << 504
+        while v.bit_length() - 1 >= 63:
+            v ^= p0 << (v.bit_length() - 1 - 63)
+        assert out_tbl[b] == v
+
+
+def test_rabin_aes_get_chunker():
+    # without a key, get_chunker uses an all-zero key; chunking must still work and be deterministic
+    data = os.urandom(2 * 1024 * 1024)
+    a = cf_expand(get_chunker(*RABIN_AES_PARAMS, key=None).chunkify(BytesIO(data)))
+    b = cf_expand(get_chunker("rabin-aes", 19, 23, 21, 2, key=None).chunkify(BytesIO(data)))
+    assert a == b
+    assert b"".join(a) == data
+
+
+def test_rabin_aes_params_parsing():
+    from argparse import ArgumentTypeError
+
+    from ...helpers import ChunkerParams
+
+    # rabin-aes, chunk_min, chunk_max, chunk_mask, nc_level (no window field)
+    assert ChunkerParams("rabin-aes,19,23,21,2") == (CH_RABIN_AES, 19, 23, 21, 2)
+    assert ChunkerParams("rabin-aes,10,23,16,0") == (CH_RABIN_AES, 10, 23, 16, 0)
+    # a 6-field (buzhash64-style, with window) rabin-aes must be rejected
+    with pytest.raises(ArgumentTypeError):
+        ChunkerParams("rabin-aes,19,23,21,4095,2")
+    # a 4-field rabin-aes (missing nc_level) must be rejected, not fall into old-style compat mode
+    with pytest.raises(ArgumentTypeError):
+        ChunkerParams("rabin-aes,19,23,21")
+    # nc_level out of range (chunk_mask - nc_level < 1)
+    with pytest.raises(ArgumentTypeError):
+        ChunkerParams("rabin-aes,19,23,21,21")
+    # chunk_min <= chunk_mask <= chunk_max violated
+    with pytest.raises(ArgumentTypeError):
+        ChunkerParams("rabin-aes,19,23,24,2")
+    # chunk_min < 6 is not allowed (the 64-byte window needs 64 bytes of in-chunk history)
+    with pytest.raises(ArgumentTypeError):
+        ChunkerParams("rabin-aes,5,23,21,2")
+
+
+@pytest.mark.skipif("BORG_TESTS_SLOW" not in os.environ, reason="slow tests not enabled, use BORG_TESTS_SLOW=1")
+@pytest.mark.parametrize("worker", range(os.cpu_count() or 1))
+def test_fuzz_rabin_aes(worker):
+    # Fuzz rabin-aes with random and uniform data of misc. sizes and misc keys.
+    def rnd_key():
+        return os.urandom(32)
+
+    # decompose RABIN_AES_PARAMS = (algo, min_exp, max_exp, mask_bits, nc_level)
+    algo, min_exp, max_exp, mask_bits, nc_level = RABIN_AES_PARAMS
+    assert algo == CH_RABIN_AES
+
+    keys = [b"\0" * 32] + [rnd_key() for _ in range(10)]
+    sizes = [random.randint(1, 4 * 1024 * 1024) for _ in range(50)]
+
+    for key in keys:
+        chunker = ChunkerRabinAES(key, min_exp, max_exp, mask_bits, nc_level)
+        for size in sizes:
+            # Random data
+            data = os.urandom(size)
+            with BytesIO(data) as bio:
+                parts = cf_expand(chunker.chunkify(bio))
+            assert b"".join(parts) == data
+
+            # All-same data (non-zero)
+            data = b"\x42" * size
+            with BytesIO(data) as bio:
+                parts = cf_expand(chunker.chunkify(bio))
+            assert b"".join(parts) == data
+
+            # All-zero data
+            data = b"\x00" * size
+            with BytesIO(data) as bio:
+                parts = cf_expand(chunker.chunkify(bio))
+            assert b"".join(parts) == data