satcfdi.portal.SATPortal.home_page() - Code Metrics - Inspection of "Merge pull request #3 from SAT-CFDI/F2022_11_27" - SAT-CFDI/python-satcfdi - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( 6294a4...0182a3 )

by Sat CFDI

created 2022-11-28 07:21 UTC

satcfdi.portal.SATPortal.home_page() A

↳ Parent: satcfdi.portal

Complexity

Conditions

Size

Total Lines	4
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	1.125

Importance

Changes

Metric	Value
cc	1
eloc	4
nop	1
dl	0
loc	4
ccs	1
cts	2
cp	0.5
crap	1.125
rs	10
c	0
b	0
f	0

import json
import pickle
from time import time

import requests
import urllib3

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 = '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
}


class PortalManager(requests.Session):
    def __init__(self, signer: Signer):
        super().__init__()
        urllib3.util.ssl_.DEFAULT_CIPHERS += ':HIGH:!DH'
        self.signer = signer

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

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

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

    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

        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.get(
            url='https://loginda.siat.sat.gob.mx/nidp/app?sid=0',
            headers=DEFAULT_HEADERS
        )

    def logout(self):
        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'
            },
            allow_redirects=False
        )

    def declaraciones_provisionales_login(self):
        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.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 | {
                'Origin': f'{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()

        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)


1	1	import json
2	1	import pickle
3	1	from time import time
4
5	1	import requests
6	1	import urllib3
7
8	1	from .utils import get_post_form, generate_token, request_ref_headers, request_verification_token, random_ajax_id
9	1	from .. import Signer, ResponseError
10
11	1	CONSTANCIA_URL = 'https://rfcampc.siat.sat.gob.mx/PTSC/IdcSiat/IdcGeneraConstancia.jsf'
12	1	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'
13
14	1	DEFAULT_HEADERS = {
15		'Accept': 'text/html,application/xhtml+xml,application/xml',
16		'Pragma': 'no-cache',
17		'Cache-Control': 'no-cache',
18		'User-Agent': USER_AGENT
19		}
20
21
22	1	class PortalManager(requests.Session):
23	1	def __init__(self, signer: Signer):
24	1	super().__init__()
25	1	urllib3.util.ssl_.DEFAULT_CIPHERS += ':HIGH:!DH'
26	1	self.signer = signer
27
28	1	def save_session(self, target):
29	1	pickle.dump(self.cookies, target)
30
31	1	def load_session(self, source):
32		self.cookies.update(pickle.load(source))
33
34	1	def form_request(self, action, referer_url, data):
35		res = self.post(
36		url=action,
37		headers=DEFAULT_HEADERS \| request_ref_headers(referer_url),
38		data=data
39		)
40		assert res.status_code == 200
41		return res
42
43	1	def fiel_login(self, login_response):
44		action, data = get_post_form(login_response, id='certform')
45		return self.form_request(
46		action,
47		login_response.request.url,
48		data \| {
49		'token': generate_token(self.signer, code=data['guid']),
50		'fert': self.signer.certificate.get_notAfter()[2:].decode(),
51		}
52		)
53
54
55	1	class SATPortal(PortalManager):
56	1	def login(self):
57		LOGIN_URL = 'https://loginda.siat.sat.gob.mx/nidp/app/login?id=fiel'
58
59		res = self.get(
60		url=LOGIN_URL,
61		headers=DEFAULT_HEADERS
62		)
63		assert res.status_code == 200
64
65		action, data = get_post_form(res)
66		return self.fiel_login(
67		login_response=self.form_request(action, res.request.url, data)
68		)
69
70	1	def home_page(self):
71		return self.get(
72		url='https://loginda.siat.sat.gob.mx/nidp/app?sid=0',
73		headers=DEFAULT_HEADERS
74		)
75
76	1	def logout(self):
77		return self.get(
78		url='https://loginda.siat.sat.gob.mx/nidp/app/logout',
79		headers=DEFAULT_HEADERS \| {
80		'referer': 'https://loginda.siat.sat.gob.mx/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		headers=DEFAULT_HEADERS,
89		allow_redirects=True
90		)
91		assert res.status_code == 200
92
93		action, data = get_post_form(res)
94		res = self.form_request(action, res.request.url, data)
95		return res
96
97
98	1	class SATFacturaElectronica(PortalManager):
99	1	BASE_URL = 'https://portal.facturaelectronica.sat.gob.mx'
100	1	REQUEST_CONTEXT = 'appId=cid-v1:20ff76f4-0bca-495f-b7fd-09ca520e39f7'
101
102	1	def __init__(self, signer: Signer):
103		super().__init__(signer)
104		self._ajax_id = random_ajax_id()
105		self._request_verification_token = None
106
107	1	def login(self):
108		res = self.get(
109		url=self.BASE_URL,
110		headers=DEFAULT_HEADERS
111		)
112		assert res.status_code == 200
113
114		try:
115		action, data = get_post_form(res)
116		except IndexError as ex:
117		raise ValueError("Login form not found, please try again") from ex
118
119		if action.startswith('https://cfdiau.sat.gob.mx/'):
120		assert 'nidp/wsfed/ep?id=SATUPCFDiCon' in action
121
122		res = self.fiel_login(
123		login_response=self.form_request(
124		action.replace('nidp/wsfed/ep?id=SATUPCFDiCon', 'nidp/app/login?id=SATx509Custom'),
125		res.request.url,
126		data
127		)
128		)
129
130		action, data = get_post_form(res)
131		res = self.form_request(action, res.request.url, data)
132
133		action, data = get_post_form(res)
134
135		res = self.form_request(action, res.request.url, data)
136
137		self._request_verification_token = request_verification_token(res)
138		self._ajax_id = random_ajax_id()
139		return res
140
141	1	def _reload_verification_token(self):
142		res = self.get(
143		url=f'{self.BASE_URL}/Factura/GeneraFactura',
144		headers=DEFAULT_HEADERS,
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=DEFAULT_HEADERS \| {
156		'Origin': f'{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		if method.upper() == 'POST':
169		headers = {
170		'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8'
171		}
172		else:
173		headers = {}
174
175		res = self.request(
176		method=method,
177		url=f'{self.BASE_URL}/{path}',
178		headers=DEFAULT_HEADERS \| headers \| {
179		'Origin': self.BASE_URL,
180		'Authority': self.BASE_URL,
181		'Request-Context': self.REQUEST_CONTEXT,
182		'__RequestVerificationToken': self._request_verification_token,
183		'Request-Id': f'\|{self._ajax_id}.{random_ajax_id()}' # \|pR4Px.o0yAS
184		},
185		data=data,
186		params=params,
187		allow_redirects=False
188		)
189		if res.status_code == 200:
190		return res.json()
191		else:
192		raise ResponseError(res)
193
194	1	def legal_name_valid(self, rfc, legal_name):
195		res = self._request(
196		method='POST',
197		path='Clientes/ValidaRazonSocialRFC',
198		data={
199		'rfcValidar': rfc.upper(),
200		'razonSocial': legal_name.upper(),
201		})
202		if not res['exitoso']:
203		raise ResponseError(res)
204		return res['resultado']
205
206	1	def rfc_valid(self, rfc):
207		res = self._request(
208		method='POST',
209		path='Clientes/ExisteLrfc',
210		data={
211		'rfcValidar': rfc.upper()
212		}
213		)
214		if not res['exitoso']:
215		raise ResponseError(res)
216		return res['resultado']
217
218	1	def lco_details(self, rfc, apply_border_region=True):
219		res = self._request(
220		method='GET',
221		path='Clientes/ValidaLco',
222		params={
223		'rfcValidar': rfc.upper(),
224		'aplicaRegionFronteriza': apply_border_region,
225		"_": int(time() * 1000)
226		}
227		)
228		return json.loads(res)
229

SAT-CFDI / python-satcfdi

Push — main ( 6294a4...0182a3 )

satcfdi.portal.SATPortal.home_page() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like