satcfdi.portal.SATFacturaElectronica._reload_verification_token() - Code Metrics - Inspection of "fix SSL" - SAT-CFDI/python-satcfdi - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( cf8f99...b508f6 )

by Sat CFDI

created 2023-11-06 18:32 UTC

SATFacturaElectronica._reload_verification_token() A

↳ Parent: satcfdi.portal

Complexity

Conditions

Size

Total Lines	9
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	4.048

Importance

Changes

Metric	Value
cc	2
eloc	7
nop	1
dl	0
loc	9
ccs	1
cts	5
cp	0.2
crap	4.048
rs	10
c	0
b	0
f	0

import json
import pickle
from time import time

import requests
from requests.structures import CaseInsensitiveDict

from .utils import get_form, generate_token, request_ref_headers, request_verification_token, random_ajax_id, SSLAdapter
from ..exceptions import ResponseError
from ..models import Signer


class PortalManager(requests.Session):
    def __init__(self, signer: Signer):
        super().__init__()
        self.mount('https://', SSLAdapter())
        self.signer = signer

        self.headers = CaseInsensitiveDict(
            {
                "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',
                "Accept-Encoding": 'gzip, deflate, br',
                "Accept": 'text/html,application/xhtml+xml,application/xml',
                "Connection": "keep-alive",
                'Pragma': 'no-cache',
                'Cache-Control': 'no-cache',
            }
        )

    def save_session(self, target):
        pickle.dump(self.cookies, target)

    def load_session(self, source):
        self.cookies.update(pickle.load(source))

    def form(self, action, referer_url, data):
        res = self.post(
            url=action,
            headers=request_ref_headers(referer_url),
            data=data
        )
        assert res.status_code == 200
        return res

    def fiel_login(self, login_response):
        action, data = get_form(login_response, id='certform')
        return self.form(
            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):
    BASE_URL = 'https://loginda.siat.sat.gob.mx'

    def login(self):
        res = self.get(
            url=f'{self.BASE_URL}/nidp/app/login?id=fiel'
        )
        assert res.status_code == 200
        action, data = get_form(res)

        return self.fiel_login(
            login_response=self.form(action, res.request.url, data)
        )

    def home_page(self):
        return self.get(
            url=f'{self.BASE_URL}/nidp/app?sid=0'
        )

    def logout(self):
        return self.get(
            url=f'{self.BASE_URL}/nidp/app/logout',
            headers={
                'referer': f'{self.BASE_URL}/nidp/app?sid=0'
            },
            allow_redirects=False
        )

    def declaraciones_provisionales_login(self):
        res = self.get(
            url='https://ptscdecprov.clouda.sat.gob.mx',
            allow_redirects=True
        )
        assert res.status_code == 200

        action, data = get_form(res)
        res = self.form(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
        )
        assert res.status_code == 200
        try:
            action, data = get_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(
                    action.replace('nidp/wsfed/ep?id=SATUPCFDiCon', 'nidp/app/login?id=SATx509Custom'),
                    res.request.url,
                    data
                )
            )

            action, data = get_form(res)
            res = self.form(action, res.request.url, data)

            action, data = get_form(res)

        res = self.form(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',
            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={
                'Origin': self.BASE_URL,
                'Request-Context': self.REQUEST_CONTEXT,
                'Request-Id': f'|{self._ajax_id}.{random_ajax_id()}'
            },
            allow_redirects=False
        )
        return res

    def _request(self, method, path, data=None, params=None):
        if self._request_verification_token is None:
            self._reload_verification_token()

        res = self.request(
            method=method,
            url=f'{self.BASE_URL}/{path}',
            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)


class SATPortalConstancia(PortalManager):
    BASE_URL = 'https://login.siat.sat.gob.mx'

    def login(self):
        res = self.get(
            url=f'{self.BASE_URL}/nidp/idff/sso?id=fiel_Aviso'
        )
        assert res.status_code == 200
        action, data = get_form(res)
        if action is None:
            return res

        return self.fiel_login(
            login_response=self.form(action, res.request.url, data)
        )

    def generar_constancia(self):
        self.login()
        res = self.get(
            url="https://rfcampc.siat.sat.gob.mx/app/seg/SessionBroker?url=/PTSC/IdcSiat/autc/ReimpresionTramite/ConsultaTramite.jsf&parametro=c&idSessionBit=&idSessionBit=null",
            allow_redirects=True
        )
        assert res.status_code == 200

        # Execute Authentication Request
        action, data = get_form(res)
        if action == "https://login.siat.sat.gob.mx/nidp/saml2/sso":
            res = self.form(action, res.request.url, data)
            assert res.status_code == 200

            # Execute Authentication Response
            action, data = get_form(res)
            assert action == "https://rfcampc.siat.sat.gob.mx/saml2/sp/acs/post"
            res = self.form(action, res.request.url, data)
            assert res.status_code == 200

        # Execute formReimpAcuse
        action, data = get_form(res)
        if action == "https://rfcampc.siat.sat.gob.mx/PTSC/IdcSiat/autc/ReimpresionTramite/ConsultaTramite.jsf":
            data = {
                'javax.faces.partial.ajax': "true",
                'javax.faces.source': 'formReimpAcuse:j_idt50',
                'javax.faces.partial.execute': '@all',
                'formReimpAcuse:j_idt50': 'formReimpAcuse:j_idt50',
                'formReimpAcuse': 'formReimpAcuse',
                'formReimpAcuse:tipoTramite_input': '0',
                'formReimpAcuse:tipoTramite_focus': '',
                'formReimpAcuse:fechaInicio_input:': '',
                'formReimpAcuse:fechaFin_input': '',
                'formReimpAcuse:folio': '',
                'javax.faces.ViewState': data['javax.faces.ViewState']
            }
            res = self.form(action, res.request.url, data)
            assert res.status_code == 200

        res = self.get(
            url='https://rfcampc.siat.sat.gob.mx/PTSC/IdcSiat/IdcGeneraConstancia.jsf'
        )
        assert res.status_code == 200
        return res.content


class SATPortalOpinionCumplimiento(PortalManager):
    BASE_URL = 'https://login.mat.sat.gob.mx'

    def login(self):
        res = self.get(
            url=f'{self.BASE_URL}/nidp/app/login?id=contr-dual-eFirma-totp'
        )
        assert res.status_code == 200
        action, data = get_form(res)

        res = self.form(action, res.request.url, data)
        assert res.status_code == 200

        res = self.form(action, res.request.url, data)
        assert res.status_code == 200

        res = self.form(action, res.request.url, data)
        assert res.status_code == 200

        return res

    def generar_opinion_cumplimiento_login(self):
        res = self.get(
            url="https://ptsc32d.clouda.sat.gob.mx/?/reporteOpinion32DContribuyente",
            allow_redirects=True
        )
        assert res.status_code == 200

        # Execute Authentication Request
        action, data = get_form(res)
        if action.startswith("https://login.mat.sat.gob.mx/nidp//app/login"):
            res = self.form(action, res.request.url, data)
            assert res.status_code == 200

            # Execute Authentication Response
            action, data = get_form(res)
            res = self.form(action, res.request.url, data)
            assert res.status_code == 200


1	1	import json
2	1	import pickle
3	1	from time import time
4
5	1	import requests
6	1	from requests.structures import CaseInsensitiveDict
7
8	1	from .utils import get_form, generate_token, request_ref_headers, request_verification_token, random_ajax_id, SSLAdapter
9	1	from ..exceptions import ResponseError
10	1	from ..models import Signer
11
12
13	1	class PortalManager(requests.Session):
14	1	def __init__(self, signer: Signer):
15	1	super().__init__()
16	1	self.mount('https://', SSLAdapter())
17	1	self.signer = signer
18
19	1	self.headers = CaseInsensitiveDict(
20		{
21		"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',
22		"Accept-Encoding": 'gzip, deflate, br',
23		"Accept": 'text/html,application/xhtml+xml,application/xml',
24		"Connection": "keep-alive",
25		'Pragma': 'no-cache',
26		'Cache-Control': 'no-cache',
27		}
28		)
29
30	1	def save_session(self, target):
31	1	pickle.dump(self.cookies, target)
32
33	1	def load_session(self, source):
34		self.cookies.update(pickle.load(source))
35
36	1	def form(self, action, referer_url, data):
37		res = self.post(
38		url=action,
39		headers=request_ref_headers(referer_url),
40		data=data
41		)
42		assert res.status_code == 200
43		return res
44
45	1	def fiel_login(self, login_response):
46		action, data = get_form(login_response, id='certform')
47		return self.form(
48		action,
49		login_response.request.url,
50		data \| {
51		'token': generate_token(self.signer, code=data['guid']),
52		'fert': self.signer.certificate.get_notAfter()[2:].decode(),
53		}
54		)
55
56
57	1	class SATPortal(PortalManager):
58	1	BASE_URL = 'https://loginda.siat.sat.gob.mx'
59
60	1	def login(self):
61		res = self.get(
62		url=f'{self.BASE_URL}/nidp/app/login?id=fiel'
63		)
64		assert res.status_code == 200
65		action, data = get_form(res)
66
67		return self.fiel_login(
68		login_response=self.form(action, res.request.url, data)
69		)
70
71	1	def home_page(self):
72		return self.get(
73		url=f'{self.BASE_URL}/nidp/app?sid=0'
74		)
75
76	1	def logout(self):
77		return self.get(
78		url=f'{self.BASE_URL}/nidp/app/logout',
79		headers={
80		'referer': f'{self.BASE_URL}/nidp/app?sid=0'
81		},
82		allow_redirects=False
83		)
84
85	1	def declaraciones_provisionales_login(self):
86		res = self.get(
87		url='https://ptscdecprov.clouda.sat.gob.mx',
88		allow_redirects=True
89		)
90		assert res.status_code == 200
91
92		action, data = get_form(res)
93		res = self.form(action, res.request.url, data)
94		return res
95
96
97	1	class SATFacturaElectronica(PortalManager):
98	1	BASE_URL = 'https://portal.facturaelectronica.sat.gob.mx'
99	1	REQUEST_CONTEXT = 'appId=cid-v1:20ff76f4-0bca-495f-b7fd-09ca520e39f7'
100
101	1	def __init__(self, signer: Signer):
102		super().__init__(signer)
103		self._ajax_id = random_ajax_id()
104		self._request_verification_token = None
105
106	1	def login(self):
107		res = self.get(
108		url=self.BASE_URL
109		)
110		assert res.status_code == 200
111		try:
112		action, data = get_form(res)
113		except IndexError as ex:
114		raise ValueError("Login form not found, please try again") from ex
115
116		if action.startswith('https://cfdiau.sat.gob.mx/'):
117		assert 'nidp/wsfed/ep?id=SATUPCFDiCon' in action
118
119		res = self.fiel_login(
120		login_response=self.form(
121		action.replace('nidp/wsfed/ep?id=SATUPCFDiCon', 'nidp/app/login?id=SATx509Custom'),
122		res.request.url,
123		data
124		)
125		)
126
127		action, data = get_form(res)
128		res = self.form(action, res.request.url, data)
129
130		action, data = get_form(res)
131
132		res = self.form(action, res.request.url, data)
133
134		self._request_verification_token = request_verification_token(res)
135		self._ajax_id = random_ajax_id()
136		return res
137
138	1	def _reload_verification_token(self):
139		res = self.get(
140		url=f'{self.BASE_URL}/Factura/GeneraFactura',
141		allow_redirects=False
142		)
143		if res.status_code == 200:
144		self._request_verification_token = request_verification_token(res)
145		else:
146		raise ValueError('Please Login Again')
147
148	1	def reactivate_session(self):
149		res = self.post(
150		url=f'{self.BASE_URL}/Home/ReActiveSession',
151		headers={
152		'Origin': self.BASE_URL,
153		'Request-Context': self.REQUEST_CONTEXT,
154		'Request-Id': f'\|{self._ajax_id}.{random_ajax_id()}'
155		},
156		allow_redirects=False
157		)
158		return res
159
160	1	def _request(self, method, path, data=None, params=None):
161		if self._request_verification_token is None:
162		self._reload_verification_token()
163
164		res = self.request(
165		method=method,
166		url=f'{self.BASE_URL}/{path}',
167		headers={
168		'Origin': self.BASE_URL,
169		'Authority': self.BASE_URL,
170		'Request-Context': self.REQUEST_CONTEXT,
171		'__RequestVerificationToken': self._request_verification_token,
172		'Request-Id': f'\|{self._ajax_id}.{random_ajax_id()}' # \|pR4Px.o0yAS
173		},
174		data=data,
175		params=params,
176		allow_redirects=False
177		)
178		if res.status_code == 200:
179		return res.json()
180		else:
181		raise ResponseError(res)
182
183	1	def legal_name_valid(self, rfc, legal_name):
184		res = self._request(
185		method='POST',
186		path='Clientes/ValidaRazonSocialRFC',
187		data={
188		'rfcValidar': rfc.upper(),
189		'razonSocial': legal_name.upper(),
190		})
191		if not res['exitoso']:
192		raise ResponseError(res)
193		return res['resultado']
194
195	1	def rfc_valid(self, rfc):
196		res = self._request(
197		method='POST',
198		path='Clientes/ExisteLrfc',
199		data={
200		'rfcValidar': rfc.upper()
201		}
202		)
203		if not res['exitoso']:
204		raise ResponseError(res)
205		return res['resultado']
206
207	1	def lco_details(self, rfc, apply_border_region=True):
208		res = self._request(
209		method='GET',
210		path='Clientes/ValidaLco',
211		params={
212		'rfcValidar': rfc.upper(),
213		'aplicaRegionFronteriza': apply_border_region,
214		"_": int(time() * 1000)
215		}
216		)
217		return json.loads(res)
218
219
220	1	class SATPortalConstancia(PortalManager):
221	1	BASE_URL = 'https://login.siat.sat.gob.mx'
222
223	1	def login(self):
224		res = self.get(
225		url=f'{self.BASE_URL}/nidp/idff/sso?id=fiel_Aviso'
226		)
227		assert res.status_code == 200
228		action, data = get_form(res)
229		if action is None:
230		return res
231
232		return self.fiel_login(
233		login_response=self.form(action, res.request.url, data)
234		)
235
236	1	def generar_constancia(self):
237		self.login()
238		res = self.get(
239		url="https://rfcampc.siat.sat.gob.mx/app/seg/SessionBroker?url=/PTSC/IdcSiat/autc/ReimpresionTramite/ConsultaTramite.jsf&parametro=c&idSessionBit=&idSessionBit=null",
240		allow_redirects=True
241		)
242		assert res.status_code == 200
243
244		# Execute Authentication Request
245		action, data = get_form(res)
246		if action == "https://login.siat.sat.gob.mx/nidp/saml2/sso":
247		res = self.form(action, res.request.url, data)
248		assert res.status_code == 200
249
250		# Execute Authentication Response
251		action, data = get_form(res)
252		assert action == "https://rfcampc.siat.sat.gob.mx/saml2/sp/acs/post"
253		res = self.form(action, res.request.url, data)
254		assert res.status_code == 200
255
256		# Execute formReimpAcuse
257		action, data = get_form(res)
258		if action == "https://rfcampc.siat.sat.gob.mx/PTSC/IdcSiat/autc/ReimpresionTramite/ConsultaTramite.jsf":
259		data = {
260		'javax.faces.partial.ajax': "true",
261		'javax.faces.source': 'formReimpAcuse:j_idt50',
262		'javax.faces.partial.execute': '@all',
263		'formReimpAcuse:j_idt50': 'formReimpAcuse:j_idt50',
264		'formReimpAcuse': 'formReimpAcuse',
265		'formReimpAcuse:tipoTramite_input': '0',
266		'formReimpAcuse:tipoTramite_focus': '',
267		'formReimpAcuse:fechaInicio_input:': '',
268		'formReimpAcuse:fechaFin_input': '',
269		'formReimpAcuse:folio': '',
270		'javax.faces.ViewState': data['javax.faces.ViewState']
271		}
272		res = self.form(action, res.request.url, data)
273		assert res.status_code == 200
274
275		res = self.get(
276		url='https://rfcampc.siat.sat.gob.mx/PTSC/IdcSiat/IdcGeneraConstancia.jsf'
277		)
278		assert res.status_code == 200
279		return res.content
280
281
282	1	class SATPortalOpinionCumplimiento(PortalManager):
283	1	BASE_URL = 'https://login.mat.sat.gob.mx'
284
285	1	def login(self):
286		res = self.get(
287		url=f'{self.BASE_URL}/nidp/app/login?id=contr-dual-eFirma-totp'
288		)
289		assert res.status_code == 200
290		action, data = get_form(res)
291
292		res = self.form(action, res.request.url, data)
293		assert res.status_code == 200
294
295		res = self.form(action, res.request.url, data)
296		assert res.status_code == 200
297
298		res = self.form(action, res.request.url, data)
299		assert res.status_code == 200
300
301		return res
302
303	1	def generar_opinion_cumplimiento_login(self):
304		res = self.get(
305		url="https://ptsc32d.clouda.sat.gob.mx/?/reporteOpinion32DContribuyente",
306		allow_redirects=True
307		)
308		assert res.status_code == 200
309
310		# Execute Authentication Request
311		action, data = get_form(res)
312		if action.startswith("https://login.mat.sat.gob.mx/nidp//app/login"):
313		res = self.form(action, res.request.url, data)
314		assert res.status_code == 200
315
316		# Execute Authentication Response
317		action, data = get_form(res)
318		res = self.form(action, res.request.url, data)
319		assert res.status_code == 200
320

SAT-CFDI / python-satcfdi

Push — main ( cf8f99...b508f6 )

SATFacturaElectronica._reload_verification_token() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like