DNS Beacon：分片重组与密文认证

import base64
from ipaddress import IPv4Address

def xor_value(answer, idle="8.8.4.4"):
    return int(IPv4Address(answer)) ^ int(IPv4Address(idle))

def counted_labels(labels):
    if not labels or not labels[0] or labels[0][0] not in "123456789":
        raise ValueError("invalid label count")
    count = int(labels[0][0])
    if len(labels) != count:
        raise ValueError("label count mismatch")
    return labels[0][1:] + "".join(labels[1:])

def assemble(parts, expected):
    unique = {}
    for counter, data in parts:
        if counter in unique and unique[counter] != data:
            raise ValueError("conflicting duplicate")
        unique[counter] = data
    if sorted(unique) != list(range(1, len(unique) + 1)):
        raise ValueError("missing sequence")
    result = b"".join(unique[n] for n in sorted(unique))
    if len(result) != expected:
        raise ValueError("length mismatch")
    return result

assert xor_value("8.8.4.4") == 0
assert xor_value("8.8.4.246") == 0xF2
assert xor_value("8.8.4.68") == 64
assert xor_value("8.8.4.116") == 112
txt = "ZUZBozZmBi10KvISBcqS0nxp32b7h6WxUBw4n70cOLP13eN7PgcnUVOWdO+tDCbeElzdrp0b0N5DIEhB7eQ9Yg=="
download = base64.b64decode(txt, validate=True)
assert len(download) == 64
first = [
    "2942880f933a45cf2d048b0c14917493df0cd10a0de26ea103d0eb1b3",
    "4adf28c63a97deb5cbe4e20b26902d1ef427957323967835f7d18a42",
]
last = ["1debfa06ab4786477"]
a = bytes.fromhex(counted_labels(first))
b = bytes.fromhex(counted_labels(last))
expected = int(counted_labels(["140"]), 16)
upload = assemble([(2, b), (1, a), (1, a)], expected)
assert (len(a), len(b), len(upload)) == (56, 8, 64)
assert upload != download
addresses = [
    "19.64.240.89", "241.225.135.56", "127.132.170.127",
    "87.30.231.4", "97.156.155.27", "253.162.241.39",
    "61.217.211.72", "154.197.14.224", "211.139.207.53",
    "150.38.89.208",
]
partial = b"".join(IPv4Address(x).packed for x in addresses)
assert len(partial) == 40 and 112 - len(partial) == 72
for parts, length in (
    ([(1, a), (1, b)], 64),
    ([(2, b)], 8),
    ([(1, a)], 64),
):
    try:
        assemble(parts, length)
    except ValueError:
        pass
    else:
        raise AssertionError("invalid input accepted")
print("PASS: 4 XOR cases; TXT64; output56+8; 40/112 A bytes; reorder/dedup; 3 rejected cases")

import {createHmac, createDecipheriv, timingSafeEqual} from "node:crypto";

export function verifyAndDecrypt(blob, aesKey, hmacKey) {
  if (aesKey.length !== 16 || hmacKey.length !== 16)
    throw new Error("expected sample-specific 16-byte keys");
  if (blob.length < 32 || (blob.length - 16) % 16 !== 0)
    throw new Error("invalid encrypted envelope length");
  const cipher = blob.subarray(0, -16);
  const tag = blob.subarray(-16);
  const expected = createHmac("sha256", hmacKey)
    .update(cipher).digest().subarray(0, 16);
  if (!timingSafeEqual(tag, expected))
    throw new Error("HMAC mismatch");
  const decipher = createDecipheriv(
    "aes-128-cbc", aesKey, Buffer.from("abcdefghijklmnop"));
  decipher.setAutoPadding(false);
  return Buffer.concat([decipher.update(cipher), decipher.final()]);
}