Merge 43bbc0e293 into 6eec69795c

pytrustnfe/certificado.py

@@ -4,12 +4,25 @@
 
 import tempfile
 from OpenSSL import crypto
+from datetime import datetime
 
 
 class Certificado(object):
     def __init__(self, pfx, password):
         self.pfx = pfx
         self.password = password
+        pfx = crypto.load_pkcs12(pfx, password)
+
+        cert = pfx.get_certificate()
+        cert_date =  int(str(cert.get_notAfter(),'UTF-8').strip('Z'))
+        sha1_fingerprint = cert.digest("sha1")
+        now  = datetime.now()
+        date = int(now.strftime("%Y%m%d%H%M%S"))
+        '''
+        Exceto certificado de testes
+        '''
+        if cert_date < date or str(sha1_fingerprint,'UTF-8') ==  "DE:08:15:1E:DA:12:B3:5F:76:BF:5D:4E:56:C1:14:12:8A:85:B6:47":
+            print("WARNING: Certificado expirado")
 
     def save_pfx(self):
         pfx_temp = tempfile.mkstemp()[1]
@@ -20,7 +33,10 @@ class Certificado(object):
 
 
 def extract_cert_and_key_from_pfx(pfx, password):
-    pfx = crypto.load_pkcs12(pfx, password)
+    try:
+        pfx = crypto.load_pkcs12(pfx, password)
+    except:
+        print("WARING: Falha ao ler certiticado. Verifique a senha")
     # PEM formatted private key
     key = crypto.dump_privatekey(crypto.FILETYPE_PEM, pfx.get_privatekey())
     # PEM formatted certificate

pytrustnfe/client.py

@@ -12,7 +12,13 @@ def get_authenticated_client(base_url, cert, key):
     cache = suds.cache.DocumentCache(location=cache_location)
 
     session = requests.Session()
-    session.cert = (cert, key)
+    session.cert = (cert, key)  
+
+    # Testa sessao https
+    r = requests.get(base_url, cert=(cert, key))
+    if r.status_code == 403:
+        print("ERROR: Falha na conexão utilizando o certificado digital e senha infomados. Verifique a validade do certificado")
+        exit()
     return suds.client.Client(
         base_url, cache=cache, transport=suds_requests.RequestsTransport(session)
     )

tests/test_certificado.py

@@ -60,7 +60,7 @@ class test_assinatura(unittest.TestCase):
 
     def test_save_pfx(self):
         pfx_source = open(os.path.join(self.caminho, "teste.pfx"), "rb").read()
-        pfx = Certificado(pfx_source, "123")
+        pfx = Certificado(pfx_source, "123456")
         path = pfx.save_pfx()
         saved = open(path, "rb").read()
         self.assertEqual(