python: strict receiver validation per BBQr BIP draft

python/bbqr/consts.py

@@ -8,6 +8,9 @@
 # Standard defines a fixed-length header
 HEADER_LEN = 8
 
+# Default cap on decoded/decompressed transfer size (overridable per call)
+MAX_SIZE = 16 * 1024 * 1024
+
 # Human names
 FILETYPE_NAMES = dict(P='PSBT', T='Transaction', J='JSON', C='CBOR', U='Unicode Text',
                         X='Executable', B='Binary',

python/bbqr/join.py

@@ -3,14 +3,23 @@
 #
 # - joins QR codes
 #
+import re
 from .utils import decode_data
-from .consts import HEADER_LEN, KNOWN_FILETYPES
+from .consts import HEADER_LEN, KNOWN_FILETYPES, MAX_SIZE
 
-def join_qrs(parts):
-    # take a bunch of scanned data. 
+# strict header grammar: B$ magic, known encoding, one uppercase letter of
+# file type, then uppercase base-36 digits for part count and index
+HEADER_RE = re.compile(r'\AB\$[H2Z][A-Z][0-9A-Z]{2}[0-9A-Z]{2}\Z')
+
+def join_qrs(parts, max_size=MAX_SIZE):
+    # take a bunch of scanned data.
     # - put into order, decode, return type code and raw data bytes
     # - lazy desktop code here
-    hdr = set(p[0:6] for p in parts)
+    hdr = set()
+    for p in parts:
+        assert HEADER_RE.match(p[0:HEADER_LEN]), f'invalid header: {p[0:HEADER_LEN]!r}'
+        assert len(p) > HEADER_LEN, 'empty body'
+        hdr.add(p[0:6])
     assert len(hdr) == 1, 'conflicting/variable filetype/encodings/sizes'
     hdr = hdr.pop()
 
@@ -24,6 +33,7 @@ def join_qrs(parts):
 
     # ok to have dups here, just need them all
     data = {}
+    body_len = None
     for p in parts:
         idx = int(p[6:8], 36)
         assert idx < num_parts, f'got part {idx} but only expecting {num_parts}'
@@ -33,12 +43,24 @@ def join_qrs(parts):
         else:
             data[idx] = p[8:]
 
+        if idx != num_parts - 1:
+            # all non-final bodies must share one length
+            if body_len is None:
+                body_len = len(p) - HEADER_LEN
+            assert len(p) - HEADER_LEN == body_len, 'non-final parts must have equal length'
+
     missing = set(range(num_parts)) - set(data)
     assert not missing, f'parts missing: {missing!r}'
 
+    if num_parts > 1:
+        # final body must be no longer than the others
+        assert len(data[num_parts - 1]) <= body_len, 'final part too long'
+
     parts = [data[i] for i in range(num_parts)]
 
-    raw = decode_data(parts, encoding)
+    raw = decode_data(parts, encoding, max_size)
+
+    assert raw, 'empty transfer'
 
     # maybe: decode objects here... U=>text, C=>obj, J=>obj
 

python/bbqr/utils.py

@@ -3,8 +3,11 @@
 #
 # - helpers and basics
 #
-import zlib
+import re, zlib
 from base64 import b32encode, b32decode
+from .consts import MAX_SIZE
+
+HEX_RE = re.compile(r'\A[0-9A-F]*\Z')
 
 def version_to_chars(v):
     # return number of **chars** that fit into indicated version QR
@@ -55,24 +58,167 @@ def encode_data(raw, encoding=None):
 
     return encoding, data, 8
 
-def decode_data(parts, encoding):
+def scan_deflate_distances(stream, max_dist=1024):
+    # Validate-only parse of a raw DEFLATE stream (RFC 1951): every
+    # back-reference distance must fit max_dist and the output so far.
+    # zlib cannot check this itself: its strict window check is compiled
+    # out (INFLATE_STRICT) and CPython exposes no knob for it.
+    LEN_BASE = [3,4,5,6,7,8,9,10,11,13,15,17,19,23,27,31,35,43,51,59,
+                67,83,99,115,131,163,195,227,258]
+    LEN_EXTRA = [0,0,0,0,0,0,0,0,1,1,1,1,2,2,2,2,3,3,3,3,4,4,4,4,5,5,5,5,0]
+    DIST_BASE = [1,2,3,4,5,7,9,13,17,25,33,49,65,97,129,193,257,385,513,769,
+                 1025,1537,2049,3073,4097,6145,8193,12289,16385,24577]
+    DIST_EXTRA = [0,0,0,0,1,1,2,2,3,3,4,4,5,5,6,6,7,7,8,8,9,9,10,10,11,11,12,12,13,13]
+    CLC_ORDER = [16,17,18,0,8,7,9,6,10,5,11,4,12,3,13,2,14,1,15]
+
+    idx, bitbuf, bitcnt = 0, 0, 0
+
+    def bits(n):
+        nonlocal idx, bitbuf, bitcnt
+        while bitcnt < n:
+            assert idx < len(stream), 'incomplete DEFLATE stream'
+            bitbuf |= stream[idx] << bitcnt
+            idx += 1
+            bitcnt += 8
+        rv = bitbuf & ((1 << n) - 1)
+        bitbuf >>= n
+        bitcnt -= n
+        return rv
+
+    def build(lengths):
+        # canonical Huffman code table: (num bits, code) => symbol
+        table = {}
+        code = 0
+        for ln in range(1, 16):
+            for sym, l in enumerate(lengths):
+                if l == ln:
+                    table[(ln, code)] = sym
+                    code += 1
+            code <<= 1
+        return table
+
+    def decode(table):
+        ln = code = 0
+        while True:
+            code = (code << 1) | bits(1)
+            ln += 1
+            assert ln <= 15, 'bad Huffman code'
+            if (ln, code) in table:
+                return table[(ln, code)]
+
+    produced = 0
+    while True:
+        final = bits(1)
+        btype = bits(2)
+        assert btype != 3, 'bad block type'
+
+        if btype == 0:
+            # stored block: byte-align, LEN/NLEN, skip contents
+            bits(bitcnt % 8)
+            idx -= bitcnt // 8
+            bitbuf = bitcnt = 0
+            assert idx + 4 <= len(stream), 'incomplete DEFLATE stream'
+            ln = stream[idx] | (stream[idx+1] << 8)
+            nlen = stream[idx+2] | (stream[idx+3] << 8)
+            assert ln ^ nlen == 0xFFFF, 'bad stored block'
+            idx += 4
+            assert idx + ln <= len(stream), 'incomplete DEFLATE stream'
+            idx += ln
+            produced += ln
+        else:
+            if btype == 1:
+                lit = build([8]*144 + [9]*112 + [7]*24 + [8]*8)
+                dist = build([5]*32)
+            else:
+                hlit = bits(5) + 257
+                hdist = bits(5) + 1
+                hclen = bits(4) + 4
+                cl_lens = [0] * 19
+                for i in range(hclen):
+                    cl_lens[CLC_ORDER[i]] = bits(3)
+                cl = build(cl_lens)
+                lens = []
+                while len(lens) < hlit + hdist:
+                    sym = decode(cl)
+                    if sym < 16:
+                        lens.append(sym)
+                    elif sym == 16:
+                        assert lens, 'bad code lengths'
+                        lens += [lens[-1]] * (3 + bits(2))
+                    elif sym == 17:
+                        lens += [0] * (3 + bits(3))
+                    else:
+                        lens += [0] * (11 + bits(7))
+                assert len(lens) == hlit + hdist, 'bad code lengths'
+                lit = build(lens[:hlit])
+                dist = build(lens[hlit:])
+
+            while True:
+                sym = decode(lit)
+                if sym == 256:
+                    break
+                if sym < 256:
+                    produced += 1
+                    continue
+                assert sym <= 285, 'bad length code'
+                length = LEN_BASE[sym-257] + bits(LEN_EXTRA[sym-257])
+                dsym = decode(dist)
+                assert dsym <= 29, 'bad distance code'
+                d = DIST_BASE[dsym] + bits(DIST_EXTRA[dsym])
+                assert d <= produced, 'invalid distance too far back'
+                assert d <= max_dist, 'distance exceeds window'
+                produced += length
+
+        if final:
+            break
+
+def decode_data(parts, encoding, max_size=MAX_SIZE):
     # give back the bytes after decoding
     # - already in order
     # - keeps the parts separate here to validate correct split from encoder
     if encoding == 'H':
-        return b''.join(bytes.fromhex(p) for p in parts)
+        rv = b''
+        for p in parts:
+            assert HEX_RE.match(p), 'non-canonical hex body'
+            rv += bytes.fromhex(p)
+        assert len(rv) <= max_size, 'decoded data too large'
+        return rv
 
     # base32 decode, but insert padding for API
     rv = b''
-    for p in parts:
-        padding = (8 - (len(p) % 8)) % 8
-        rv += b32decode(p + (padding*'='))
+    for n, p in enumerate(parts):
+        residue = len(p) % 8
+        is_final = (n == len(parts) - 1)
+        bad_length = residue in (1, 3, 6) if is_final else residue != 0
+        assert not bad_length, 'invalid Base32 body length'
+
+        padding = (8 - residue) % 8
+        here = b32decode(p + (padding*'='))
+        # non-zero pad bits in the final Base32 symbol are non-canonical
+        assert b32encode(here).decode('ascii').rstrip('=') == p, 'non-canonical Base32 body'
+        rv += here
+
+    assert len(rv) <= max_size, 'decoded data too large'
 
     if encoding == 'Z':
-        # decompress
+        # exact 1k window enforcement, which zlib alone cannot provide
+        scan_deflate_distances(rv)
+
+        # decompress in 1k chunks so the size cap applies while inflating,
+        # instead of after the full output has been buffered
         z = zlib.decompressobj(wbits=-10)
-        rv = z.decompress(rv)
-        rv += z.flush()
+        chunks = []
+        total = 0
+        while rv and not z.eof:
+            here = z.decompress(rv, 1024)
+            total += len(here)
+            assert total <= max_size, 'decompressed data too large'
+            chunks.append(here)
+            rv = z.unconsumed_tail
+        chunks.append(z.flush())
+        assert z.eof, 'incomplete DEFLATE stream'
+        assert not z.unused_data, 'trailing data after DEFLATE stream'
+        rv = b''.join(chunks)
 
     return rv
 

python/tests/test_vectors.py

@@ -0,0 +1,60 @@
+#
+# (c) Copyright 2023 by Coinkite Inc. This file is in the public domain.
+#
+
+from context import bbqr
+import pytest, json, hashlib, zlib
+
+with open('../test_data/bip-test-vectors.json') as f:
+    DOC = json.load(f)
+
+@pytest.mark.parametrize('vec', DOC['vectors'], ids=lambda v: v['name'])
+def test_positive_vectors(vec):
+    for frames in (vec['frames'], list(reversed(vec['frames']))):
+        file_type, raw = bbqr.join_qrs(frames)
+        assert file_type == vec['file_type']
+        assert len(raw) == vec['input_length']
+        assert hashlib.sha256(raw).hexdigest() == vec['input_sha256']
+
+@pytest.mark.parametrize('case', DOC['invalid_cases'], ids=lambda c: c['name'])
+def test_negative_vectors(case):
+    with pytest.raises((AssertionError, ValueError, zlib.error)):
+        bbqr.join_qrs(case['frames'])
+
+@pytest.mark.parametrize('case', DOC['strict_policy_cases'], ids=lambda c: c['name'])
+def test_strict_policy_cases(case):
+    # draft v4: receivers MAY ignore later duplicates without comparing;
+    # we compare bodies and fail on conflict - stricter local policy
+    with pytest.raises(AssertionError):
+        bbqr.join_qrs(case['frames'])
+
+def test_overwide_deflate_distance():
+    # back-reference distance of 2048 needs a bigger window than wbits=10 allows
+    frame = open('../test_data/deflate-overwide-distance.txt').read().strip()
+    with pytest.raises(AssertionError, match='window'):
+        bbqr.join_qrs([frame])
+
+def test_deflate_window_boundary():
+    # distance 1024 is the window maximum and must decode
+    frame = open('../test_data/deflate-dist1024.txt').read().strip()
+    _, raw = bbqr.join_qrs([frame])
+    assert len(raw) == 2048
+    assert hashlib.sha256(raw).hexdigest() == \
+        'c30537f307aa7aed41677a596ea4f60de232ff2dc2ef7b478e6ae53e300db05d'
+
+    # distance 1025 exceeds the window and must be rejected
+    frame = open('../test_data/deflate-dist1025.txt').read().strip()
+    with pytest.raises(AssertionError, match='window'):
+        bbqr.join_qrs([frame])
+
+def test_decompressed_size_cap():
+    vec = next(v for v in DOC['vectors'] if v['name'] == 'deflate-psbt')
+
+    # compressed input fits the cap; decompressed output must not
+    with pytest.raises(AssertionError, match='too large'):
+        bbqr.join_qrs(vec['frames'], max_size=vec['input_length'] - 1)
+
+    _, raw = bbqr.join_qrs(vec['frames'])
+    assert len(raw) == vec['input_length']
+
+# EOF