cfdi-descarga/source/sat/sat_web.py

#!/usr/bin/env python3

import base64
import hashlib
import uuid
from datetime import datetime, timedelta

import httpx
import lxml.etree as ET


class SATWebService():
    BASE = 'https://cfdidescargamasivasolicitud.clouda.sat.gob.mx'
    URL = {
        'AUTH': f'{BASE}/Autenticacion/Autenticacion.svc',
        'REQ': f'{BASE}/SolicitaDescargaService.svc',
    }
    XMLNS = 'http://DescargaMasivaTerceros.gob.mx'
    ACTIONS = {
        'AUTH': f'{XMLNS}/IAutenticacion/Autentica',
        'REQ': f'{XMLNS}/ISolicitaDescargaService/SolicitaDescarga',
    }
    HEADERS = {
        'Content-type': 'text/xml;charset="utf-8"',
        'Accept': 'text/xml',
        'Cache-Control': 'no-cache',
    }
    NS = {
        's': 'http://schemas.xmlsoap.org/soap/envelope/',
        'u': 'http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd',
        'o': 'http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-secext-1.0.xsd'
        'des': 'http://DescargaMasivaTerceros.sat.gob.mx',
        'xd': 'http://www.w3.org/2000/09/xmldsig#',
    }

    def __init__(self, cert):
        self._cert = cert
        self._error = ''
        self._token = self._get_token()

    @property
    def is_authenticate(self):
        return bool(self._token)

    @property
    def error(self):
        return self._error

    def _get_data_auth(self):
        NSMAP = {'s': self.NS['s'], 'u': self.NS['u']}
        FORMAT = '%Y-%m-%dT%H:%M:%S.%fZ'
        UID = str(uuid.uuid4())

        now = datetime.utcnow()
        date_created = now.strftime(FORMAT)
        date_expires = (now + timedelta(seconds=300)).strftime(FORMAT)

        node_name = f"{{{self.NS['s']}}}Envelope"
        root = ET.Element(node_name, nsmap=NSMAP)

        node_name = f"{{{self.NS['s']}}}Header"
        header = ET.SubElement(root, node_name)

        node_name = f"{{{self.NS['o']}}}Security"
        nsmap = {'o': self.NS['o']}
        attr_name = f"{{{self.NS['s']}}}mustUnderstand"
        attr = {attr_name: '1'}
        security = ET.SubElement(header, node_name, attr, nsmap=nsmap)

        node_name = f"{{{self.NS['u']}}}Timestamp"
        attr_name = f"{{{self.NS['u']}}}Id"
        attr = {attr_name: '_0'}
        timestamp = ET.SubElement(security, node_name, attr)
        node_name = f"{{{self.NS['u']}}}Created"
        ET.SubElement(timestamp, node_name).text = date_created
        node_name = f"{{{self.NS['u']}}}Expires"
        ET.SubElement(timestamp, node_name).text = date_expires

        node_name = f"{{{self.NS['o']}}}BinarySecurityToken"
        attr = {
            f"{{{self.NS['u']}}}Id": UID,
            'ValueType': 'http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-x509-token-profile-1.0#X509v3',
            'EncodingType': 'http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-soap-message-security-1.0#Base64Binary'
        }
        ET.SubElement(security, node_name, attr).text = self._cert.cer_txt

        nsmap = {None: 'http://www.w3.org/2000/09/xmldsig#'}
        signature = ET.SubElement(security, 'Signature', nsmap=nsmap)
        signedinfo = ET.SubElement(signature, 'SignedInfo')
        attr1 = {'Algorithm': 'http://www.w3.org/2001/10/xml-exc-c14n#'}
        ET.SubElement(signedinfo, 'CanonicalizationMethod', attr1)
        attr = {'Algorithm': 'http://www.w3.org/2000/09/xmldsig#rsa-sha1'}
        ET.SubElement(signedinfo, 'SignatureMethod', attr)

        attr = {'URI': '#_0'}
        reference = ET.SubElement(signedinfo, 'Reference', attr)
        transforms = ET.SubElement(reference, 'Transforms')
        ET.SubElement(transforms, 'Transform', attr1)
        attr = {'Algorithm': 'http://www.w3.org/2000/09/xmldsig#sha1'}
        ET.SubElement(reference, 'DigestMethod', attr)

        dvalue = ET.tostring(timestamp, method='c14n', exclusive=1)
        dvalue = base64.b64encode(hashlib.new('sha1', dvalue).digest())
        ET.SubElement(reference, 'DigestValue').text = dvalue

        signature_value = ET.tostring(signedinfo, method='c14n', exclusive=1)
        signature_value = self._cert.sign_sha1(signature_value)
        ET.SubElement(signature, 'SignatureValue').text = signature_value
        keyinfo = ET.SubElement(signature, 'KeyInfo')

        node_name = f"{{{self.NS['o']}}}SecurityTokenReference"
        security_token = ET.SubElement(keyinfo, node_name)

        node_name = f"{{{self.NS['o']}}}Reference"
        attr = {
            'ValueType': 'http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-x509-token-profile-1.0#X509v3',
            'URI': f'#{UID}',
        }
        ET.SubElement(security_token, node_name, attr)

        node_name = f"{{{self.NS['s']}}}Body"
        body = ET.SubElement(root, node_name)
        nsmap = {None: self.XMLNS}
        ET.SubElement(body, 'Autentica', nsmap=nsmap)

        # ~ soap = ET.tostring(root, pretty_print=True, encoding='utf-8')
        soap = ET.tostring(root)

        return soap

    def _get_token(self):
        headers = self.HEADERS.copy()
        headers['SOAPAction'] = self.ACTIONS['AUTH']
        data = self._get_data_auth()

        response = httpx.post(self.URL['AUTH'], data=data, headers=headers)
        if response.status_code != httpx.codes.OK:
            self._error = f'Status: {response.status_code} - {response.text}'
            return

        result = ET.fromstring(response.text)
        nsmap = {'s': self.NS['s'], None: self.XMLNS}
        node_name = 's:Body/AutenticaResponse/AutenticaResult'
        token = result.find(node_name, namespaces=nsmap).text

        return token

    def _get_data_req(self, args):

        return

    def _get_request(self, args):
        headers = self.HEADERS.copy()
        headers['SOAPAction'] = self.ACTIONS['REQ']
        headers['Authorization'] = f'WRAP access_token="{self._token}"'
        data = self._get_data_req(args)

        response = httpx.post(self.URL['REQ'], data=data, headers=headers)
        if response.status_code != httpx.codes.OK:
            self._error = f'Status: {response.status_code} - {response.text}'
            return

        print(response.text)
        result = ET.fromstring(response.text)

        return

    def download(self, args):
        request = self._get_request(args)
        print(request)
        return