satcfdi.portal.SATPortalConstancia.generar_constancia_login() - Code Metrics - Inspection of "added class to get constancia" - SAT-CFDI/python-satcfdi - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( 88e831...35fe8a )

by Sat CFDI

created 2023-11-06 17:35 UTC

SATPortalConstancia.generar_constancia_login() A

↳ Parent: satcfdi.portal

Complexity

Conditions

Size

Total Lines	38
Code Lines	30

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	10.5039

Importance

Changes

Metric	Value
cc	3
eloc	30
nop	1
dl	0
loc	38
ccs	1
cts	17
cp	0.0588
crap	10.5039
rs	9.16
c	0
b	0
f	0

import json
import pickle
from time import time

import requests
import urllib3
from requests.structures import CaseInsensitiveDict

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


class PortalManager(requests.Session):
    def __init__(self, signer: Signer):
        super().__init__()
        try:
            urllib3.util.ssl_.DEFAULT_CIPHERS += ':HIGH:!DH'
        except:
            pass
        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_login(self):
        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
        return res


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

SAT-CFDI / python-satcfdi

Push — main ( 88e831...35fe8a )

SATPortalConstancia.generar_constancia_login() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like