"""IMM0008-only allowlisted XFA writer. Adobe signing and validation stay manual. Dataset bindings and saved widget state follow the reviewed language contract. Official scripts, lookup packets and repeat limits remain unchanged. """ from io import BytesIO import hashlib import json import os from pathlib import Path import re from lxml import etree as E import pikepdf XFA_DATA = 'http://www.xfa.org/schema/xfa-data/1.0/' FORM_NS = 'http://www.xfa.org/schema/xfa-form/2.8/' PART = re.compile(r'([^\[\].]+)(?:\[(\d+)\])?') PHONE_PREFIXES = tuple('form1.Page1.contactInformation.' + path for path in ( 'q3-4.Phone.', 'q3-4.AltPhone.', 'q5-6.FaxPhone.')) def parse(raw): return E.parse(BytesIO(raw), E.XMLParser(remove_blank_text=False, remove_comments=False, remove_pis=False, resolve_entities=False)) def local(node): return E.QName(node).localname if isinstance(node.tag, str) else '' def parts(path): result = [] for segment in path.split('.'): match = PART.fullmatch(segment) if not match: raise ValueError('Invalid IMM0008 path: ' + path) result.append((match[1], int(match[2]) if match[2] else None)) return result def resolve(root, path, named=False, create=False, kinds=None): """Exact local path; omitted indexes must be unambiguous.""" parent = root for offset, (name, index) in enumerate(parts(path)): matches = [c for c in parent if (c.get('name') if named else local(c)) == name] if index is None and len(matches) > 1: raise ValueError('Ambiguous IMM0008 path: ' + path) selected = index if index is not None else 0 if selected >= len(matches): if not create: raise ValueError('Unresolved IMM0008 target: ' + path) for _ in range(selected + 1 - len(matches)): if named: node = E.SubElement(parent, '{' + FORM_NS + '}' + kinds[offset], name=name) else: node = E.SubElement(parent, name) matches.append(node) parent = matches[selected] return parent def form_value(node, value, value_type='text'): for old in list(node): if local(old) == 'value': node.remove(old) value_node = E.SubElement(node, '{' + FORM_NS + '}value') E.SubElement(value_node, '{' + FORM_NS + '}' + value_type).text = value or None def checkbox_type(widget): if widget.xpath('./*[local-name()="ui"]/*[local-name()="checkButton"]'): exports = widget.xpath('./*[local-name()="items"]/*') if exports and local(exports[0]) in ('text', 'integer'): return local(exports[0]) return 'text' def template_chain(template, path): """Resolve every named ancestor and preserve actual widget kinds.""" parent = template chain = [] for name, index in parts(path): matches = [c for c in parent if c.get('name') == name] if index is None and len(matches) > 1: raise ValueError('Ambiguous IMM0008 template target: ' + path) selected = index if index is not None else 0 if selected >= len(matches): raise ValueError('Unresolved IMM0008 template target: ' + path) parent = matches[selected] chain.append(parent) if local(parent) not in ('field', 'exclGroup'): raise ValueError('Unresolved IMM0008 widget target: ' + path) return chain def write_state(form, path, chain, value): node = resolve(form, path, named=True, create=True, kinds=[local(n) for n in chain]) if path.startswith(PHONE_PREFIXES): # Match Reader's saved, dataset-bound telephone widgets rather than # storing a second scalar value in the form packet. for old in list(node): if local(old) == 'value': node.remove(old) if path.rsplit('.', 1)[-1] in ('CanadaUS', 'Other'): if value == '1': E.SubElement(node, '{' + FORM_NS + '}value', override='1') ui = next((c for c in node if local(c) == 'ui'), None) if ui is None: ui = E.SubElement(node, '{' + FORM_NS + '}ui') if not any(local(c) == 'checkButton' for c in ui): E.SubElement(ui, '{' + FORM_NS + '}checkButton') return form_value(node, value, checkbox_type(chain[-1])) if local(chain[-1]) == 'exclGroup': for widget in chain[-1]: if local(widget) != 'field' or not widget.get('name'): continue saved = next((c for c in node if c.get('name') == widget.get('name')), None) if saved is None: saved = E.SubElement(node, '{' + FORM_NS + '}field', name=widget.get('name')) exports = widget.xpath('./*[local-name()="items"]/*') export = (exports[0].text or '') if exports else '' form_value(saved, export if export == value else '', checkbox_type(widget)) def replace_data_packet(raw, data): """Replace only xfa:data; leave the official LOV subtree byte-identical.""" from xml.parsers import expat parser = expat.ParserCreate(namespace_separator='}') span = [] depth = 0 active_depth = None def start(tag, attrs): nonlocal depth, active_depth depth += 1 if tag == XFA_DATA + '}data': if span: raise ValueError('Ambiguous IMM0008 data packet') span.append(parser.CurrentByteIndex) active_depth = depth def end(tag): nonlocal depth if active_depth == depth and tag == XFA_DATA + '}data': span.append(raw.index(b'>', parser.CurrentByteIndex) + 1) depth -= 1 parser.StartElementHandler, parser.EndElementHandler = start, end parser.Parse(raw, True) if len(span) != 2: raise ValueError('Unresolved IMM0008 data packet') return raw[:span[0]] + E.tostring(data, with_tail=False) + raw[span[1]:] def load_contract(language): return json.loads((Path(__file__).parent / f'imm0008-contract.{language}.json').read_text(encoding='utf-8')) def fill_imm0008(payload, output_path, blank_path): """Fill only reviewed IMM0008 targets; preserve the official template scripts.""" # Symfony's Windows web-process environment can omit SystemRoot. asyncio # needs it to initialise Winsock, even though PDF filling uses no network. if os.name == 'nt' and not os.environ.get('SystemRoot'): system_root = os.environ.get('WINDIR') if not system_root: import ctypes buffer = ctypes.create_unicode_buffer(32768) length = ctypes.windll.kernel32.GetWindowsDirectoryW(buffer, len(buffer)) if not 0 < length < len(buffer): raise OSError('IMM0008 could not resolve the Windows system directory') system_root = buffer.value os.environ['SystemRoot'] = system_root from imm0008_pdf_state import apply_state, normalize_fields, prepare_phone_state from imm0008_xfa_checksum import form_checksum from pyhanko.pdf_utils.incremental_writer import IncrementalPdfFileWriter from pyhanko.pdf_utils import generic if payload.get('form') != 'IMM0008': raise ValueError('IMM0008 writer rejects other forms') language = payload.get('language', 'en') if language not in ('en', 'fr'): raise ValueError('Unsupported IMM0008 language') contract = load_contract(language) blank = Path(blank_path) if Path(output_path).resolve() == blank.resolve(): raise ValueError('IMM0008 output must not overwrite the official blank') if hashlib.sha256(blank.read_bytes()).hexdigest() != contract['sha256']: raise ValueError('IMM0008 template requires review') fields = payload.get('fields', {}) # PHP serializes an empty associative array as []; keep reviewable blank # drafts downloadable while rejecting every non-empty positional array. if fields == []: fields = {} if not isinstance(fields, dict): raise ValueError('IMM0008 payload.fields must be an object') for path, value in fields.items(): meta = contract['fields'].get(path) if meta is None: raise ValueError('Unallowlisted IMM0008 field: ' + str(path)) if not isinstance(value, str): raise ValueError('Invalid IMM0008 field value: ' + path) if len(value) > meta.get('max_chars', 2000): raise ValueError('IMM0008 field too long: ' + path) for key in ('allowed_values', 'choices'): if value and meta.get(key) and value not in meta[key]: raise ValueError('Invalid IMM0008 export: ' + path) program = fields.get('form1.Page1.genDetails.q1.Program', '') category = fields.get('form1.Page1.genDetails.q1.Category', '') if category and (not program or category not in contract['program_categories'].get(program, {})): raise ValueError('Invalid IMM0008 program/category combination') count = payload.get('imm0008_dependent_count') if count is None: count = max([0] + [1 if '.depPrime.' in p else int(re.search(r'\.dependants\[(\d+)\]', p)[1]) + 2 for p in fields if '.depPrime.' in p or re.search(r'\.dependants\[(\d+)\]', p)]) if type(count) is not int or not 0 <= count <= 5: raise ValueError('Invalid IMM0008 dependent count') fields = normalize_fields(dict(fields)) # The French blank has no future-contact consent widget. Do not create it. future_contact = 'form1.Page1.declaration.FutureContact' if future_contact not in contract['fields'] and fields.get(future_contact) == '': fields.pop(future_contact, None) for path in fields: if path not in contract['fields']: raise ValueError('Unallowlisted IMM0008 state target: ' + path) with pikepdf.open(blank) as pdf: xfa = pdf.Root.AcroForm.XFA streams = {str(xfa[i]): xfa[i + 1] for i in range(0, len(xfa), 2)} original_template = streams['template'].read_bytes() template = parse(original_template).getroot() plans = [] for path, value in fields.items(): meta = contract['fields'][path] template_path = re.sub(r'\.dependants\[\d+\]', '.dependants', meta['template_path']) chain = template_chain(template, template_path) saved_path = meta.get('state_path', path) plans.append((path, value, meta.get('data_path', path), saved_path, chain)) original_datasets = streams['datasets'].read_bytes() datasets_tree = parse(original_datasets) data = datasets_tree.getroot().find('{' + XFA_DATA + '}data') if data is None: data = E.SubElement(datasets_tree.getroot(), '{' + XFA_DATA + '}data') form_tree = parse(streams['form'].read_bytes()) if 'form' in streams else None form = form_tree.getroot() if form_tree is not None else None # Materialize every requested slot, including an empty invalid middle slot. page = resolve(data, 'form1.Page1', create=True) for name, limit in (('depPrime', int(count > 0)), ('dependants', max(0, count - 1))): nodes = [n for n in page if local(n) == name] for node in nodes[limit:]: page.remove(node) if limit: resolve(data, 'form1.Page1.' + name + f'[{limit-1}]', create=True) if form is not None: saved_page = resolve(form, 'form1.Page1', named=True, create=True, kinds=['subform', 'subform']) for name, limit in (('depPrime', int(count > 0)), ('dependants', max(0, count - 1))): nodes = [n for n in saved_page if n.get('name') == name] for node in nodes[limit:]: saved_page.remove(node) if limit: resolve(form, 'form1.Page1.' + name + f'[{limit-1}]', named=True, create=True, kinds=['subform', 'subform', 'subform']) # depPrime is present by default in the signed template. Retain a # saved visibility override even with no dependant dataset record. for name in ('depPrime', 'depInfoHeader', 'depHeader'): resolve(form, 'form1.Page1.' + name, named=True, create=True, kinds=['subform', 'subform', 'subform']) for path, value, data_path, saved_path, chain in plans: if '.depPrime.' in path and count == 0: continue match = re.search(r'\.dependants\[(\d+)\]', path) if match and int(match[1]) >= count - 1: continue target = resolve(data, data_path, create=True) for old in list(target): target.remove(old) target.text = value or None if path.startswith(PHONE_PREFIXES): # Match Reader's saved scalar telephone binding, without # redundant dataNode attributes or widget value deltas. target.attrib.pop('{' + XFA_DATA + '}dataNode', None) else: target.set('{' + XFA_DATA + '}dataNode', 'dataValue') if form is not None: write_state(form, saved_path, chain, value) # Never retain signatures, dates, validation status or generated barcodes. def manual(name): return name in ('TextField2', 'C1CertificateIssueDate', 'globValidatePressedState') or bool(re.search(r'(?i)(signature|barcode|validated|validationdate|datelastvalidated)', name)) for node in data.iter(): if manual(local(node)): for descendant in node.iter(): descendant.text = None updates = {} updates['datasets'] = replace_data_packet(original_datasets, data) if form is not None: for node in form.iter(): if manual(node.get('name', '')): for descendant in list(node.iter()): if local(descendant) == 'value' and local(descendant.getparent()) in ('field', 'exclGroup'): descendant.getparent().remove(descendant) prepare_phone_state(form, template, fields) # Reader discards saved state when this retains the blank's digest. # PDFium no longer checks it, so native rendering alone misses this. form.set('checksum', form_checksum(original_template, updates['datasets'])) updates['form'] = apply_state(E.tostring(form_tree), fields, count) # A full rewrite invalidates the signed Reader barcode rights. Append only # fill-in packets; never modify the signed template, permissions or issuer # signatures. All widget state deltas live in the saved form packet. output = Path(output_path) output.parent.mkdir(parents=True, exist_ok=True) with blank.open('rb') as source: writer = IncrementalPdfFileWriter(source) if writer.prev.encrypted: writer.encrypt('') xfa = writer.root['/AcroForm']['/XFA'] for index in range(0, len(xfa), 2): name = str(xfa[index]) if name in updates: reference = xfa.raw_get(index + 1) stream = generic.StreamObject(stream_data=updates[name]) stream.compress() writer.objects[(reference.generation, reference.idnum)] = stream with output.open('wb') as target: writer.write(target) return output