Move VAT ID validation logic from vat_moss to core, support Norway

src/pretix/base/models/tax.py

@@ -114,7 +114,7 @@ EU_CURRENCIES = {
     'RO': 'RON',
     'SE': 'SEK'
 }
-VAT_ID_COUNTRIES = EU_COUNTRIES | {'CH'}
+VAT_ID_COUNTRIES = EU_COUNTRIES | {'CH', 'NO'}
 
 
 def is_eu_country(cc):

src/pretix/base/services/tax.py

@@ -22,9 +22,9 @@
 import logging
 import os
 import re
-from urllib.error import HTTPError
+from xml.etree import ElementTree
 
-import vat_moss.errors
+import requests
 import vat_moss.id
 from django.conf import settings
 from django.utils.translation import gettext_lazy as _
@@ -35,6 +35,16 @@ from zeep.exceptions import Fault
 from pretix.base.models.tax import cc_to_vat_prefix, is_eu_country
 
 logger = logging.getLogger(__name__)
+error_messages = {
+    'unavailable': _(
+        'Your VAT ID could not be checked, as the VAT checking service of '
+        'your country is currently not available. We will therefore '
+        'need to charge VAT on your invoice. You can get the tax amount '
+        'back via the VAT reimbursement process.'
+    ),
+    'invalid': _('This VAT ID is not valid. Please re-check your input.'),
+    'country_mismatch': _('Your VAT ID does not match the selected country.'),
+}
 
 
 class VATIDError(Exception):
@@ -50,33 +60,97 @@ class VATIDTemporaryError(VATIDError):
     pass
 
 
-def _validate_vat_id_EU(vat_id, country_code):
-    if vat_id[:2] != cc_to_vat_prefix(country_code):
-        raise VATIDFinalError(_('Your VAT ID does not match the selected country.'))
+def _validate_vat_id_NO(vat_id, country_code):
+    # Inspired by vat_moss library
+    vat_id = vat_moss.id.normalize(vat_id)
+
+    if not vat_id or len(vat_id) < 3 or not re.match('^\\d{9}MVA$', vat_id[2:]):
+        raise VATIDFinalError(error_messages['invalid'])
+
+    organization_number = vat_id[2:].replace('MVA', '')
+    validation_url = 'https://data.brreg.no/enhetsregisteret/api/enheter/%s' % organization_number
 
     try:
-        result = vat_moss.id.validate(vat_id)
-        if result:
-            country_code, normalized_id, company_name = result
-            return normalized_id
-    except (vat_moss.errors.InvalidError, ValueError):
-        raise VATIDFinalError(_('This VAT ID is not valid. Please re-check your input.'))
-    except vat_moss.errors.WebServiceUnavailableError:
+        response = requests.get(validation_url, timeout=10)
+        if response.status_code in (404, 400):
+            raise VATIDFinalError(error_messages['invalid'])
+
+        response.raise_for_status()
+
+        info = response.json()
+        # This should never happen, but keeping it incase the API is changed
+        if 'organisasjonsnummer' not in info or info['organisasjonsnummer'] != organization_number:
+            logger.warning(
+                'VAT ID checking failed for Norway due to missing or mismatching organisasjonsnummer in repsonse'
+            )
+            raise VATIDFinalError(error_messages['invalid'])
+    except requests.RequestException:
         logger.exception('VAT ID checking failed for country {}'.format(country_code))
-        raise VATIDTemporaryError(_(
-            'Your VAT ID could not be checked, as the VAT checking service of '
-            'your country is currently not available. We will therefore '
-            'need to charge VAT on your invoice. You can get the tax amount '
-            'back via the VAT reimbursement process.'
-        ))
-    except (vat_moss.errors.WebServiceError, HTTPError):
+        raise VATIDTemporaryError(error_messages['unavailable'])
+    else:
+        return vat_id
+
+
+def _validate_vat_id_EU(vat_id, country_code):
+    # Inspired by vat_moss library
+    vat_id = vat_moss.id.normalize(vat_id)
+    number = vat_id[2:]
+
+    if vat_id[:2] != cc_to_vat_prefix(country_code):
+        raise VATIDFinalError(error_messages['country_mismatch'])
+
+    if not re.match(vat_moss.id.ID_PATTERNS[country_code]['regex'], number):
+        raise VATIDFinalError(error_messages['invalid'])
+
+    payload = """
+            <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:urn="urn:ec.europa.eu:taxud:vies:services:checkVat:types">
+               <soapenv:Header/>
+               <soapenv:Body>
+                  <urn:checkVat>
+                     <urn:countryCode>%s</urn:countryCode>
+                     <urn:vatNumber>%s</urn:vatNumber>
+                  </urn:checkVat>
+               </soapenv:Body>
+            </soapenv:Envelope>
+    """.strip() % (country_code, number)
+
+    try:
+        response = requests.post(
+            'https://ec.europa.eu/taxation_customs/vies/services/checkVatService',
+            data=payload,
+            timeout=10,
+        )
+        response.raise_for_status()
+
+        return_xml = response.text
+
+        try:
+            envelope = ElementTree.fromstring(return_xml)
+        except ElementTree.ParseError:
+            logger.error(
+                f'VAT ID checking failed for {country_code} due to XML parse error'
+            )
+            raise VATIDTemporaryError(error_messages['unavailable'])
+
+        namespaces = {
+            'soap': 'http://schemas.xmlsoap.org/soap/envelope/',
+            'vat': 'urn:ec.europa.eu:taxud:vies:services:checkVat:types'
+        }
+        valid_elements = envelope.findall('./soap:Body/vat:checkVatResponse/vat:valid', namespaces)
+        if not valid_elements:
+            logger.error(
+                f'VAT ID checking failed for {country_code} due to missing <valid> tag'
+            )
+            raise VATIDTemporaryError(error_messages['unavailable'])
+
+        if valid_elements[0].text.lower() != 'true':
+            raise VATIDFinalError(error_messages['invalid'])
+
+    except requests.RequestException:
         logger.exception('VAT ID checking failed for country {}'.format(country_code))
-        raise VATIDTemporaryError(_(
-            'Your VAT ID could not be checked, as the VAT checking service of '
-            'your country returned an incorrect result. We will therefore '
-            'need to charge VAT on your invoice. Please contact support to '
-            'resolve this manually.'
-        ))
+        raise VATIDTemporaryError(error_messages['unavailable'])
+    else:
+        return vat_id
 
 
 def _validate_vat_id_CH(vat_id, country_code):
@@ -85,10 +159,13 @@ def _validate_vat_id_CH(vat_id, country_code):
 
     vat_id = re.sub('[^A-Z0-9]', '', vat_id.replace('HR', '').replace('MWST', ''))
     try:
-        transport = Transport(cache=SqliteCache(os.path.join(settings.CACHE_DIR, "validate_vat_id_ch_zeep_cache.db")))
+        transport = Transport(
+            cache=SqliteCache(os.path.join(settings.CACHE_DIR, "validate_vat_id_ch_zeep_cache.db")),
+            timeout=10
+        )
         client = Client(
             'https://www.uid-wse.admin.ch/V5.0/PublicServices.svc?wsdl',
-            transport=transport
+            transport=transport,
         )
         result = client.service.ValidateUID(uid=vat_id)
     except Fault as e:
@@ -125,10 +202,14 @@ def _validate_vat_id_CH(vat_id, country_code):
 
 
 def validate_vat_id(vat_id, country_code):
+    if not vat_id:
+        return vat_id
     country_code = str(country_code)
     if is_eu_country(country_code):
         return _validate_vat_id_EU(vat_id, country_code)
     elif country_code == 'CH':
         return _validate_vat_id_CH(vat_id, country_code)
+    elif country_code == 'NO':
+        return _validate_vat_id_NO(vat_id, country_code)
 
     raise VATIDTemporaryError(f'VAT ID should not be entered for country {country_code}')