Browse Source

resolvendo um conflito aqui

tags/0.1
Italo Maia 16 years ago
parent
commit
40e3cce4e1
  1. 121
      pynfe/processamento/assinatura.py
  2. 86
      pynfe/processamento/comunicacao.py
  3. 4
      pynfe/processamento/serializacao.py
  4. 6
      pynfe/utils/__init__.py
  5. 103
      run_fake_soap_server.py
  6. 2
      tests/03-processamento-01-serializacao-xml.txt
  7. 58
      tests/03-processamento-04-comunicacao.txt
  8. 27
      tests/certificado.pem
  9. 15
      tests/key.pem

121
pynfe/processamento/assinatura.py

@ -1,89 +1,84 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
import xmlsec, libxml2 # FIXME: verificar ambiguidade de dependencias: lxml e libxml2 import xmlsec, libxml2 # FIXME: verificar ambiguidade de dependencias: lxml e libxml2
from geraldo.utils import memoize
from pynfe.utils import etree, StringIO
from pynfe.utils import etree, StringIO, extrair_tag
from pynfe.utils.flags import NAMESPACE_NFE, NAMESPACE_SIG from pynfe.utils.flags import NAMESPACE_NFE, NAMESPACE_SIG
class Assinatura(object): class Assinatura(object):
"""Classe abstrata responsavel por definir os metodos e logica das classes """Classe abstrata responsavel por definir os metodos e logica das classes
de assinatura digital."""
de assinatura digital."""
certificado = None certificado = None
senha = None senha = None
def __init__(self, certificado, senha): def __init__(self, certificado, senha):
self.certificado = certificado self.certificado = certificado
self.senha = senha self.senha = senha
def assinar_arquivo(self, caminho_arquivo): def assinar_arquivo(self, caminho_arquivo):
"""Efetua a assinatura dos arquivos XML informados""" """Efetua a assinatura dos arquivos XML informados"""
pass pass
def assinar_xml(self, xml): def assinar_xml(self, xml):
"""Efetua a assinatura numa string contendo XML valido.""" """Efetua a assinatura numa string contendo XML valido."""
pass pass
def assinar_etree(self, raiz): def assinar_etree(self, raiz):
u"""Efetua a assinatura numa instancia da biblioteca lxml.etree. u"""Efetua a assinatura numa instancia da biblioteca lxml.etree.
Este metodo de assinatura será utilizado internamente pelos demais,
sendo que eles convertem para uma instancia lxml.etree para somente
depois efetivar a assinatura.
TODO: Verificar o funcionamento da PyXMLSec antes de efetivar isso."""
Este metodo de assinatura será utilizado internamente pelos demais,
sendo que eles convertem para uma instancia lxml.etree para somente
depois efetivar a assinatura.
TODO: Verificar o funcionamento da PyXMLSec antes de efetivar isso."""
pass pass
def assinar_objetos(self, objetos): def assinar_objetos(self, objetos):
"""Efetua a assinatura em instancias do PyNFe""" """Efetua a assinatura em instancias do PyNFe"""
pass pass
def verificar_arquivo(self, caminho_arquivo): def verificar_arquivo(self, caminho_arquivo):
pass pass
def verificar_xml(self, xml): def verificar_xml(self, xml):
pass pass
def verificar_etree(self, raiz): def verificar_etree(self, raiz):
pass pass
def verificar_objetos(self, objetos): def verificar_objetos(self, objetos):
pass pass
@memoize
def extrair_tag(root):
return root.tag.split('}')[-1]
class AssinaturaA1(Assinatura): class AssinaturaA1(Assinatura):
"""Classe abstrata responsavel por efetuar a assinatura do certificado """Classe abstrata responsavel por efetuar a assinatura do certificado
digital no XML informado."""
digital no XML informado."""
def assinar_arquivo(self, caminho_arquivo, salva=True): def assinar_arquivo(self, caminho_arquivo, salva=True):
# Carrega o XML do arquivo # Carrega o XML do arquivo
raiz = etree.parse(caminho_arquivo) raiz = etree.parse(caminho_arquivo)
# Efetua a assinatura # Efetua a assinatura
xml = self.assinar_etree(raiz, retorna_xml=True) xml = self.assinar_etree(raiz, retorna_xml=True)
# Grava XML assinado no arquivo # Grava XML assinado no arquivo
if salva: if salva:
fp = file(caminho_arquivo, 'w') fp = file(caminho_arquivo, 'w')
fp.write(xml) fp.write(xml)
fp.close() fp.close()
return xml return xml
def assinar_xml(self, xml): def assinar_xml(self, xml):
raiz = etree.parse(StringIO(xml)) raiz = etree.parse(StringIO(xml))
# Efetua a assinatura # Efetua a assinatura
return self.assinar_etree(raiz, retorna_xml=True) return self.assinar_etree(raiz, retorna_xml=True)
def assinar_etree(self, raiz, retorna_xml=False): def assinar_etree(self, raiz, retorna_xml=False):
# Extrai a tag do elemento raiz # Extrai a tag do elemento raiz
tipo = extrair_tag(raiz.getroot()) tipo = extrair_tag(raiz.getroot())
# doctype compatível com o tipo da tag raiz # doctype compatível com o tipo da tag raiz
if tipo == u'NFe': if tipo == u'NFe':
doctype = u'<!DOCTYPE NFe [<!ATTLIST infNFe Id ID #IMPLIED>]>' doctype = u'<!DOCTYPE NFe [<!ATTLIST infNFe Id ID #IMPLIED>]>'
@ -93,7 +88,7 @@ digital no XML informado."""
doctype = u'<!DOCTYPE cancNFe [<!ATTLIST infCanc Id ID #IMPLIED>]>' doctype = u'<!DOCTYPE cancNFe [<!ATTLIST infCanc Id ID #IMPLIED>]>'
elif tipo == u'DPEC': elif tipo == u'DPEC':
doctype = u'<!DOCTYPE DPEC [<!ATTLIST infDPEC Id ID #IMPLIED>]>' doctype = u'<!DOCTYPE DPEC [<!ATTLIST infDPEC Id ID #IMPLIED>]>'
# Tag de assinatura # Tag de assinatura
if raiz.getroot().find('Signature') is None: if raiz.getroot().find('Signature') is None:
signature = etree.Element( signature = etree.Element(
@ -101,35 +96,35 @@ digital no XML informado."""
URI=raiz.getroot().getchildren()[0].attrib['Id'], URI=raiz.getroot().getchildren()[0].attrib['Id'],
nsmap={'sig': NAMESPACE_SIG}, nsmap={'sig': NAMESPACE_SIG},
) )
signed_info = etree.SubElement(signature, '{%s}SignedInfo'%NAMESPACE_SIG) signed_info = etree.SubElement(signature, '{%s}SignedInfo'%NAMESPACE_SIG)
etree.SubElement(signed_info, 'CanonicalizationMethod', Algorithm="http://www.w3.org/TR/2001/REC-xml-c14n-20010315") etree.SubElement(signed_info, 'CanonicalizationMethod', Algorithm="http://www.w3.org/TR/2001/REC-xml-c14n-20010315")
etree.SubElement(signed_info, 'SignatureMethod', Algorithm="http://www.w3.org/2000/09/xmldsig#rsa-sha1") etree.SubElement(signed_info, 'SignatureMethod', Algorithm="http://www.w3.org/2000/09/xmldsig#rsa-sha1")
reference = etree.SubElement(signed_info, '{%s}Reference'%NAMESPACE_SIG, URI=raiz.getroot().getchildren()[0].attrib['Id']) reference = etree.SubElement(signed_info, '{%s}Reference'%NAMESPACE_SIG, URI=raiz.getroot().getchildren()[0].attrib['Id'])
transforms = etree.SubElement(reference, 'Transforms', URI=raiz.getroot().getchildren()[0].attrib['Id']) transforms = etree.SubElement(reference, 'Transforms', URI=raiz.getroot().getchildren()[0].attrib['Id'])
etree.SubElement(transforms, 'Transform', Algorithm="http://www.w3.org/2000/09/xmldsig#enveloped-signature") etree.SubElement(transforms, 'Transform', Algorithm="http://www.w3.org/2000/09/xmldsig#enveloped-signature")
etree.SubElement(transforms, 'Transform', Algorithm="http://www.w3.org/TR/2001/REC-xml-c14n-20010315") etree.SubElement(transforms, 'Transform', Algorithm="http://www.w3.org/TR/2001/REC-xml-c14n-20010315")
etree.SubElement(reference, '{%s}DigestMethod'%NAMESPACE_SIG, Algorithm="http://www.w3.org/2000/09/xmldsig#sha1") etree.SubElement(reference, '{%s}DigestMethod'%NAMESPACE_SIG, Algorithm="http://www.w3.org/2000/09/xmldsig#sha1")
digest_value = etree.SubElement(reference, '{%s}DigestValue'%NAMESPACE_SIG) digest_value = etree.SubElement(reference, '{%s}DigestValue'%NAMESPACE_SIG)
signature_value = etree.SubElement(signature, '{%s}SignatureValue'%NAMESPACE_SIG) signature_value = etree.SubElement(signature, '{%s}SignatureValue'%NAMESPACE_SIG)
key_info = etree.SubElement(signature, '{%s}KeyInfo'%NAMESPACE_SIG) key_info = etree.SubElement(signature, '{%s}KeyInfo'%NAMESPACE_SIG)
x509_data = etree.SubElement(key_info, '{%s}X509Data'%NAMESPACE_SIG) x509_data = etree.SubElement(key_info, '{%s}X509Data'%NAMESPACE_SIG)
x509_certificate = etree.SubElement(x509_data, '{%s}X509Certificate'%NAMESPACE_SIG) x509_certificate = etree.SubElement(x509_data, '{%s}X509Certificate'%NAMESPACE_SIG)
raiz.getroot().insert(0, signature) raiz.getroot().insert(0, signature)
# Acrescenta a tag de doctype (como o lxml nao suporta alteracao do doctype, # Acrescenta a tag de doctype (como o lxml nao suporta alteracao do doctype,
# converte para string para faze-lo) # converte para string para faze-lo)
xml = etree.tostring(raiz, xml_declaration=True, encoding='utf-8') xml = etree.tostring(raiz, xml_declaration=True, encoding='utf-8')
if xml.find('<!DOCTYPE ') == -1: if xml.find('<!DOCTYPE ') == -1:
pos = xml.find('>') + 1 pos = xml.find('>') + 1
xml = xml[:pos] + doctype + xml[pos:] xml = xml[:pos] + doctype + xml[pos:]
#raiz = etree.parse(StringIO(xml)) #raiz = etree.parse(StringIO(xml))
doc_xml, ctxt, noh_assinatura, assinador = self._antes_de_assinar_ou_verificar(raiz) doc_xml, ctxt, noh_assinatura, assinador = self._antes_de_assinar_ou_verificar(raiz)
# Realiza a assinatura # Realiza a assinatura
@ -144,21 +139,21 @@ digital no XML informado."""
x509_certificate.text = certificados[len(certificados)-1].content.replace(u'\n', u'') x509_certificate.text = certificados[len(certificados)-1].content.replace(u'\n', u'')
resultado = assinador.status == xmlsec.DSigStatusSucceeded resultado = assinador.status == xmlsec.DSigStatusSucceeded
# Gera o XML para retornar # Gera o XML para retornar
xml = doc_xml.serialize() xml = doc_xml.serialize()
# Limpa objetos da memoria e desativa funções criptográficas # Limpa objetos da memoria e desativa funções criptográficas
self._depois_de_assinar_ou_verificar(doc_xml, ctxt, assinador) self._depois_de_assinar_ou_verificar(doc_xml, ctxt, assinador)
if retorna_xml: if retorna_xml:
return xml return xml
else: else:
return etree.parse(StringIO(xml)) return etree.parse(StringIO(xml))
def _ativar_funcoes_criptograficas(self): def _ativar_funcoes_criptograficas(self):
# FIXME: descobrir forma de evitar o uso do libxml2 neste processo # FIXME: descobrir forma de evitar o uso do libxml2 neste processo
# Ativa as funções de análise de arquivos XML FIXME # Ativa as funções de análise de arquivos XML FIXME
libxml2.initParser() libxml2.initParser()
libxml2.substituteEntitiesDefault(1) libxml2.substituteEntitiesDefault(1)
@ -170,8 +165,8 @@ digital no XML informado."""
def _desativar_funcoes_criptograficas(self): def _desativar_funcoes_criptograficas(self):
''' Desativa as funções criptográficas e de análise XML ''' Desativa as funções criptográficas e de análise XML
As funções devem ser chamadas aproximadamente na ordem inversa da ativação
'''
As funções devem ser chamadas aproximadamente na ordem inversa da ativação
'''
# Shutdown xmlsec-crypto library # Shutdown xmlsec-crypto library
xmlsec.cryptoShutdown() xmlsec.cryptoShutdown()
@ -184,35 +179,35 @@ As funções devem ser chamadas aproximadamente na ordem inversa da ativação
# Shutdown LibXML2 FIXME: descobrir forma de evitar o uso do libxml2 neste processo # Shutdown LibXML2 FIXME: descobrir forma de evitar o uso do libxml2 neste processo
libxml2.cleanupParser() libxml2.cleanupParser()
def verificar_arquivo(self, caminho_arquivo): def verificar_arquivo(self, caminho_arquivo):
# Carrega o XML do arquivo # Carrega o XML do arquivo
raiz = etree.parse(caminho_arquivo) raiz = etree.parse(caminho_arquivo)
return self.verificar_etree(raiz) return self.verificar_etree(raiz)
def verificar_xml(self, xml): def verificar_xml(self, xml):
raiz = etree.parse(StringIO(xml)) raiz = etree.parse(StringIO(xml))
return self.verificar_etree(raiz) return self.verificar_etree(raiz)
def verificar_etree(self, raiz): def verificar_etree(self, raiz):
doc_xml, ctxt, noh_assinatura, assinador = self._antes_de_assinar_ou_verificar(raiz) doc_xml, ctxt, noh_assinatura, assinador = self._antes_de_assinar_ou_verificar(raiz)
# Verifica a assinatura # Verifica a assinatura
assinador.verify(noh_assinatura) assinador.verify(noh_assinatura)
resultado = assinador.status == xmlsec.DSigStatusSucceeded resultado = assinador.status == xmlsec.DSigStatusSucceeded
# Limpa objetos da memoria e desativa funções criptográficas # Limpa objetos da memoria e desativa funções criptográficas
self._depois_de_assinar_ou_verificar(doc_xml, ctxt, assinador) self._depois_de_assinar_ou_verificar(doc_xml, ctxt, assinador)
return resultado return resultado
def _antes_de_assinar_ou_verificar(self, raiz): def _antes_de_assinar_ou_verificar(self, raiz):
# Converte etree para string # Converte etree para string
xml = etree.tostring(raiz, xml_declaration=True, encoding='utf-8') xml = etree.tostring(raiz, xml_declaration=True, encoding='utf-8')
# Ativa funções criptográficas # Ativa funções criptográficas
self._ativar_funcoes_criptograficas() self._ativar_funcoes_criptograficas()
# Colocamos o texto no avaliador XML FIXME: descobrir forma de evitar o uso do libxml2 neste processo # Colocamos o texto no avaliador XML FIXME: descobrir forma de evitar o uso do libxml2 neste processo
doc_xml = libxml2.parseMemory(xml, len(xml)) doc_xml = libxml2.parseMemory(xml, len(xml))
@ -237,15 +232,15 @@ As funções devem ser chamadas aproximadamente na ordem inversa da ativação
# Atribui a chave ao assinador # Atribui a chave ao assinador
assinador.signKey = chave assinador.signKey = chave
return doc_xml, ctxt, noh_assinatura, assinador return doc_xml, ctxt, noh_assinatura, assinador
def _depois_de_assinar_ou_verificar(self, doc_xml, ctxt, assinador): def _depois_de_assinar_ou_verificar(self, doc_xml, ctxt, assinador):
# Libera a memória do assinador; isso é necessário, pois na verdade foi feita uma chamada # Libera a memória do assinador; isso é necessário, pois na verdade foi feita uma chamada
# a uma função em C cujo código não é gerenciado pelo Python # a uma função em C cujo código não é gerenciado pelo Python
assinador.destroy() assinador.destroy()
ctxt.xpathFreeContext() ctxt.xpathFreeContext()
doc_xml.freeDoc() doc_xml.freeDoc()
# E, por fim, desativa todas as funções ativadas anteriormente # E, por fim, desativa todas as funções ativadas anteriormente
self._desativar_funcoes_criptograficas() self._desativar_funcoes_criptograficas()

86
pynfe/processamento/comunicacao.py

@ -1,27 +1,34 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
import datetime
from httplib import HTTPSConnection, HTTPResponse from httplib import HTTPSConnection, HTTPResponse
from pynfe.utils import etree, StringIO
from pynfe.utils import etree, StringIO, so_numeros
from pynfe.utils.flags import NAMESPACE_NFE, NAMESPACE_SOAP, VERSAO_PADRAO from pynfe.utils.flags import NAMESPACE_NFE, NAMESPACE_SOAP, VERSAO_PADRAO
from pynfe.utils.flags import CODIGOS_ESTADOS, VERSAO_PADRAO
from assinatura import AssinaturaA1
class Comunicacao(object): class Comunicacao(object):
u"""Classe abstrata responsavel por definir os metodos e logica das classes u"""Classe abstrata responsavel por definir os metodos e logica das classes
de comunicação com os webservices da NF-e.""" de comunicação com os webservices da NF-e."""
_ambiente = 1 # 1 = Produção, 2 = Homologação
servidor = None servidor = None
porta = 80
certificado = None certificado = None
certificado_senha = None certificado_senha = None
def __init__(self, servidor, certificado, certificado_senha):
def __init__(self, servidor, porta, certificado, certificado_senha, homologacao=False):
self.servidor = servidor self.servidor = servidor
self.porta = porta
self.certificado = certificado self.certificado = certificado
self.certificado_senha = certificado_senha self.certificado_senha = certificado_senha
self._ambiente = homologacao and 2 or 1
class ComunicacaoSefaz(Comunicacao): class ComunicacaoSefaz(Comunicacao):
u"""Classe de comunicação que segue o padrão definido para as SEFAZ dos Estados.""" u"""Classe de comunicação que segue o padrão definido para as SEFAZ dos Estados."""
_versao = VERSAO_PADRAO _versao = VERSAO_PADRAO
_assinatura = AssinaturaA1
def transmitir(self, nota_fiscal): def transmitir(self, nota_fiscal):
pass pass
@ -41,8 +48,8 @@ class ComunicacaoSefaz(Comunicacao):
# Monta XML para envio da requisição # Monta XML para envio da requisição
xml = self._construir_xml_soap( xml = self._construir_xml_soap(
metodo='CadConsultaCadastro',
tag_metodo='consultaCadastro',
metodo='nfeRecepcao2', # FIXME
tag_metodo='nfeStatusServicoNF2', # FIXME
cabecalho=self._cabecalho_soap(), cabecalho=self._cabecalho_soap(),
dados=dados, dados=dados,
) )
@ -51,18 +58,69 @@ class ComunicacaoSefaz(Comunicacao):
retorno = self._post(post, xml, self._post_header()) retorno = self._post(post, xml, self._post_header())
# Transforma o retorno em etree # Transforma o retorno em etree
try:
retorno = etree.parse(StringIO(retorno))
return retorno
except TypeError:
pass
#retorno = etree.parse(StringIO(retorno))
return bool(retorno)
def consultar_cadastro(self, instancia): def consultar_cadastro(self, instancia):
#post = '/nfeweb/services/cadconsultacadastro.asmx' #post = '/nfeweb/services/cadconsultacadastro.asmx'
post = '/nfeweb/services/nfeconsulta.asmx' post = '/nfeweb/services/nfeconsulta.asmx'
def inutilizar_faixa_numeracao(self, faixa):
pass
def inutilizar_faixa_numeracao(self, numero_inicial, numero_final, emitente, certificado,
senha, ano=None, serie='1', justificativa=''):
post = '/nfeweb/services/nfestatusservico.asmx'
# Valores default
ano = str(ano or datetime.date.today().year)[-2:]
uf = CODIGOS_ESTADOS[emitente.endereco_uf]
cnpj = so_numeros(emitente.cnpj)
# Identificador da TAG a ser assinada formada com Código da UF + Ano (2 posições) +
# CNPJ + modelo + série + nro inicial e nro final precedida do literal “ID”
id_unico = 'ID%(uf)s%(ano)s%(cnpj)s%(modelo)s%(serie)s%(num_ini)s%(num_fin)s'%{
'uf': uf,
'ano': ano,
'cnpj': cnpj,
'modelo': '55',
'serie': serie.zfill(3),
'num_ini': str(numero_inicial).zfill(9),
'num_fin': str(numero_final).zfill(9),
}
# Monta XML do corpo da requisição # FIXME
raiz = etree.Element('inutNFe', xmlns="http://www.portalfiscal.inf.br/nfe", versao="1.07")
inf_inut = etree.SubElement(raiz, 'infInut', Id=id_unico)
etree.SubElement(inf_inut, 'tpAmb').text = str(self._ambiente)
etree.SubElement(inf_inut, 'xServ').text = 'INUTILIZAR'
etree.SubElement(inf_inut, 'cUF').text = uf
etree.SubElement(inf_inut, 'ano').text = ano
etree.SubElement(inf_inut, 'CNPJ').text = emitente.cnpj
etree.SubElement(inf_inut, 'mod').text = '55'
etree.SubElement(inf_inut, 'serie').text = serie
etree.SubElement(inf_inut, 'nNFIni').text = str(numero_inicial)
etree.SubElement(inf_inut, 'nNFFin').text = str(numero_final)
etree.SubElement(inf_inut, 'xJust').text = justificativa
#dados = etree.tostring(raiz, encoding='utf-8', xml_declaration=True)
# Efetua assinatura
assinatura = self._assinatura(certificado, senha)
dados = assinatura.assinar_etree(etree.ElementTree(raiz), retorna_xml=True)
# Monta XML para envio da requisição
xml = self._construir_xml_soap(
metodo='nfeRecepcao2', # XXX
tag_metodo='nfeInutilizacaoNF', # XXX
cabecalho=self._cabecalho_soap(),
dados=dados,
)
# Chama método que efetua a requisição POST no servidor SOAP
retorno = self._post(post, xml, self._post_header())
# Transforma o retorno em etree # TODO
#retorno = etree.parse(StringIO(retorno))
return retorno
def _cabecalho_soap(self): def _cabecalho_soap(self):
u"""Monta o XML do cabeçalho da requisição SOAP""" u"""Monta o XML do cabeçalho da requisição SOAP"""
@ -99,11 +157,11 @@ class ComunicacaoSefaz(Comunicacao):
caminho_chave, caminho_cert = self.certificado.separar_arquivo(senha=self.certificado_senha) caminho_chave, caminho_cert = self.certificado.separar_arquivo(senha=self.certificado_senha)
# Abre a conexão HTTPS # Abre a conexão HTTPS
con = HTTPSConnection(self.servidor, key_file=caminho_chave, cert_file=caminho_cert)
con = HTTPSConnection(self.servidor, self.porta, key_file=caminho_chave, cert_file=caminho_cert)
try: try:
#con.set_debuglevel(100) #con.set_debuglevel(100)
con.request(u'POST', post, xml, header) con.request(u'POST', post, xml, header)
resp = con.getresponse() resp = con.getresponse()

4
pynfe/processamento/serializacao.py

@ -17,7 +17,7 @@ class Serializacao(object):
Nao deve ser instanciada diretamente!""" Nao deve ser instanciada diretamente!"""
_fonte_dados = None _fonte_dados = None
_ambiente = 1
_ambiente = 1 # 1 = Produção, 2 = Homologação
_nome_aplicacao = 'PyNFe' _nome_aplicacao = 'PyNFe'
def __new__(cls, *args, **kwargs): def __new__(cls, *args, **kwargs):
@ -232,7 +232,7 @@ class SerializacaoXML(Serializacao):
etree.SubElement(ide, 'indPag').text = str(nota_fiscal.forma_pagamento) etree.SubElement(ide, 'indPag').text = str(nota_fiscal.forma_pagamento)
etree.SubElement(ide, 'mod').text = str(nota_fiscal.modelo) etree.SubElement(ide, 'mod').text = str(nota_fiscal.modelo)
etree.SubElement(ide, 'serie').text = nota_fiscal.serie etree.SubElement(ide, 'serie').text = nota_fiscal.serie
etree.SubElement(ide, 'nNF').text = nota_fiscal.numero_nf
etree.SubElement(ide, 'nNF').text = str(nota_fiscal.numero_nf)
etree.SubElement(ide, 'dEmi').text = nota_fiscal.data_emissao.strftime('%Y-%m-%d') etree.SubElement(ide, 'dEmi').text = nota_fiscal.data_emissao.strftime('%Y-%m-%d')
etree.SubElement(ide, 'dSaiEnt').text = nota_fiscal.data_saida_entrada.strftime('%Y-%m-%d') etree.SubElement(ide, 'dSaiEnt').text = nota_fiscal.data_saida_entrada.strftime('%Y-%m-%d')
etree.SubElement(ide, 'tpNF').text = str(nota_fiscal.tipo_documento) etree.SubElement(ide, 'tpNF').text = str(nota_fiscal.tipo_documento)

6
pynfe/utils/__init__.py

@ -64,4 +64,8 @@ def obter_municipio_por_codigo(codigo, uf):
municipios = carregar_arquivo_municipios(uf) municipios = carregar_arquivo_municipios(uf)
return municipios[codigo] return municipios[codigo]
@memoize
def extrair_tag(root):
return root.tag.split('}')[-1]

103
run_fake_soap_server.py

@ -1,24 +1,95 @@
from soaplib.wsgi_soap import SimpleWSGISoapApp
from soaplib.service import soapmethod
from soaplib.serializers.primitive import String, Integer, Array, Null
# -*- coding: utf-8 -*-
class ServidorNFEFalso(SimpleWSGISoapApp):
@soapmethod(String, Integer, _returns=Array(String))
def ping(self, nome, vezes):
ret = [nome for i in range(vezes)]
print ret
return ret
"""Este script deve ser executado com Python 2.6+ e OpenSSL"""
import os, datetime
CUR_DIR = os.path.dirname(os.path.abspath(__file__))
#from soaplib.wsgi_soap import SimpleWSGISoapApp
#from soaplib.service import soapmethod
#from soaplib.serializers.primitive import String, Integer, Array, Null
#import tornado.wsgi
import tornado.httpserver
import tornado.ioloop
import tornado.web
import tornado.options
from pynfe.utils import etree, StringIO, extrair_tag
from pynfe.utils.flags import CODIGOS_ESTADOS
#class ServidorNFEFalso(SimpleWSGISoapApp):
# @soapmethod(String, Integer, _returns=Array(String))
# def ping(self, nome, vezes):
# ret = [nome for i in range(vezes)]
# return ret
class HandlerStatusServico(tornado.web.RequestHandler):
sigla_servidor = 'GO'
def post(self):
# Obtem o body da request
xml = self.request.body
# Transforma em etree
raiz = etree.parse(StringIO(xml))
# Extrai a tag do método da request
tag = extrair_tag(raiz.getroot().getchildren()[0].getchildren()[0])
# Chama o método respectivo para a tag
print 'Metodo:', tag
getattr(self, tag)(raiz)
def nfeStatusServicoNF2(self, raiz):
data_hora = datetime.datetime.now().strftime('%Y-%m-%dT%H:%M:%S')
ret = etree.Element('retConsStatServ')
etree.SubElement(ret, 'versao').text = '1.00' # FIXME
etree.SubElement(ret, 'tbAmb').text = '2' # Homologação
etree.SubElement(ret, 'verAplic').text = self.sigla_servidor
etree.SubElement(ret, 'cStat').text = '1' # FIXME
etree.SubElement(ret, 'xMotivo').text = 'Servico em funcionamento normal' # FIXME
etree.SubElement(ret, 'cUF').text = CODIGOS_ESTADOS[self.sigla_servidor]
etree.SubElement(ret, 'dhRecbto').text = data_hora
etree.SubElement(ret, 'tMed').text = '10'
etree.SubElement(ret, 'dhRetorno').text = data_hora
etree.SubElement(ret, 'xObs').text = 'Nenhuma informacao adicional'
xml = etree.tostring(ret, encoding='utf-8', xml_declaration=True)
self.write(xml)
def nfeInutilizacaoNF(self, raiz):
data_hora = datetime.datetime.now().strftime('%Y-%m-%dT%H:%M:%S')
ret = etree.Element('retInutNFe')
etree.SubElement(ret, 'versao').text = '1.00' # FIXME
xml_dados = raiz.getroot().getchildren()[0].getchildren()[0].getchildren()[1].text
xml = etree.tostring(ret, encoding='utf-8', xml_declaration=True)
self.write(xml)
if __name__ == '__main__': if __name__ == '__main__':
porta = 8080 porta = 8080
# Via Tornado
import tornado.wsgi
import tornado.httpserver
import tornado.ioloop
application = ServidorNFEFalso()
container = tornado.wsgi.WSGIContainer(application)
http_server = tornado.httpserver.HTTPServer(container)
# Codigo específico da soaplib
#application = ServidorNFEFalso()
#container = tornado.wsgi.WSGIContainer(application)
#http_server = tornado.httpserver.HTTPServer(container)
tornado.options.parse_command_line()
application = tornado.web.Application([
(r'^/nfeweb/services/nfestatusservico.asmx$', HandlerStatusServico), # Consulta de status do serviço
])
ssl_options = {
'certfile': os.path.join(CUR_DIR, 'tests', 'certificado.pem'),
'keyfile': os.path.join(CUR_DIR, 'tests', 'key.pem'),
}
http_server = tornado.httpserver.HTTPServer(application, ssl_options=ssl_options)
http_server.listen(porta) http_server.listen(porta)
tornado.ioloop.IOLoop.instance().start() tornado.ioloop.IOLoop.instance().start()

2
tests/03-processamento-01-serializacao-xml.txt

@ -63,7 +63,7 @@ Instancia a NF
... uf='GO', ... uf='GO',
... modelo=55, ... modelo=55,
... serie='1', ... serie='1',
... numero_nf='1',
... numero_nf=1,
... data_emissao=datetime.date(2010,1,13), ... data_emissao=datetime.date(2010,1,13),
... data_saida_entrada=datetime.date(2010,1,13), ... data_saida_entrada=datetime.date(2010,1,13),
... natureza_operacao='Venda a vista', ... natureza_operacao='Venda a vista',

58
tests/03-processamento-04-comunicacao.txt

@ -11,13 +11,67 @@ Carregando certificado digital tipo A1
Instancia de comunicacao Instancia de comunicacao
>>> comunicacao = ComunicacaoSefaz( >>> comunicacao = ComunicacaoSefaz(
... #servidor='localhost:8080',
... servidor='homologacao.nfe.fazenda.sp.gov.br',
... servidor='localhost',
... porta=8080,
... #servidor='homologacao.nfe.fazenda.sp.gov.br',
... certificado=certificado, ... certificado=certificado,
... certificado_senha='associacao', ... certificado_senha='associacao',
... ) ... )
Instancia do emitente (auxiliar para este teste)
>>> from pynfe.entidades import Emitente
>>> emitente = Emitente(
... cnpj='12.345.678/0001-90',
... razao_social='Tarsila Calcados Ltda.',
... nome_fantasia='Tarsila Calcados Ltda.',
... inscricao_estadual='123456789012',
... endereco_logradouro='Rua 10',
... endereco_numero='15',
... endereco_complemento='qd 17, lt 10',
... endereco_bairro='Setor Oeste',
... endereco_municipio='5208806', # Goiania
... endereco_uf='GO',
... endereco_cep='75370-000',
... endereco_telefone='6242421212',
... )
Instancia do certificado
>>> from pynfe.entidades import CertificadoA1
>>> certificado = CertificadoA1(caminho_arquivo='tests/certificado.pfx')
Verifica o status do servico Verifica o status do servico
>>> comunicacao.status_servico() >>> comunicacao.status_servico()
True
Transmissao de NF-e
>>> #comunicacao.transmitir(nota_fiscal)
Cancelamento de NF-e
>>> #comunicacao.cancelar(nota_fiscal)
Consulta situacao de NF-e
>>> #comunicacao.situacao_nfe(nota_fiscal)
Consulta de cadastro (???)
>>> #comunicacao.consultar_cadastro()
Inulitilizacao de faixa de numeracao
>>> comunicacao.inutilizar_faixa_numeracao(
... certificado=certificado,
... senha='associacao',
... numero_inicial=10,
... numero_final=20,
... emitente=emitente,
... ano=2009,
... serie='1',
... justificativa='AJUSTE DA SEQUENCIA DE NUMERACAO',
... )

27
tests/certificado.pem

@ -0,0 +1,27 @@
-----BEGIN CERTIFICATE-----
MIIEqzCCA5OgAwIBAgIDMTg4MA0GCSqGSIb3DQEBBQUAMIGSMQswCQYDVQQGEwJC
UjELMAkGA1UECBMCUlMxFTATBgNVBAcTDFBvcnRvIEFsZWdyZTEdMBsGA1UEChMU
VGVzdGUgUHJvamV0byBORmUgUlMxHTAbBgNVBAsTFFRlc3RlIFByb2pldG8gTkZl
IFJTMSEwHwYDVQQDExhORmUgLSBBQyBJbnRlcm1lZGlhcmlhIDEwHhcNMDkwNTIy
MTcwNzAzWhcNMTAxMDAyMTcwNzAzWjCBnjELMAkGA1UECBMCUlMxHTAbBgNVBAsT
FFRlc3RlIFByb2pldG8gTkZlIFJTMR0wGwYDVQQKExRUZXN0ZSBQcm9qZXRvIE5G
ZSBSUzEVMBMGA1UEBxMMUE9SVE8gQUxFR1JFMQswCQYDVQQGEwJCUjEtMCsGA1UE
AxMkTkZlIC0gQXNzb2NpYWNhbyBORi1lOjk5OTk5MDkwOTEwMjcwMIGfMA0GCSqG
SIb3DQEBAQUAA4GNADCBiQKBgQCx1O/e1Q+xh+wCoxa4pr/5aEFt2dEX9iBJyYu/
2a78emtorZKbWeyK435SRTbHxHSjqe1sWtIhXBaFa2dHiukT1WJyoAcXwB1GtxjT
2VVESQGtRiujMa+opus6dufJJl7RslAjqN/ZPxcBXaezt0nHvnUB/uB1K8WT9G7E
S0V17wIDAQABo4IBfjCCAXowIgYDVR0jAQEABBgwFoAUPT5TqhNWAm+ZpcVsvB7m
alDBjEQwDwYDVR0TAQH/BAUwAwEBADAPBgNVHQ8BAf8EBQMDAOAAMAwGA1UdIAEB
AAQCMAAwgawGA1UdEQEBAASBoTCBnqA4BgVgTAEDBKAvBC0yMjA4MTk3Nzk5OTk5
OTk5OTk5MDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDCgEgYFYEwBAwKgCQQHREZU
LU5GZaAZBgVgTAEDA6AQBA45OTk5OTA5MDkxMDI3MKAXBgVgTAEDB6AOBAwwMDAw
MDAwMDAwMDCBGmRmdC1uZmVAcHJvY2VyZ3MucnMuZ292LmJyMCAGA1UdJQEB/wQW
MBQGCCsGAQUFBwMCBggrBgEFBQcDBDBTBgNVHR8BAQAESTBHMEWgQ6BBhj9odHRw
Oi8vbmZlY2VydGlmaWNhZG8uc2VmYXoucnMuZ292LmJyL0xDUi9BQ0ludGVybWVk
aWFyaWEzOC5jcmwwDQYJKoZIhvcNAQEFBQADggEBAJFytXuiS02eJO0iMQr/Hi+O
x7/vYiPewiDL7s5EwO8A9jKx9G2Baz0KEjcdaeZk9a2NzDEgX9zboPxhw0RkWahV
CP2xvRFWswDIa2WRUT/LHTEuTeKCJ0iF/um/kYM8PmWxPsDWzvsCCRp146lc0lz9
LGm5ruPVYPZ/7DAoimUk3bdCMW/rzkVYg7iitxHrhklxH7YWQHUwbcqPt7Jv0RJx
clc1MhQlV2eM2MO1iIlk8Eti86dRrJVoicR1bwc6/YDqDp4PFONTi1ddewRu6elG
S74AzCcNYRSVTINYiZLpBZO0uivrnTEnsFguVnNtWb9MAHGt3tkR0gAVs6S0fm8=
-----END CERTIFICATE-----

15
tests/key.pem

@ -0,0 +1,15 @@
-----BEGIN RSA PRIVATE KEY-----
MIICXQIBAAKBgQCx1O/e1Q+xh+wCoxa4pr/5aEFt2dEX9iBJyYu/2a78emtorZKb
WeyK435SRTbHxHSjqe1sWtIhXBaFa2dHiukT1WJyoAcXwB1GtxjT2VVESQGtRiuj
Ma+opus6dufJJl7RslAjqN/ZPxcBXaezt0nHvnUB/uB1K8WT9G7ES0V17wIDAQAB
AoGAAYOcnzsVLR/JJKSa1uukis0WcYb/PsL7t2Ud6X5C/SdVrsh7jRMQ7oXNV4n7
U2wayiHyQY/sZhhQoMYvVO6b2Wm/anq+F5L2QNT0FUxckRugDtH6tY6wqbs+AoTK
+eTBcSowB6HP6J9yZss02jSzh1xqX1t7q7dJEJjTGmoRczECQQDRrXux9N5Bdnzu
vTQFm7W1cYkxLrxPZ/gHFaHLAjd/h29cG1UTpkmUeKFDDPETE4q6Zmg3sWOAeMZB
tZQA7nf1AkEA2R5hS2Z+vvXLEGzxSTwYH2pRBoXUzuj801YqMhe4T/pJu4H3Bzab
4/SEEZdcFEa51HdOmqtOtQj1NDy/z3Lb0wJBAI1YaGEvU8BHcrKxgucg715QGg64
laLl0HJeJ8IlTWo/z1cE6dYkK8fVhcggakbUzpkXPbwFbbEGOYfEMvBp0R0CQQCr
G98vriIbWthjJIhv3/Ve5Mngax6QxltiLpjoi3sNRMJRDRbiz23CFBT1TCUcMbUI
NdJz4KgR0nJ0bZ/43JtTAkBLJwkUNQDLdxx2EMif8KMRfsTAqUQ0d4wWnUP0hPHH
4qjUm/un000TF7eB85tszXxGxWATaGx4OPT91dXwzDsh
-----END RSA PRIVATE KEY-----
Loading…
Cancel
Save