1"""Offline technical validation. No network calls or invoice logging."""
2from pathlib import Path
3from functools import lru_cache
4from decimal import Decimal, ROUND_FLOOR, localcontext
5from collections import Counter
6import hashlib, io, re
7from lxml import etree
8
9ROOT=Path(__file__).resolve().parents[1]
10CBC='urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2'
11CAC='urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2'
12NS={'cbc':CBC,'cac':CAC}
13PROFILES=('UBL-2.1-XSD','EN16931-1.3.17','EN16931-1.3.17+AFK-1.0')
14DEFAULT_PROFILE=PROFILES[-1]
15MAX_BYTES=2*1024*1024
16MAX_ELEMENTS=30000
17MAX_DEPTH=64
18SVRL={'s':'http://purl.oclc.org/dsdl/svrl'}
19
20class NoExternal(etree.Resolver):
21 def resolve(self,url,pubid,context):
22 raise OSError('External resource blocked')
23
24def parser():
25 p=etree.XMLParser(resolve_entities=False,load_dtd=False,no_network=True,
26 huge_tree=False,recover=False,remove_comments=True)
27 p.resolvers.add(NoExternal())
28 return p
29
30class SchemaOnly(etree.Resolver):
31 def resolve(self,url,pubid,context):
32 p=Path(url).resolve()
33 base=(ROOT/'vendor/ubl-2.1/xsd').resolve()
34 if p.is_relative_to(base) and p.suffix=='.xsd':
35 return self.resolve_filename(str(p),context)
36 raise OSError('Schema import outside bundled schemas')
37
38@lru_cache(maxsize=2)
39def schema(doc_type):
40 p=etree.XMLParser(resolve_entities=False,load_dtd=False,no_network=True)
41 p.resolvers.add(SchemaOnly())
42 return etree.XMLSchema(etree.parse(str(ROOT/f'vendor/ubl-2.1/xsd/maindoc/UBL-{doc_type}-2.1.xsd'),p))
43
44def qround(value):
45 return ((value*100+Decimal('.5')).to_integral_value(rounding=ROUND_FLOOR)/100)
46
47def dec(node,path,default='0'):
48 text=node.findtext(path,namespaces=NS)
49 return Decimal(text if text is not None else default)
50
51def sum_dec(nodes):
52 return sum((Decimal(n.text) for n in nodes),Decimal(0))
53
54def expected_totals(root):
55 t=root.find('cac:LegalMonetaryTotal',NS)
56 if t is None:return {}
57 lines=root.xpath('cac:InvoiceLine/cbc:LineExtensionAmount|cac:CreditNoteLine/cbc:LineExtensionAmount',namespaces=NS)
58 allowances=root.xpath("cac:AllowanceCharge[normalize-space(cbc:ChargeIndicator)='false' or normalize-space(cbc:ChargeIndicator)='0']/cbc:Amount",namespaces=NS)
59 charges=root.xpath("cac:AllowanceCharge[normalize-space(cbc:ChargeIndicator)='true' or normalize-space(cbc:ChargeIndicator)='1']/cbc:Amount",namespaces=NS)
60 curr=root.findtext('cbc:DocumentCurrencyCode',namespaces=NS)
61 taxes=root.xpath('cac:TaxTotal/cbc:TaxAmount[@currencyID=$c]',namespaces=NS,c=curr)
62 return {
63 'BR-CO-10':('cbc:LineExtensionAmount',qround(sum_dec(lines))),
64 'BR-CO-11':('cbc:AllowanceTotalAmount',qround(sum_dec(allowances))),
65 'BR-CO-12':('cbc:ChargeTotalAmount',qround(sum_dec(charges))),
66 'BR-CO-13':('cbc:TaxExclusiveAmount',qround(dec(t,'cbc:LineExtensionAmount')+dec(t,'cbc:ChargeTotalAmount')-dec(t,'cbc:AllowanceTotalAmount'))),
67 'BR-CO-15':('cbc:TaxInclusiveAmount',qround(dec(t,'cbc:TaxExclusiveAmount')+sum_dec(taxes))),
68 'BR-CO-16':('cbc:PayableAmount',qround(dec(t,'cbc:TaxInclusiveAmount')-dec(t,'cbc:PrepaidAmount')+dec(t,'cbc:PayableRoundingAmount')))
69 }
70
71def validate(xml:bytes,options:dict|None=None)->dict:
72 options=options or {}
73 if isinstance(xml,str):xml=xml.encode('utf-8')
74 profile=options.get('validationProfile',DEFAULT_PROFILE)
75 version=options.get('schemaVersion','2.1')
76 limit=options.get('maxFindings',100)
77 if isinstance(limit,bool) or not isinstance(limit,int) or not 1<=limit<=500:limit=100
78 rid='none' if profile=='UBL-2.1-XSD' else profile
79 report={'document_id':hashlib.sha256(xml).hexdigest()[:24],'invoice_number':None,
80 'document_type':None,'schema_version':version,'ruleset_version':rid,
81 'validation_profile':profile,'completed':True,'billable':True,'xsd_valid':None,
82 'profile_valid':None,'signature_verified':False,'findings':[],
83 'layers':{'xml':'pending','xsd':'not_run','profile':'not_requested' if profile=='UBL-2.1-XSD' else 'not_run','arithmetic':'not_requested'},
84 'error_count':0,'warning_count':0,'finding_count':0,'truncated':False,
85 'scope_notice':'Technical report only; no Peppol/network, country tax or legal acceptance. Digital signatures are not verified.'}
86 def add(layer,rule,severity,path,message,expected=None,actual=None):
87 report['finding_count']+=1
88 if severity=='error':report['error_count']+=1
89 if severity=='warning':report['warning_count']+=1
90 if len(report['findings'])<limit:
91 report['findings'].append({'document_id':report['document_id'],'invoice_number':report['invoice_number'],
92 'validation_layer':layer,'rule_id':rule,'severity':severity,'field_path':str(path or '/')[:1500],
93 'message':str(message)[:600],'expected_value':None if expected is None else str(expected)[:600],
94 'actual_value':None if actual is None else str(actual)[:300],
95 'schema_version':version,'ruleset_version':rid})
96 def finish():
97 report['truncated']=report['finding_count']>len(report['findings'])
98 report['status']='technical_failure' if not report['completed'] else ('invalid' if report['error_count'] else ('valid_with_warnings' if report['warning_count'] else 'valid'))
99 return report
100 def stop(rule,message,technical=False,layer='security'):
101 if technical:report.update(completed=False,billable=False)
102 add(layer,rule,'error','/',message)
103 return finish()
104 if profile not in PROFILES or version!='2.1' or options.get('documentType','auto') not in ('auto','Invoice','CreditNote'):
105 return stop('CONFIG-UNSUPPORTED','Unsupported schema version, profile or document type.',True,'configuration')
106 if 'maxFindings' in options and (isinstance(options['maxFindings'],bool) or not isinstance(options['maxFindings'],int) or not 1<=options['maxFindings']<=500):
107 return stop('CONFIG-LIMIT','maxFindings must be an integer from 1 to 500.',True,'configuration')
108 if len(xml)>MAX_BYTES:return stop('XML-SIZE','XML exceeds the 2 MiB limit.',True)
109 try:
110 tree=etree.parse(io.BytesIO(xml),parser())
111 except (etree.XMLSyntaxError,ValueError,OSError) as exc:
112
113 compact=xml.replace(b'\x00',b'').lower()
114 if b'<!doctype' in compact or b'<!entity' in compact:
115 return stop('XML-UNSAFE','DTD and entity declarations are forbidden.')
116 if any('depth' in e.message.lower() or 'resource limit' in e.message.lower() for e in getattr(exc,'error_log',[])):
117 return stop('XML-DEPTH','XML exceeds parser depth or structural safety limits.',True)
118 report['layers']['xml']='invalid'
119 return stop('XML-SYNTAX','XML is not well formed or exceeds parser safety limits.',False,'xml')
120 if tree.docinfo.doctype:return stop('XML-UNSAFE','DTD and entity declarations are forbidden.')
121 root=tree.getroot()
122 depth=0;elements=0
123 for event,node in etree.iterwalk(root,events=('start','end')):
124 if event=='start':
125 depth+=1;elements+=1
126 if depth>MAX_DEPTH:return stop('XML-DEPTH','XML exceeds the depth limit of 64.',True)
127 if elements>MAX_ELEMENTS:return stop('XML-ELEMENTS','XML exceeds 30000 elements.',True)
128 else:depth-=1
129 if root.xpath('//*[namespace-uri()="http://www.w3.org/2001/XInclude"]') or tree.xpath('//processing-instruction()'):
130 return stop('XML-UNSAFE','XInclude and processing instructions are forbidden.')
131 report['layers']['xml']='valid'
132 doc=etree.QName(root)
133 if doc.localname not in ('Invoice','CreditNote') or doc.namespace!=f'urn:oasis:names:specification:ubl:schema:xsd:{doc.localname}-2':
134 return stop('DOC-TYPE','Expected the exact UBL 2.1 Invoice or CreditNote namespace.',False,'xsd')
135 typ=doc.localname
136 report['document_type']=typ
137 report['invoice_number']=(root.findtext('cbc:ID',namespaces=NS) or '')[:200] or None
138 if options.get('documentType','auto') not in ('auto',typ):
139 return stop('DOC-TYPE','documentType does not match the XML root.',False,'xsd')
140 declared=root.findtext('cbc:UBLVersionID',namespaces=NS)
141 if declared and declared!='2.1':return stop('DOC-VERSION','Declared UBLVersionID does not match selected schema 2.1.',False,'xsd')
142 xsd=schema(typ)
143 valid=xsd.validate(tree)
144 report['xsd_valid']=valid;report['layers']['xsd']='valid' if valid else 'invalid'
145 if not valid:
146 for error in xsd.error_log:
147 add('xsd','XSD-'+error.type_name,'error',error.path,
148 error.message,expected='OASIS UBL 2.1 '+typ+' schema')
149 if profile!='UBL-2.1-XSD':report['layers']['profile']='skipped_xsd_invalid'
150 return finish()
151 if profile=='UBL-2.1-XSD':return finish()
152
153 from saxonche import PySaxonProcessor
154 with PySaxonProcessor(license=False) as proc:
155 proc.set_configuration_property('http://saxon.sf.net/feature/allow-external-functions','false')
156 proc.set_configuration_property('http://saxon.sf.net/feature/allowedProtocols','')
157 transformer=proc.new_xslt30_processor().compile_stylesheet(stylesheet_text=(ROOT/'vendor/en16931-1.3.17/xslt/EN16931-UBL-validation.xslt').read_text())
158 node=proc.parse_xml(xml_text=etree.tostring(root,encoding='unicode'))
159 svrl=transformer.transform_to_string(xdm_node=node)
160 result=etree.fromstring(svrl.encode(),parser())
161 totals=expected_totals(root)
162 arithmetic_prefix=('BR-CO-10','BR-CO-11','BR-CO-12','BR-CO-13','BR-CO-14','BR-CO-15','BR-CO-16','BR-CO-17','BR-S-08','BR-S-09','BR-DEC')
163 for finding in result.xpath('//s:failed-assert|//s:successful-report',namespaces=SVRL):
164 rule=finding.get('id','EN16931')
165 flag=finding.get('flag','fatal').lower()
166 severity='warning' if flag=='warning' else 'error'
167 expected=finding.get('test')
168 actual=None
169 path=finding.get('location','/')
170 if rule in totals:
171 name,amount=totals[rule]
172 expected=format(amount,'.2f')
173 actual=root.findtext('cac:LegalMonetaryTotal/'+name,namespaces=NS)
174 add('arithmetic' if rule.startswith(arithmetic_prefix) else 'profile',rule,severity,path,
175 ''.join(finding.xpath('s:text//text()',namespaces=SVRL)).strip(),expected,actual)
176 report['layers']['profile']='invalid' if report['error_count'] else 'valid'
177 report['layers']['arithmetic']='checked_official_EN16931'
178 if profile==DEFAULT_PROFILE:
179 with localcontext() as ctx:
180 ctx.prec=100
181 for line in root.xpath('cac:InvoiceLine|cac:CreditNoteLine',namespaces=NS):
182 path=tree.getpath(line)
183 base=dec(line,'cac:Price/cbc:BaseQuantity','1')
184 qty=dec(line,'cbc:InvoicedQuantity' if typ=='Invoice' else 'cbc:CreditedQuantity')
185 price=dec(line,'cac:Price/cbc:PriceAmount')
186 if base<=0:
187 add('arithmetic','AFK-BASE-01','error',path+'/cac:Price/cbc:BaseQuantity','Price base quantity must be greater than zero.', '> 0',base)
188 continue
189 charges=sum_dec(line.xpath("cac:AllowanceCharge[normalize-space(cbc:ChargeIndicator)='true' or normalize-space(cbc:ChargeIndicator)='1']/cbc:Amount",namespaces=NS))
190 allowances=sum_dec(line.xpath("cac:AllowanceCharge[normalize-space(cbc:ChargeIndicator)='false' or normalize-space(cbc:ChargeIndicator)='0']/cbc:Amount",namespaces=NS))
191 expected=qround(qty*price/base+charges-allowances)
192 actual=dec(line,'cbc:LineExtensionAmount')
193 if actual!=expected:
194 add('arithmetic','AFK-LINE-01','error',path+'/cbc:LineExtensionAmount',
195 'AFK selected rule: round(quantity × net price / base quantity + line charges − line allowances, 2), using XPath round ties toward positive infinity.',
196 format(expected,'.2f'),actual)
197 currency=root.findtext('cbc:DocumentCurrencyCode',namespaces=NS)
198 for amount in root.xpath('//*[@currencyID]',namespaces=NS):
199
200 if amount.getparent().tag=='{'+CAC+'}TaxTotal' and amount.tag=='{'+CBC+'}TaxAmount' and not amount.getparent().findall('cac:TaxSubtotal',NS):
201 if amount.get('currencyID')==root.findtext('cbc:TaxCurrencyCode',namespaces=NS):continue
202 if amount.get('currencyID')!=currency:
203 add('arithmetic','AFK-CURRENCY-01','error',tree.getpath(amount),
204 'AFK selected rule: monetary currency must match DocumentCurrencyCode (separate accounting-currency tax total excepted).',currency,amount.get('currencyID'))
205 report['layers']['arithmetic']='checked_EN16931_and_AFK'
206 report['profile_valid']=report['error_count']==0
207 return finish()
