Added Portal Sat Factura Electronica

satcfdi · satcfdi · commit 6cc232e9825e · 2022-11-28T01:25:11.000-06:00
diff --git a/docs/pages/getting_started/52_portal_facturacion.rst b/docs/pages/getting_started/52_portal_facturacion.rst
@@ -0,0 +1,40 @@
+Portal SAT - Factura Electrónica
+================================================
+
+Validación
+______________________
+
+.. code-block:: python
+
+    from satcfdi import Signer
+    from satcfdi.portal import SATFacturaElectronica
+    
+    # Load Fiel
+    signer = Signer.load(
+        certificate=open('csd/xiqb891116qe4.cer', 'rb').read(),
+        key=open('csd/xiqb891116qe4.key', 'rb').read(),
+        password=open('csd/xiqb891116qe4.txt', 'r').read()
+    )
+    
+    sat_session = SATFacturaElectronica(signer)
+    sat_session.login()
+    
+    
+    # Validación RFC
+    res = sat_session.rfc_valid(
+        rfc='XIQB891116QE4'
+    )
+    print(res)
+    
+    # Validación Razón Social
+    res = sat_session.legal_name_valid(
+        rfc='XIQB891116QE4',
+        legal_name='KIJ, S.A DE C.V.'
+    )
+    print(res)
+    
+    # LCO Detalles
+    res = sat_session.lco_details(rfc="XIQB891116QE4")
+    print(res)
+    
+
diff --git a/readme.rst b/readme.rst
@@ -74,10 +74,15 @@ ____________________
 * Listado 69B
 * Exportar Comprobantes a Excel
 * Descarga de Constancia de Situación Fiscal
+* Portal SAT - Factura Electrónica
+
+  * Validación de RFC, Razón Social
+  * LCO - Lista de Contribuyentes Obligados
 * DIOT - Declaración Informativa de Operaciones con Terceros
 * Certifica - Solicitud de Certificados, Renovación de Fiel
 * PLD - Prevención de Lavado de Dinero
 
+
 Installation
 ____________________
 
diff --git a/satcfdi/portal/__init__.py b/satcfdi/portal/__init__.py
@@ -1,83 +1,80 @@
+import json
 import pickle
+from time import time
+
 import requests
 import urllib3
 
-from .utils import get_post_form, generate_token
-from .. import Signer, __version__
+from .utils import get_post_form, generate_token, request_ref_headers, request_verification_token, random_ajax_id
+from .. import Signer, ResponseError
 
 CONSTANCIA_URL = 'https://rfcampc.siat.sat.gob.mx/PTSC/IdcSiat/IdcGeneraConstancia.jsf'
-USER_AGENT = __version__.__user_agent__
+USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36'
 
 DEFAULT_HEADERS = {
-    'accept': 'text/html,application/xhtml+xml,application/xml',
-    'pragma': 'no-cache',
-    'cache-control': 'no-cache',
-    'user-agent': USER_AGENT
+    'Accept': 'text/html,application/xhtml+xml,application/xml',
+    'Pragma': 'no-cache',
+    'Cache-Control': 'no-cache',
+    'User-Agent': USER_AGENT
 }
 
 
-def debug_response(res):
-    print('-- RESPONSE DEBUG --')
-    print(res.status_code)
-    for k, v in res.request.headers.items():
-        print(k, v)
-    print(res.request.body)
-    print('results')
-    print(res.text)
-    print('-- RESPONSE DEBUG END --')
-
-
-class SATSession:
+class PortalManager(requests.Session):
     def __init__(self, signer: Signer):
+        super().__init__()
         urllib3.util.ssl_.DEFAULT_CIPHERS += ':HIGH:!DH'
         self.signer = signer
-        self.session = requests.session()
 
-    def login(self):
-        LOGIN_URL = 'https://loginda.siat.sat.gob.mx/nidp/app/login?id=fiel'
-        LOGIN_URL_ORIGIN = 'https://loginda.siat.sat.gob.mx'
+    def save_session(self, target):
+        pickle.dump(self.cookies, target)
 
-        res = self.session.get(
-            url=LOGIN_URL,
-            headers=DEFAULT_HEADERS
-        )
-        assert res.status_code == 200
+    def load_session(self, source):
+        self.cookies.update(pickle.load(source))
 
-        action, data = get_post_form(res)
-        res = self.session.post(
+    def form_request(self, action, referer_url, data):
+        res = self.post(
             url=action,
-            headers=DEFAULT_HEADERS | {
-                'origin': LOGIN_URL_ORIGIN,
-                'referer': LOGIN_URL,
-            },
+            headers=DEFAULT_HEADERS | request_ref_headers(referer_url),
             data=data
         )
         assert res.status_code == 200
+        return res
 
-        action, data = get_post_form(res, id='certform')
-        res = self.session.post(
-            url=action,
-            headers=DEFAULT_HEADERS | {
-                'origin': LOGIN_URL_ORIGIN,
-                'referer': LOGIN_URL,
-            },
-            data=data | {
+    def fiel_login(self, login_response):
+        action, data = get_post_form(login_response, id='certform')
+        return self.form_request(
+            action,
+            login_response.request.url,
+            data | {
                 'token': generate_token(self.signer, code=data['guid']),
                 'fert': self.signer.certificate.get_notAfter()[2:].decode(),
             }
         )
+
+
+class SATPortal(PortalManager):
+    def login(self):
+        LOGIN_URL = 'https://loginda.siat.sat.gob.mx/nidp/app/login?id=fiel'
+
+        res = self.get(
+            url=LOGIN_URL,
+            headers=DEFAULT_HEADERS
+        )
         assert res.status_code == 200
 
-        return res
+        action, data = get_post_form(res)
+        return self.fiel_login(
+            login_response=self.form_request(action, res.request.url, data)
+        )
 
     def home_page(self):
-        return self.session.get(
+        return self.get(
             url='https://loginda.siat.sat.gob.mx/nidp/app?sid=0',
             headers=DEFAULT_HEADERS
         )
 
     def logout(self):
-        return self.session.get(
+        return self.get(
             url='https://loginda.siat.sat.gob.mx/nidp/app/logout',
             headers=DEFAULT_HEADERS | {
                 'referer': 'https://loginda.siat.sat.gob.mx/nidp/app?sid=0'
@@ -86,27 +83,146 @@ def logout(self):
         )
 
     def declaraciones_provisionales_login(self):
-        res = self.session.get(
-            url='https://ptscdecprov.clouda.sat.gob.mx/',
+        res = self.get(
+            url='https://ptscdecprov.clouda.sat.gob.mx',
             headers=DEFAULT_HEADERS,
             allow_redirects=True
         )
         assert res.status_code == 200
 
         action, data = get_post_form(res)
-        res = self.session.post(
-            url=action,
+        res = self.form_request(action, res.request.url, data)
+        return res
+
+
+class SATFacturaElectronica(PortalManager):
+    BASE_URL = 'https://portal.facturaelectronica.sat.gob.mx'
+    REQUEST_CONTEXT = 'appId=cid-v1:20ff76f4-0bca-495f-b7fd-09ca520e39f7'
+
+    def __init__(self, signer: Signer):
+        super().__init__(signer)
+        self._ajax_id = random_ajax_id()
+        self._request_verification_token = None
+
+    def login(self):
+        res = self.get(
+            url=self.BASE_URL,
+            headers=DEFAULT_HEADERS
+        )
+        assert res.status_code == 200
+
+        try:
+            action, data = get_post_form(res)
+        except IndexError as ex:
+            raise ValueError("Login form not found, please try again") from ex
+
+        if action.startswith('https://cfdiau.sat.gob.mx/'):
+            assert 'nidp/wsfed/ep?id=SATUPCFDiCon' in action
+
+            res = self.fiel_login(
+                login_response=self.form_request(
+                    action.replace('nidp/wsfed/ep?id=SATUPCFDiCon', 'nidp/app/login?id=SATx509Custom'),
+                    res.request.url,
+                    data
+                )
+            )
+
+            action, data = get_post_form(res)
+            res = self.form_request(action, res.request.url, data)
+
+            action, data = get_post_form(res)
+
+        res = self.form_request(action, res.request.url, data)
+
+        self._request_verification_token = request_verification_token(res)
+        self._ajax_id = random_ajax_id()
+        return res
+
+    def _reload_verification_token(self):
+        res = self.get(
+            url=f'{self.BASE_URL}/Factura/GeneraFactura',
+            headers=DEFAULT_HEADERS,
+            allow_redirects=False
+        )
+        if res.status_code == 200:
+            self._request_verification_token = request_verification_token(res)
+        else:
+            raise ValueError('Please Login Again')
+
+    def reactivate_session(self):
+        res = self.post(
+            url=f'{self.BASE_URL}/Home/ReActiveSession',
             headers=DEFAULT_HEADERS | {
-                'content-type': 'application/x-www-form-urlencoded',
-                'origin': 'https://loginda.siat.sat.gob.mx',
-                'referer': 'https://loginda.siat.sat.gob.mx/',
+                'Origin': f'{self.BASE_URL}',
+                'Request-Context': self.REQUEST_CONTEXT,
+                'Request-Id': f'|{self._ajax_id}.{random_ajax_id()}'
             },
-            data=data
+            allow_redirects=False
         )
         return res
 
-    def save_session(self, target):
-        pickle.dump(self.session.cookies, target)
+    def _request(self, method, path, data=None, params=None):
+        if self._request_verification_token is None:
+            self._reload_verification_token()
 
-    def load_session(self, source):
-        self.session.cookies.update(pickle.load(source))
+        if method.upper() == 'POST':
+            headers = {
+                'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8'
+            }
+        else:
+            headers = {}
+
+        res = self.request(
+            method=method,
+            url=f'{self.BASE_URL}/{path}',
+            headers=DEFAULT_HEADERS | headers | {
+                'Origin': self.BASE_URL,
+                'Authority': self.BASE_URL,
+                'Request-Context': self.REQUEST_CONTEXT,
+                '__RequestVerificationToken': self._request_verification_token,
+                'Request-Id': f'|{self._ajax_id}.{random_ajax_id()}'  # |pR4Px.o0yAS
+            },
+            data=data,
+            params=params,
+            allow_redirects=False
+        )
+        if res.status_code == 200:
+            return res.json()
+        else:
+            raise ResponseError(res)
+
+    def legal_name_valid(self, rfc, legal_name):
+        res = self._request(
+            method='POST',
+            path='Clientes/ValidaRazonSocialRFC',
+            data={
+                'rfcValidar': rfc.upper(),
+                'razonSocial': legal_name.upper(),
+            })
+        if not res['exitoso']:
+            raise ResponseError(res)
+        return res['resultado']
+
+    def rfc_valid(self, rfc):
+        res = self._request(
+            method='POST',
+            path='Clientes/ExisteLrfc',
+            data={
+                'rfcValidar': rfc.upper()
+            }
+        )
+        if not res['exitoso']:
+            raise ResponseError(res)
+        return res['resultado']
+
+    def lco_details(self, rfc, apply_border_region=True):
+        res = self._request(
+            method='GET',
+            path='Clientes/ValidaLco',
+            params={
+                'rfcValidar': rfc.upper(),
+                'aplicaRegionFronteriza': apply_border_region,
+                "_": int(time() * 1000)
+            }
+        )
+        return json.loads(res)
diff --git a/satcfdi/portal/utils.py b/satcfdi/portal/utils.py
@@ -1,4 +1,6 @@
 import base64
+import random
+import string
 from urllib.parse import urlparse, urlunparse
 
 from bs4 import BeautifulSoup
@@ -67,3 +69,19 @@ def get_post_form(res: Response, id=None):
     assert form.attrs['method'].upper() == "POST"
     return action_url(form.attrs.get('action'), res.url), data
 
+
+def request_ref_headers(url):
+    parts = urlparse(url)
+    return {
+        'origin': urlunparse((parts.scheme, parts.netloc, '', '', '', '')),
+        'referer': url
+    }
+
+
+def request_verification_token(res: Response):
+    html = BeautifulSoup(res.text, 'html.parser')
+    return html.find(name='input', attrs={'name': '__RequestVerificationToken'}).attrs['value']
+
+
+def random_ajax_id():
+    return ''.join(random.choice(string.ascii_letters + string.digits) for _ in range(5))
diff --git a/tests/test_portal.py b/tests/test_portal.py
@@ -1,7 +1,7 @@
 import os
 
 from satcfdi.portal.utils import generate_token, verify_token, action_url
-from satcfdi.portal import SATSession
+from satcfdi.portal import SATPortal
 from tests.utils import get_signer
 
 current_dir = os.path.dirname(__file__)
@@ -39,7 +39,7 @@ def test_action_url():
 
 def test_start_session():
     signer = get_signer('cacx7605101p8')
-    res = SATSession(signer)
+    res = SATPortal(signer)
 
     try:
         with open(session_file, 'rb') as f:
