OneTeam/oc-facho
16
0
Fork 0
Code Issues 2 Pull Requests Packages Projects Releases Wiki Activity

Fix: Formateo PEP8 (79 cols) y limpieza flake8 en paquete facho

Browse Source 
- Líneas >79 cols reajustadas en todo el paquete
- Fix F821: etree.cleanup_namespaces(security) en signature.py (era 'header', NameError)
- Fix bug cune_xpath -> xpath en nomina/informacion_general() (NameError)
- Elimina clase TaxScheme duplicada en form/__init__.py
- Imports explícitos (F403/F405) y __all__ en form_xml/invoice.py, nomina
- Elimina imports y variables muertas (F401/F841), E712 (== None -> is None)
This commit is contained in:
aserrador
2026-08-12 19:56:29 -05:00
parent 26ce89ca33
commit 4a7c2e2947
37 changed files with 1915 additions and 1103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
facho/fe/client/dian.py
View File

@@ -1,23 +1,13 @@
from facho import facho
import zeep
from zeep.wsse.username import UsernameToken
from .wsse.signature import Signature, BinarySignature
from zeep.wsa import WsAddressingPlugin
from .wsse.signature import BinarySignature
import xmlsec
import urllib.request
from datetime import datetime
from dataclasses import dataclass, asdict, field
import http.client
import hashlib
import secrets
import base64
from dataclasses import dataclass, asdict
__all__ = ['DianClient',
'ConsultaResolucionesFacturacionPeticion',
'ConsultaResolucionesFacturacionRespuesta']
__all__ = ['DianClient']
class SOAPService:
@@ -33,6 +23,7 @@ class SOAPService:
def todict(self):
return asdict(self)
@dataclass
class GetNumberingRangeResponse:
@@ -49,7 +40,6 @@ class GetNumberingRangeResponse:
NumberRangeResponse: list[NumberRangeResponse]
@classmethod
def fromdict(cls, data):
return cls(
@@ -100,6 +90,7 @@ class SendTestSetAsyncResponse:
data['ErrorMessageList'] or []
)
@dataclass
class SendTestSetAsync(SOAPService):
fileName: str
@@ -115,6 +106,7 @@ class SendTestSetAsync(SOAPService):
def build_response(self, as_dict):
return SendTestSetAsyncResponse.fromdict(as_dict)
@dataclass
class SendBillSync(SOAPService):
fileName: str
@@ -129,6 +121,7 @@ class SendBillSync(SOAPService):
def build_response(self, as_dict):
return as_dict
@dataclass
class GetStatusResponse:
IsValid: bool
@@ -142,12 +135,12 @@ class GetStatusResponse:
error_message = data['ErrorMessage']['string']
else:
error_message = None
return cls(data['IsValid'],
data['StatusDescription'],
data['StatusCode'],
error_message)
@dataclass
class GetStatus(SOAPService):
@@ -162,6 +155,7 @@ class GetStatus(SOAPService):
def build_response(self, as_dict):
return GetStatusResponse.fromdict(as_dict)
@dataclass
class GetStatusZip(SOAPService):
trackId: bytes
@@ -175,6 +169,7 @@ class GetStatusZip(SOAPService):
def build_response(self, as_dict):
return GetStatusResponse.fromdict(as_dict[0])
@dataclass
class SendNominaSync(SOAPService):
contentFile: bytes
@@ -188,7 +183,7 @@ class SendNominaSync(SOAPService):
def build_response(self, as_dict):
return as_dict
class Habilitacion:
WSDL = 'https://vpfe-hab.dian.gov.co/WcfDianCustomerServices.svc?wsdl'
@@ -220,6 +215,7 @@ class Habilitacion:
def get_wsdl(self):
return Habilitacion.WSDL
class DianGateway:
def _open(self, service):
@@ -250,7 +246,11 @@ class DianClient(DianGateway):
self._password = password
def _open(self, service):
return zeep.Client(service.get_wsdl(), wsse=UsernameToken(self._username, self._password))
return zeep.Client(
service.get_wsdl(),
wsse=UsernameToken(
self._username,
self._password))
class DianSignatureClient(DianGateway):
@@ -262,13 +262,10 @@ class DianSignatureClient(DianGateway):
def _open(self, service):
# RESOLUCCION 0004: pagina 756
from zeep.wsse import utils
client = zeep.Client(service.get_wsdl(), wsse=
BinarySignature(
self.private_key_path, self.public_key_path, self.password,
signature_method=xmlsec.Transform.RSA_SHA256,
digest_method=xmlsec.Transform.SHA256)
,
client = zeep.Client(service.get_wsdl(), wsse=BinarySignature(
self.private_key_path, self.public_key_path, self.password,
signature_method=xmlsec.Transform.RSA_SHA256,
digest_method=xmlsec.Transform.SHA256),
)
return client

68
facho/fe/client/wsse/signature.py
View File

@@ -70,8 +70,11 @@ class MemorySignature(object):
def apply(self, envelope, headers):
key = _make_sign_key(self.key_data, self.cert_data, self.password)
_sign_envelope_with_key(
envelope, key, self.signature_method, self.digest_method, expires_dt=self.expires_dt
)
envelope,
key,
self.signature_method,
self.digest_method,
expires_dt=self.expires_dt)
return envelope, headers
def verify(self, envelope):
@@ -81,7 +84,7 @@ class MemorySignature(object):
class Signature(MemorySignature):
"""Sign given SOAP envelope with WSSE sig using given key file and cert file."""
"""Sign given SOAP envelope with WSSE sig using given key and cert file."""
def __init__(
self,
@@ -101,15 +104,18 @@ class Signature(MemorySignature):
class BinarySignature(Signature):
"""Sign given SOAP envelope with WSSE sig using given key file and cert file.
"""Sign given SOAP envelope with WSSE sig using given key and cert file.
Place the key information into BinarySecurityElement."""
def apply(self, envelope, headers):
key = _make_sign_key(self.key_data, self.cert_data, self.password)
_sign_envelope_with_key_binary(
envelope, key, self.signature_method, self.digest_method, expires_dt = self.expires_dt
)
envelope,
key,
self.signature_method,
self.digest_method,
expires_dt=self.expires_dt)
return envelope, headers
@@ -219,9 +225,11 @@ def sign_envelope(
"""
# Load the signing key and certificate.
key = _make_sign_key(_read_file(keyfile), _read_file(certfile), password)
return _sign_envelope_with_key(envelope, key, signature_method, digest_method)
return _sign_envelope_with_key(
envelope, key, signature_method, digest_method)
def get_timestamp(timestamp = None, delta=None):
def get_timestamp(timestamp=None, delta=None):
timestamp = timestamp or datetime.now(timezone.utc)
if delta:
timestamp += delta
@@ -230,24 +238,33 @@ def get_timestamp(timestamp = None, delta=None):
timestamp = timestamp.replace(tzinfo=pytz.utc, microsecond=0)
return timestamp.strftime(format_)
def _append_timestamp(security, expires_dt=None):
if expires_dt is None:
expires_dt = timedelta(seconds=6000)
etimestamp = utils.WSU.Timestamp({'{http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd}Id': utils.get_unique_id()})
etimestamp = utils.WSU.Timestamp({
'{http://docs.oasis-open.org/wss/2004/01/'
'oasis-200401-wss-wssecurity-utility-1.0.xsd}Id':
utils.get_unique_id()})
etimestamp.append(utils.WSU.Created(get_timestamp()))
etimestamp.append(utils.WSU.Expires(get_timestamp(delta=expires_dt)))
security.insert(0, etimestamp)
if etree.LXML_VERSION[:2] >= (3, 5):
etree.cleanup_namespaces(security,
keep_ns_prefixes = security.nsmap,
keep_ns_prefixes=security.nsmap,
top_nsmap=utils.NSMAP)
else:
etree.cleanup_namespaces(header)
etree.cleanup_namespaces(security)
def _signature_prepare(envelope, key, signature_method, digest_method, expires_dt=None):
def _signature_prepare(
envelope,
key,
signature_method,
digest_method,
expires_dt=None):
"""Prepare envelope and sign."""
soap_env = detect_soap_env(envelope)
# Create the Signature node.
signature = xmlsec.template.create(
@@ -278,7 +295,7 @@ def _signature_prepare(envelope, key, signature_method, digest_method, expires_d
_append_timestamp(security, expires_dt=expires_dt)
timestamp = security.find(QName(ns.WSU, "Timestamp"))
if timestamp != None:
if timestamp is not None:
_sign_node(ctx, signature, timestamp, digest_method)
ctx.sign(signature)
@@ -286,18 +303,30 @@ def _signature_prepare(envelope, key, signature_method, digest_method, expires_d
# KeyInfo. The recipient expects this structure, but we can't rearrange
# like this until after signing, because otherwise xmlsec won't populate
# the X509 data (because it doesn't understand WSSE).
sec_token_ref = etree.SubElement(key_info, QName(ns.WSSE, "SecurityTokenReference"))
sec_token_ref = etree.SubElement(
key_info, QName(
ns.WSSE, "SecurityTokenReference"))
return security, sec_token_ref, x509_data
def _sign_envelope_with_key(envelope, key, signature_method, digest_method, expires_dt=None):
def _sign_envelope_with_key(
envelope,
key,
signature_method,
digest_method,
expires_dt=None):
_, sec_token_ref, x509_data = _signature_prepare(
envelope, key, signature_method, digest_method, expires_dt=expires_dt
)
sec_token_ref.append(x509_data)
def _sign_envelope_with_key_binary(envelope, key, signature_method, digest_method, expires_dt=None):
def _sign_envelope_with_key_binary(
envelope,
key,
signature_method,
digest_method,
expires_dt=None):
security, sec_token_ref, x509_data = _signature_prepare(
envelope, key, signature_method, digest_method, expires_dt=expires_dt
)
@@ -353,7 +382,10 @@ def _verify_envelope_with_key(envelope, key):
ctx = xmlsec.SignatureContext()
# Find each signed element and register its ID with the signing context.
refs = signature.xpath("ds:SignedInfo/ds:Reference", namespaces={"ds": ns.DS})
refs = signature.xpath(
"ds:SignedInfo/ds:Reference",
namespaces={
"ds": ns.DS})
for ref in refs:
# Get the reference URI and cut off the initial '#'
referenced_id = ref.get("URI")[1:]