feat: identify + descramble D-Link OTA container (Realtek Ameba XOR payload)

signatures/dlink_ota.toml

@@ -0,0 +1,21 @@
+name = "dlink_ota"
+category = "encrypted"
+confidence = "structural"
+
+# D-Link OTA container: [u32 build][u32 count][count x 24-byte entries]. The first
+# entry's "OTA1" tag sits at offset 8; the deobf validator parses the entry table,
+# derives the single-byte XOR key from the Realtek Ameba image magic "81958711" in
+# the first payload, and confirms it. Keyless + self-validating, so a bare "OTA1"
+# with no valid entry table is rejected.
+magic_offset = 8
+validator = "deobf"
+
+[[magic]]
+ascii = "OTA1"
+endian = "little"
+
+[doc]
+description = "D-Link OTA container (Realtek Ameba single-byte-XOR payload; decryptable)."
+references = [
+  { title = "Realtek Ameba (RTL8710/8711) OTA image", url = "https://www.realtek.com/" },
+]

src/deobfuscate/scheme.cpp

@@ -416,6 +416,64 @@ std::optional<DescrambleResult> engenius(const Reader& r, size_t off, bool probe
     return res;
 }
 
+// ---- D-Link OTA container: a plain [build][count][entry x count] wrapper around
+// one or more Realtek Ameba firmware images, each single-byte-XOR obfuscated. Each
+// 24-byte entry is {"OTAn", u32 0x18, u32 cksum, u32 size, u32 payload_off, u32
+// flags}; payloads tile forward and the last ends at EOF. The XOR key is derived
+// from the known Ameba image magic "81958711" and verified, so a sibling device
+// that uses a different key still decrypts (or is reported undecryptable rather
+// than emitting garbage). Observed on D-Link DSP-W118 (RTL8711B). ----
+std::optional<DescrambleResult> dlink_ota(const Reader& r, size_t off, bool probe_only) {
+    static constexpr uint8_t kAmeba[8] = {'8', '1', '9', '5', '8', '7', '1', '1'};
+    auto count = r.at<uint32_t>(off + 4, Endian::Little);
+    if (!count || *count < 1 || *count > 9) return std::nullopt;
+    if (!starts_with(r, off + 8, reinterpret_cast<const uint8_t*>("OTA1"), 4)) return std::nullopt;
+    const size_t hdr_end = 8 + static_cast<size_t>(*count) * 24;
+    size_t expect = hdr_end;
+    uint32_t ota1_off = 0, ota1_size = 0;
+    for (uint32_t i = 0; i < *count; ++i) {
+        const size_t e = off + 8 + static_cast<size_t>(i) * 24;
+        if (!starts_with(r, e, reinterpret_cast<const uint8_t*>("OTA"), 3)) return std::nullopt;
+        auto stride = r.at<uint32_t>(e + 4, Endian::Little);
+        auto size = r.at<uint32_t>(e + 12, Endian::Little);
+        auto poff = r.at<uint32_t>(e + 16, Endian::Little);
+        if (!stride || *stride != 0x18 || !size || !poff || *poff != expect) return std::nullopt;
+        if (i == 0) {
+            ota1_off = *poff;
+            ota1_size = *size;
+        }
+        expect = static_cast<size_t>(*poff) + *size;
+    }
+    const size_t total_span = expect;
+    if (!ota1_size || off + total_span > r.size()) return std::nullopt;
+
+    // Derive the one-byte XOR key from the first Ameba magic byte ('8'), verify the
+    // full 8-byte magic. Keyless + self-validating.
+    const size_t p0 = off + ota1_off;
+    auto hdr = r.bytes(p0, 8);
+    if (!hdr) return std::nullopt;
+    const uint8_t key = static_cast<uint8_t>((*hdr)[0] ^ kAmeba[0]);
+    for (size_t i = 0; i < 8; ++i)
+        if (static_cast<uint8_t>((*hdr)[i] ^ key) != kAmeba[i]) return std::nullopt;
+
+    DescrambleResult res;
+    res.scheme = "dlink_ota";
+    res.cipher = "xor-1";
+    res.key_desc = "derived 0x" + to_hex(std::span<const uint8_t>(&key, 1));
+    res.validated = "realtek_ameba";
+    res.src_offset = p0;
+    res.src_len = ota1_size;
+    res.total_span = total_span;
+    if (!probe_only) {
+        auto ct = r.bytes(p0, ota1_size);
+        if (!ct) return std::nullopt;
+        std::vector<uint8_t> out(ota1_size);
+        for (size_t i = 0; i < ota1_size; ++i) out[i] = static_cast<uint8_t>((*ct)[i] ^ key);
+        res.data = std::move(out);
+    }
+    return res;
+}
+
 // Cheap header-structure check (no crypto): does a scheme's header actually parse
 // at `offset`? A bare magic match with no valid surrounding structure is a false
 // positive (esp. DLK's 3-byte magic, which hits ~once per 16 MB of random data) —
@@ -483,21 +541,46 @@ std::optional<StructMatch> scheme_structure(const Reader& r, size_t off) {
         if (len && *len > 136 && ml && *ml <= 256) return StructMatch{"engenius", *len};
         return std::nullopt;
     }
+    // D-Link OTA: count + "OTA1" at +8 + 24-byte entries that tile forward to EOF.
+    // (The dlink_ota scheme fn above handles the keyed case; this reports a variant
+    // whose payload does not XOR to the known Ameba magic as present-but-unkeyable.)
+    if (starts_with(r, off + 8, reinterpret_cast<const uint8_t*>("OTA1"), 4)) {
+        auto count = u32le(off + 4);
+        if (count && *count >= 1 && *count <= 9) {
+            size_t expect = 8 + static_cast<size_t>(*count) * 24;
+            bool ok = true;
+            for (uint32_t i = 0; i < *count; ++i) {
+                const size_t e = off + 8 + static_cast<size_t>(i) * 24;
+                auto stride = u32le(e + 4);
+                auto size = u32le(e + 12);
+                auto poff = u32le(e + 16);
+                if (!starts_with(r, e, reinterpret_cast<const uint8_t*>("OTA"), 3) || !stride ||
+                    *stride != 0x18 || !size || !poff || *poff != expect) {
+                    ok = false;
+                    break;
+                }
+                expect = static_cast<size_t>(*poff) + *size;
+            }
+            if (ok && off + expect <= r.size()) return StructMatch{"dlink_ota", expect};
+        }
+        return std::nullopt;
+    }
     return std::nullopt;
 }
 
 struct Entry {
     const char* name;
     SchemeFn fn;
 };
-constexpr std::array<Entry, 7> kSchemes = {{
+constexpr std::array<Entry, 8> kSchemes = {{
     {"dlink_shrs", shrs},
     {"dlink_encrpted_img", encrpted_img},
     {"openssl_salted", openssl_salted},
     {"dlink_mh01", mh01},
     {"dlink_dlk", dlk},
     {"dlink_tlv", tlv},
     {"engenius", engenius},
+    {"dlink_ota", dlink_ota},
 }};
 
 }  // namespace

tests/gen_samples.py

@@ -807,6 +807,19 @@ def engenius():
     return bytes(out)
 
 
+def dlink_ota():
+    # D-Link OTA container: [u32 build][u32 count][count x 24-byte entries], each
+    # payload a Realtek Ameba image ("81958711" magic) single-byte-XOR obfuscated.
+    # moria derives the one-byte key from the magic and verifies it, so this
+    # identifies dlink_ota at verified tier. Payloads tile forward to EOF.
+    pt = b"81958711" + bytes(24) + b"AMEBA" + bytes(475)   # 512-byte Ameba payload
+    obf = bytes(b ^ 0xAB for b in pt)
+    count = 1
+    hdr_end = 8 + count * 24
+    entry = b"OTA1" + struct.pack("<IIIII", 0x18, 0, len(obf), hdr_end, 0)
+    return struct.pack("<II", 0x30190412, count) + entry + obf
+
+
 def android_magic_at_zero():
     # Regression for the signature dedup: an "ANDROID!" magic at offset 0 with a
     # garbage (non-boot) header. The curated android_boot validator rejects it, and
@@ -1438,6 +1451,7 @@ def spiffs_img():
     ("fw.dlk", dlink_dlk, "dlink_dlk", STRUCTURAL),
     ("fw.tlv", dlink_tlv, "dlink_tlv", STRUCTURAL),
     ("fw.engenius", engenius, "engenius", STRUCTURAL),
+    ("fw.dlink_ota", dlink_ota, "dlink_ota", VERIFIED),
     ("fw.lantronix", lantronix, "lantronix_firmware", STRUCTURAL),
     ("fw.rfp", rae_rfp, "rae_rfp", VERIFIED),
     ("fw.otra", otra, "otra", VERIFIED),

tests/run.sh

@@ -47,6 +47,8 @@ python3 tests/test_jzlzma.py
 python3 tests/test_image_size.py
 # sxverity container (SpaceX dm-verity wrapper) identify + payload + FP guards (#55).
 python3 tests/test_sxverity.py
+# D-Link OTA container (Realtek Ameba single-byte-XOR) identify + derived-key descramble + FP guards.
+python3 tests/test_dlink_ota.py
 # RFP section extraction + LZARI decode round-trip (self-contained hard gate).
 python3 tests/test_rae_rfp.py
 # Windows CE XIP ROM: ROMHDR resolution, module PE rebuild, CECompress decode.

tests/test_dlink_ota.py

@@ -0,0 +1,136 @@
+#!/usr/bin/env python3
+"""D-Link OTA container identify + descramble regression.
+
+The container is [u32 build][u32 count][count x 24-byte entries]; each entry is
+{"OTAn", u32 0x18, u32 cksum, u32 size, u32 payload_off, u32 flags}. Payloads
+tile forward from just after the entry table and the last ends at EOF. Each
+payload is a Realtek Ameba firmware image (magic "81958711") obfuscated with a
+single-byte XOR. moria derives the key from the magic and verifies it, so the
+scheme is keyless and self-validating.
+
+This test builds synthetic containers (no corpus bytes) and checks:
+  1. a valid container is identified as dlink_ota (verified) and -e writes the
+     de-XORed payload (byte-exact, starts with the Ameba magic);
+  2. the key is DERIVED, not hardcoded -- a payload XORed with a non-0xab key
+     still decrypts;
+  3. FP guards: a bare "OTA1" with no valid entry table is not identified; a
+     structurally valid container whose payload is not an Ameba image is not
+     "decryptable" (no descrambled.bin emitted).
+
+Run: python3 tests/test_dlink_ota.py [path-to-moria]
+"""
+import json
+import os
+import struct
+import subprocess
+import sys
+import tempfile
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+MORIA = sys.argv[1] if len(sys.argv) > 1 else os.path.join(HERE, "..", "build", "moria")
+AMEBA = b"81958711"
+
+fails = []
+
+
+def check(cond, msg):
+    if not cond:
+        fails.append(msg)
+
+
+def container(payloads, build=0x30190412):
+    """payloads: list of raw (already-XORed) payload bytes. Returns the container."""
+    count = len(payloads)
+    hdr_end = 8 + count * 24
+    entries = b""
+    off = hdr_end
+    body = b""
+    for i, pl in enumerate(payloads):
+        tag = ("OTA%d" % (i + 1)).encode()
+        entries += tag + struct.pack("<IIIII", 0x18, 0, len(pl), off, 0)
+        body += pl
+        off += len(pl)
+    return struct.pack("<II", build, count) + entries + body
+
+
+def ameba_payload(key, body_after_magic=b"\x00" * 24 + b"AMEBA-TEST-IMAGE", size=512):
+    """A synthetic Ameba image (magic + filler) XORed with `key`."""
+    pt = AMEBA + body_after_magic
+    pt = pt + b"\x00" * (size - len(pt))
+    return bytes(b ^ key for b in pt)
+
+
+def moria_json(data, extra=None):
+    with tempfile.NamedTemporaryFile(suffix=".bin") as f:
+        f.write(data)
+        f.flush()
+        r = subprocess.run([MORIA, "-j"] + (extra or []) + [f.name], capture_output=True, timeout=120)
+    return json.loads(r.stdout)
+
+
+def ota_findings(data):
+    return [x for x in moria_json(data).get("findings", []) if x["type"] == "dlink_ota"]
+
+
+def extract(data):
+    """Run -e; return the descrambled.bin bytes if produced, else None."""
+    with tempfile.TemporaryDirectory() as t:
+        img = os.path.join(t, "ota.bin")
+        with open(img, "wb") as f:
+            f.write(data)
+        out = os.path.join(t, "out")
+        subprocess.run([MORIA, "-e", "-C", out, img], capture_output=True, timeout=120)
+        for dp, _, fs in os.walk(out):
+            if "descrambled.bin" in fs:
+                return open(os.path.join(dp, "descrambled.bin"), "rb").read()
+    return None
+
+
+def main():
+    # 1. Valid 2-slot container, real key 0xab -> identified + byte-exact de-XOR.
+    pt = AMEBA + b"\x00" * 24 + b"AMEBA-TEST-IMAGE" + b"\x00" * 464
+    obf = bytes(b ^ 0xAB for b in pt)
+    data = container([obf, obf])
+    gs = ota_findings(data)
+    check(len(gs) == 1 and gs[0]["offset"] == 0, f"valid container: expected one finding at 0, got {gs}")
+    if gs:
+        check(gs[0]["confidence_tier"] == "verified",
+              f"valid container tier={gs[0]['confidence_tier']}, want verified")
+        check(gs[0]["size"] == len(data), f"size {gs[0]['size']} != container {len(data)}")
+    dec = extract(data)
+    check(dec is not None and dec[:8] == AMEBA, "extract: descrambled.bin missing or wrong magic")
+    check(dec == pt, "extract: de-XOR not byte-exact")
+
+    # 2. Key is DERIVED, not hardcoded: a non-0xab key still decrypts.
+    for key in (0x3C, 0x5A, 0x01):
+        d2 = container([ameba_payload(key)])
+        gs2 = ota_findings(d2)
+        check(len(gs2) == 1 and gs2[0]["confidence_tier"] == "verified",
+              f"derived key 0x{key:02x}: not identified/verified ({gs2})")
+        dec2 = extract(d2)
+        check(dec2 is not None and dec2[:8] == AMEBA, f"derived key 0x{key:02x}: extract failed")
+
+    # 3a. FP guard: "OTA1" at offset 8 but no valid entry table (garbage) -> not dlink_ota.
+    bogus = struct.pack("<II", 0x30190412, 2) + b"OTA1" + os.urandom(44) + os.urandom(2048)
+    check(len(ota_findings(bogus)) == 0, "FP guard: bogus entry table must not match dlink_ota")
+
+    # 3b. FP guard: structurally valid container but payload is NOT an Ameba image
+    #     -> may be reported (encrypted) but must NOT be decryptable (no descrambled.bin).
+    notameba = container([b"NOTAMEBA" + b"\x11" * 504])
+    g3 = ota_findings(notameba)
+    if g3:
+        check(g3[0]["confidence_tier"] != "verified",
+              "FP guard: non-Ameba payload must not be reported decryptable")
+    check(extract(notameba) is None, "FP guard: non-Ameba payload must not yield descrambled.bin")
+
+    print("-" * 60)
+    if fails:
+        for m in fails:
+            print("FAIL:", m)
+        return 1
+    print("PASS: dlink_ota identify + derived-key XOR descramble + FP guards")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())