|
| 1 | +""" |
| 2 | +Bluetooth-specific cryptographic primitives. |
| 3 | +
|
| 4 | +Implements only what the framework needs for key analysis and |
| 5 | +Resolvable Private Address verification. No new crypto math; we use |
| 6 | +`cryptography.hazmat` for AES-128. Each function is annotated with |
| 7 | +the spec section it implements so reviewers can audit against the |
| 8 | +Core Spec. |
| 9 | +""" |
| 10 | + |
| 11 | +from __future__ import annotations |
| 12 | + |
| 13 | +import math |
| 14 | +import re |
| 15 | +from collections import Counter |
| 16 | +from typing import Dict, List, Optional, Tuple # noqa: F401 |
| 17 | + |
| 18 | +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes |
| 19 | + |
| 20 | +# --------------------------------------------------------------------------- |
| 21 | +# Byte parsing helpers |
| 22 | +# --------------------------------------------------------------------------- |
| 23 | + |
| 24 | + |
| 25 | +_HEX_RE = re.compile(r"[^0-9a-fA-F]") |
| 26 | + |
| 27 | + |
| 28 | +def parse_hex_blob(value: str) -> bytes: |
| 29 | + """Accept a hex string in any of: |
| 30 | + 'aabbccdd' |
| 31 | + 'AA:BB:CC:DD' |
| 32 | + 'aa bb cc dd' |
| 33 | + '0xAABBCCDD' |
| 34 | + and return the bytes. Raises ValueError if non-hex chars remain |
| 35 | + after stripping separators and the result is odd-length. |
| 36 | + """ |
| 37 | + if value is None: |
| 38 | + raise ValueError("empty hex blob") |
| 39 | + s = value.strip() |
| 40 | + if s.lower().startswith("0x"): |
| 41 | + s = s[2:] |
| 42 | + s = _HEX_RE.sub("", s) |
| 43 | + if not s: |
| 44 | + raise ValueError("empty hex blob") |
| 45 | + if len(s) % 2: |
| 46 | + raise ValueError(f"odd-length hex blob ({len(s)} chars)") |
| 47 | + return bytes.fromhex(s) |
| 48 | + |
| 49 | + |
| 50 | +def parse_bd_addr(addr: str) -> bytes: |
| 51 | + """Parse 'AA:BB:CC:DD:EE:FF' into 6 bytes in transmission order |
| 52 | + (most-significant first as written). Raises ValueError on bad |
| 53 | + input.""" |
| 54 | + parts = addr.strip().split(":") |
| 55 | + if len(parts) != 6: |
| 56 | + raise ValueError(f"BD_ADDR must be 6 octets, got {addr!r}") |
| 57 | + out = bytearray() |
| 58 | + for p in parts: |
| 59 | + if len(p) != 2: |
| 60 | + raise ValueError(f"BD_ADDR octet must be 2 hex chars: {p!r}") |
| 61 | + out.append(int(p, 16)) |
| 62 | + return bytes(out) |
| 63 | + |
| 64 | + |
| 65 | +# --------------------------------------------------------------------------- |
| 66 | +# AES primitives |
| 67 | +# --------------------------------------------------------------------------- |
| 68 | + |
| 69 | + |
| 70 | +def aes128_ecb(key: bytes, plaintext: bytes) -> bytes: |
| 71 | + """Single-block AES-128 ECB. Both inputs must be 16 bytes.""" |
| 72 | + if len(key) != 16: |
| 73 | + raise ValueError(f"AES-128 key must be 16 bytes, got {len(key)}") |
| 74 | + if len(plaintext) != 16: |
| 75 | + raise ValueError(f"AES block must be 16 bytes, got {len(plaintext)}") |
| 76 | + enc = Cipher(algorithms.AES(key), modes.ECB()).encryptor() |
| 77 | + return enc.update(plaintext) + enc.finalize() |
| 78 | + |
| 79 | + |
| 80 | +# --------------------------------------------------------------------------- |
| 81 | +# Core Spec primitive: ah (random address hash) |
| 82 | +# --------------------------------------------------------------------------- |
| 83 | +# |
| 84 | +# Defined in Core Spec v5.4 Vol 3 Part H section 2.2.2. |
| 85 | +# |
| 86 | +# ah(k, r) = e(k, r') mod 2^24 |
| 87 | +# |
| 88 | +# where: |
| 89 | +# - k is the 128-bit IRK |
| 90 | +# - r is a 24-bit (3 octet) random value, the upper half of the RPA |
| 91 | +# - r' is r zero-padded to 128 bits on the most-significant side |
| 92 | +# - e is single-block AES-128 ECB |
| 93 | +# |
| 94 | +# The output is the low 24 bits of the AES ciphertext. |
| 95 | + |
| 96 | + |
| 97 | +def ah(irk: bytes, prand: bytes) -> bytes: |
| 98 | + """Compute the BLE Random Address Hash function `ah` per Core Spec |
| 99 | + Vol 3 Part H 2.2.2. |
| 100 | +
|
| 101 | + Args: |
| 102 | + irk: 16 bytes, the Identity Resolving Key. |
| 103 | + prand: 3 bytes, the random part of an RPA. |
| 104 | +
|
| 105 | + Returns: |
| 106 | + 3 bytes, the low 24 bits of e(irk, r'). |
| 107 | + """ |
| 108 | + if len(irk) != 16: |
| 109 | + raise ValueError(f"IRK must be 16 bytes, got {len(irk)}") |
| 110 | + if len(prand) != 3: |
| 111 | + raise ValueError(f"prand must be 3 bytes, got {len(prand)}") |
| 112 | + # r' = 0^104 || prand (big-endian per spec) |
| 113 | + r_padded = b"\x00" * 13 + prand |
| 114 | + ct = aes128_ecb(irk, r_padded) |
| 115 | + return ct[-3:] # low 24 bits |
| 116 | + |
| 117 | + |
| 118 | +def is_resolvable_private(addr_bytes: bytes) -> bool: |
| 119 | + """True iff the address has top two bits 01 (RPA format). |
| 120 | +
|
| 121 | + Per Core Spec Vol 6 Part B 1.3.2.2 the random part bits 47:46 |
| 122 | + are '01' for a Resolvable Private Address. addr_bytes is in |
| 123 | + transmission order, so byte 0 holds the MSB. |
| 124 | + """ |
| 125 | + if len(addr_bytes) != 6: |
| 126 | + return False |
| 127 | + return (addr_bytes[0] >> 6) == 0b01 |
| 128 | + |
| 129 | + |
| 130 | +def rpa_resolves(irk: bytes, addr: str) -> bool: |
| 131 | + """Test whether `irk` resolves the Resolvable Private Address `addr`. |
| 132 | +
|
| 133 | + The RPA layout (Core Spec Vol 6 Part B 1.3.2.2) splits the 48-bit |
| 134 | + address into: |
| 135 | + addr[47:24] = prand (top 3 bytes; bits 47:46 = 01) |
| 136 | + addr[23: 0] = hash (low 3 bytes) |
| 137 | +
|
| 138 | + The check is `ah(irk, prand) == hash`. |
| 139 | +
|
| 140 | + Args: |
| 141 | + irk: 16 bytes. |
| 142 | + addr: 'AA:BB:CC:DD:EE:FF' colon-separated. |
| 143 | +
|
| 144 | + Returns: |
| 145 | + True iff the IRK resolves the address. Returns False (does |
| 146 | + not raise) if the address is not in RPA form, so callers can |
| 147 | + bulk-test mixed lists. |
| 148 | + """ |
| 149 | + try: |
| 150 | + ab = parse_bd_addr(addr) |
| 151 | + except ValueError: |
| 152 | + return False |
| 153 | + if not is_resolvable_private(ab): |
| 154 | + return False |
| 155 | + prand = ab[:3] |
| 156 | + expect_hash = ab[3:] |
| 157 | + return ah(irk, prand) == expect_hash |
| 158 | + |
| 159 | + |
| 160 | +# --------------------------------------------------------------------------- |
| 161 | +# Key-quality statistics |
| 162 | +# --------------------------------------------------------------------------- |
| 163 | + |
| 164 | + |
| 165 | +def shannon_entropy_bits(data: bytes) -> float: |
| 166 | + """Shannon entropy in bits per byte. Range 0..8. |
| 167 | +
|
| 168 | + A uniformly random 128-bit key is expected close to 8.0 over |
| 169 | + many samples, though for short blobs (16 bytes) the empirical |
| 170 | + value is typically 3.5..4.5 due to limited sample size. We |
| 171 | + surface the raw number; the caller decides the threshold. |
| 172 | + """ |
| 173 | + if not data: |
| 174 | + return 0.0 |
| 175 | + counts = Counter(data) |
| 176 | + n = len(data) |
| 177 | + h = 0.0 |
| 178 | + for c in counts.values(): |
| 179 | + p = c / n |
| 180 | + h -= p * math.log2(p) |
| 181 | + return h |
| 182 | + |
| 183 | + |
| 184 | +def chi_square_uniform_bytes(data: bytes) -> Tuple[float, int]: |
| 185 | + """Pearson chi-square statistic of `data` against the uniform |
| 186 | + 256-bin byte distribution. |
| 187 | +
|
| 188 | + Returns: |
| 189 | + (statistic, degrees_of_freedom). |
| 190 | +
|
| 191 | + For 16 bytes the degrees of freedom is 16 (one less than the |
| 192 | + sample size) and the expected statistic for true uniform is |
| 193 | + around 16; values >> 30 suggest non-uniformity. The caller is |
| 194 | + responsible for thresholding given the small sample. |
| 195 | + """ |
| 196 | + if not data: |
| 197 | + return (0.0, 0) |
| 198 | + n = len(data) |
| 199 | + counts = Counter(data) |
| 200 | + # We collapse to the bins that actually occurred + zero-count |
| 201 | + # bins, but the chi-square test for small samples has limited |
| 202 | + # power. The statistic is informative, not a hard pass/fail. |
| 203 | + expected = n / 256 |
| 204 | + chi = 0.0 |
| 205 | + for byte in range(256): |
| 206 | + observed = counts.get(byte, 0) |
| 207 | + chi += (observed - expected) ** 2 / expected |
| 208 | + return (chi, n - 1) |
| 209 | + |
| 210 | + |
| 211 | +def repeated_byte_runs(data: bytes) -> List[Tuple[int, int]]: |
| 212 | + """Find runs of >=3 identical consecutive bytes. |
| 213 | +
|
| 214 | + Returns: |
| 215 | + List of (start_offset, run_length). |
| 216 | + """ |
| 217 | + out: List[Tuple[int, int]] = [] |
| 218 | + if len(data) < 3: |
| 219 | + return out |
| 220 | + i = 0 |
| 221 | + while i < len(data): |
| 222 | + j = i |
| 223 | + while j < len(data) and data[j] == data[i]: |
| 224 | + j += 1 |
| 225 | + if j - i >= 3: |
| 226 | + out.append((i, j - i)) |
| 227 | + i = max(j, i + 1) |
| 228 | + return out |
| 229 | + |
| 230 | + |
| 231 | +KNOWN_WEAK_KEYS: Dict[bytes, str] = { |
| 232 | + b"\x00" * 16: "all-zero", |
| 233 | + b"\xFF" * 16: "all-ones", |
| 234 | + bytes(range(16)): "00..0F sequential", |
| 235 | + bytes(range(16, 32)): "10..1F sequential", |
| 236 | + bytes.fromhex("deadbeefdeadbeefdeadbeefdeadbeef"): "deadbeef test pattern", |
| 237 | + bytes.fromhex("00112233445566778899aabbccddeeff"): "00..ff classic test", |
| 238 | + b"A" * 16: "all 0x41 (ASCII A)", |
| 239 | + b"\x01" * 16: "all 0x01", |
| 240 | +} |
| 241 | + |
| 242 | + |
| 243 | +def known_weak_match(data: bytes) -> Optional[str]: |
| 244 | + """Return a description if `data` matches a known weak/test key, |
| 245 | + else None.""" |
| 246 | + return KNOWN_WEAK_KEYS.get(data) |
| 247 | + |
| 248 | + |
| 249 | +def analyze_key(data: bytes) -> Dict[str, object]: |
| 250 | + """Run every key-quality check on `data` and return a verdict dict. |
| 251 | +
|
| 252 | + The verdict is a short string suitable for a one-line summary. |
| 253 | + A 'PASS' verdict means no automatic red flag was found; it does |
| 254 | + NOT certify the key is secret or unguessable. |
| 255 | +
|
| 256 | + Keys in the returned dict: |
| 257 | + raw, length, entropy_bits, chi_square, unique_bytes, |
| 258 | + repeated_runs, known_weak, all_ascii_printable, verdict. |
| 259 | + """ |
| 260 | + weak = known_weak_match(data) |
| 261 | + entropy = shannon_entropy_bits(data) |
| 262 | + chi, _df = chi_square_uniform_bytes(data) |
| 263 | + runs = repeated_byte_runs(data) |
| 264 | + unique = len(set(data)) |
| 265 | + ascii_only = all(0x20 <= b < 0x7F for b in data) if data else False |
| 266 | + |
| 267 | + verdict = _key_verdict( |
| 268 | + weak=weak, length=len(data), unique=unique, |
| 269 | + runs=runs, ascii_only=ascii_only, entropy=entropy, |
| 270 | + ) |
| 271 | + |
| 272 | + return { |
| 273 | + "raw": data, |
| 274 | + "length": len(data), |
| 275 | + "entropy_bits": entropy, |
| 276 | + "chi_square": chi, |
| 277 | + "unique_bytes": unique, |
| 278 | + "repeated_runs": runs, |
| 279 | + "known_weak": weak, |
| 280 | + "all_ascii_printable": ascii_only, |
| 281 | + "verdict": verdict, |
| 282 | + } |
| 283 | + |
| 284 | + |
| 285 | +def _key_verdict( |
| 286 | + *, |
| 287 | + weak: Optional[str], |
| 288 | + length: int, |
| 289 | + unique: int, |
| 290 | + runs: List[Tuple[int, int]], |
| 291 | + ascii_only: bool, |
| 292 | + entropy: float, |
| 293 | +) -> str: |
| 294 | + if weak is not None: |
| 295 | + return f"FAIL: matches known weak key ({weak})" |
| 296 | + if length == 0: |
| 297 | + return "FAIL: empty key" |
| 298 | + if length < 7: |
| 299 | + return f"FAIL: key too short ({length} bytes)" |
| 300 | + if unique <= 2: |
| 301 | + return f"FAIL: only {unique} distinct byte value(s)" |
| 302 | + if ascii_only: |
| 303 | + return "WARN: key bytes are all ASCII printable; likely a passphrase, not random" |
| 304 | + if runs: |
| 305 | + worst = max(r[1] for r in runs) |
| 306 | + if worst >= length // 2: |
| 307 | + return f"FAIL: single byte repeats for {worst} of {length} positions" |
| 308 | + return f"WARN: repeated-byte run(s) detected (longest={worst})" |
| 309 | + if entropy < 2.5 and length >= 8: |
| 310 | + return f"WARN: low Shannon entropy ({entropy:.2f} bits/byte) for {length}-byte key" |
| 311 | + return "PASS: no automatic red flag" |
0 commit comments