impl: captura de assinaturas eletrônicas em matérias

requirements/requirements.txt

@@ -34,5 +34,7 @@ whitenoise==5.1.0
 kazoo==2.8.0
 django-prometheus==2.2.0
 
+asn1crypto==1.5.1
+
 git+https://github.com/interlegis/trml2pdf
 git+https://github.com/interlegis/django-admin-bootstrapped

sapl/audiencia/models.py

@@ -49,6 +49,7 @@ class TipoAudienciaPublica(models.Model):
 
 
 class AudienciaPublica(models.Model):
+
     materia = models.ForeignKey(
         MateriaLegislativa,
         on_delete=models.PROTECT,

sapl/base/receivers.py

@@ -1,23 +1,30 @@
+from datetime import datetime
 import inspect
 import logging
 
+from PyPDF4.pdf import PdfFileReader
+from asn1crypto import cms
 from django.conf import settings
 from django.contrib.contenttypes.models import ContentType
 from django.core import serializers
+from django.core.files.uploadedfile import InMemoryUploadedFile, UploadedFile
+from django.db.models.fields.files import FileField
 from django.db.models.signals import post_delete, post_save, \
-    post_migrate
+    post_migrate, pre_save
 from django.db.utils import DEFAULT_DB_ALIAS
 from django.dispatch import receiver
 from django.utils import timezone
 from django.utils.translation import ugettext_lazy as _
 
 from sapl.base.email_utils import do_envia_email_tramitacao
-from sapl.base.models import AuditLog, TipoAutor, Autor
+from sapl.base.models import AuditLog, TipoAutor, Autor, Metadata
 from sapl.decorators import receiver_multi_senders
 from sapl.materia.models import Tramitacao
+from sapl.parlamentares.models import Parlamentar
 from sapl.protocoloadm.models import TramitacaoAdministrativo
 from sapl.utils import get_base_url, models_with_gr_for_model
 
+
 models_with_gr_for_autor = models_with_gr_for_model(Autor)
 
 
@@ -192,3 +199,270 @@ def cria_models_tipo_autor(app_config=None, verbosity=2, interactive=True,
 
 
 post_migrate.connect(receiver=cria_models_tipo_autor)
+
+
+def signed_files_extraction_function(sender, instance, **kwargs):
+
+    def run_signed_name_and_date_via_fields(fields):
+        signs = []
+
+        for key, field in fields.items():
+
+            if '/FT' not in field and field['/FT'] != '/Sig':
+                continue
+            if '/V' not in field:
+                continue
+
+            content_sign = field['/V']['/Contents']
+            nome = 'Nome do assinante não localizado.'
+            oname = ''
+            try:
+                info = cms.ContentInfo.load(content_sign)
+                signed_data = info['content']
+                oun_old = []
+                for cert in signed_data['certificates']:
+                    subject = cert.native['tbs_certificate']['subject']
+                    issuer = cert.native['tbs_certificate']['issuer']
+                    oname = issuer.get('organization_name', '')
+
+                    if oname == 'Gov-Br':
+                        nome = subject['common_name'].split(':')[0]
+                        continue
+
+                    oun = subject['organizational_unit_name']
+
+                    if isinstance(oun, str):
+                        continue
+
+                    if len(oun) > len(oun_old):
+                        oun_old = oun
+                        nome = subject['common_name'].split(':')[0]
+
+                    if oun and isinstance(oun, list) and len(oun) == 4:
+                        oname += ' - ' + oun[3]
+                        break
+
+            except:
+                if '/Name' in field['/V']:
+                    nome = field['/V']['/Name']
+
+            fd = None
+            try:
+                data = str(field['/V']['/M'])
+
+                if 'D:' not in data:
+                    data = None
+                else:
+                    if not data.endswith('Z'):
+                        data = data.replace('Z', '+')
+                    data = data.replace("'", '')
+
+                    fd = datetime.strptime(data[2:], '%Y%m%d%H%M%S%z')
+            except:
+                pass
+
+            signs.append((nome, [fd, oname]))
+
+        return signs
+
+    def run_signed_name_and_date_extract(file):
+        signs = {}
+        fields = {}
+        pdfdata = file.read()
+
+        # se não tem byterange então não é assinado
+        byterange = []
+        n = -1
+        while True:
+            n = pdfdata.find(b"/ByteRange", n + 1)
+            if n == -1:
+                break
+            byterange.append(n)
+
+        if not byterange:
+            return signs
+
+        # tenta extrair via /Fields
+        try:
+            pdf = PdfFileReader(file)
+            fields = pdf.getFields()
+        except Exception as e:
+            try:
+                pdf = PdfFileReader(file, strict=False)
+                fields = pdf.getFields()
+            except Exception as ee:
+                fields = ee
+
+        try:
+            # se a extração via /Fields ocorrer sem erros e forem capturadas
+            # tantas assinaturas quanto byteranges
+            if isinstance(fields, dict):
+                signs = run_signed_name_and_date_via_fields(fields)
+                if len(signs) == len(byterange):
+                    return signs
+
+            for n in byterange:
+
+                start = pdfdata.find(b"[", n)
+                stop = pdfdata.find(b"]", start)
+                assert n != -1 and start != -1 and stop != -1
+                n += 1
+
+                br = [int(i, 10) for i in pdfdata[start + 1: stop].split()]
+                contents = pdfdata[br[0] + br[1] + 1: br[2] - 1]
+                bcontents = bytes.fromhex(contents.decode("utf8"))
+                data1 = pdfdata[br[0]: br[0] + br[1]]
+                data2 = pdfdata[br[2]: br[2] + br[3]]
+                #signedData = data1 + data2
+
+                nome = 'Nome do assinante não localizado.'
+                oname = ''
+                try:
+                    info = cms.ContentInfo.load(bcontents)
+                    signed_data = info['content']
+
+                    oun_old = []
+                    for cert in signed_data['certificates']:
+                        subject = cert.native['tbs_certificate']['subject']
+                        issuer = cert.native['tbs_certificate']['issuer']
+                        oname = issuer.get('organization_name', '')
+
+                        if oname == 'Gov-Br':
+                            nome = subject['common_name'].split(':')[0]
+                            continue
+
+                        oun = subject['organizational_unit_name']
+
+                        if isinstance(oun, str):
+                            continue
+
+                        if len(oun) > len(oun_old):
+                            oun_old = oun
+                            nome = subject['common_name'].split(':')[0]
+
+                        if oun and isinstance(oun, list) and len(oun) == 4:
+                            oname += ' - ' + oun[3]
+                            break
+
+                except Exception as e:
+                    pass
+
+                fd = None
+                signs.append((nome, [fd, oname]))
+
+        except Exception as e:
+            pass
+
+        return signs
+
+    def signed_name_and_date_extract(file):
+
+        try:
+            signs = run_signed_name_and_date_extract(file)
+        except:
+            return {}
+
+        signs = sorted(signs, key=lambda sign: (
+            sign[0], sign[1][1], sign[1][0]))
+
+        signs_dict = {}
+
+        for s in signs:
+            # or 'ICP' in s[1][1] and 'ICP' not in signs_dict[s[0]][1]:
+            if s[0] not in signs_dict:
+                signs_dict[s[0]] = s[1]
+
+        signs = sorted(signs_dict.items(), key=lambda sign: (
+            sign[0], sign[1][1], sign[1][0]))
+
+        sr = []
+
+        for s in signs:
+            tt = s[0].title().split(' ')
+            for idx, t in enumerate(tt):
+                if t in ('Dos', 'De', 'Da', 'Do', 'Das', 'E'):
+                    tt[idx] = t.lower()
+            sr.append((' '.join(tt), s[1]))
+
+        signs = sr
+
+        meta_signs = {
+            'autores': [],
+            'admin': []
+        }
+
+        for s in signs:
+            # cn = # settings.CERT_PRIVATE_KEY_NAME
+            #meta_signs['admin' if s[0] == cn else 'autores'].append(s)
+            meta_signs['autores'].append(s)
+        return meta_signs
+
+    def filefield_from_model(m):
+        fields = m._meta.get_fields()
+        fields = tuple(map(lambda f: f.name, filter(
+            lambda x: isinstance(x, FileField), fields)))
+        return fields
+
+    FIELDFILE_NAME = filefield_from_model(instance)
+
+    if not FIELDFILE_NAME:
+        return
+
+    try:
+        md = Metadata.objects.get(
+            content_type=ContentType.objects.get_for_model(
+                instance._meta.model),
+            object_id=instance.id,).metadata
+    except:
+        md = {}
+
+    for fn in FIELDFILE_NAME:  # fn -> field_name
+        ff = getattr(instance, fn)  # ff -> file_field
+
+        if md and 'signs' in md and \
+                fn in md['signs'] and\
+                md['signs'][fn]:
+            md['signs'][fn] = {}
+
+        if not ff:
+            continue
+
+        try:
+            file = ff.file.file
+            meta_signs = {}
+            if not isinstance(ff.file, UploadedFile):
+                absolute_path = ff.path
+                with open(absolute_path, "rb") as file:
+                    meta_signs = signed_name_and_date_extract(file)
+                    file.close()
+            else:
+                file.seek(0)
+                meta_signs = signed_name_and_date_extract(file)
+
+            if not meta_signs or not meta_signs['autores'] and not meta_signs['admin']:
+                continue
+
+            if not md:
+                md = {'signs': {}}
+
+            if 'signs' not in md:
+                md['signs'] = {}
+
+            md['signs'][fn] = meta_signs
+        except Exception as e:
+            # print(e)
+            pass
+
+    if md:
+        metadata = Metadata.objects.get_or_create(
+            content_type=ContentType.objects.get_for_model(
+                instance._meta.model),
+            object_id=instance.id,)
+        metadata[0].metadata = md
+        metadata[0].save()
+
+
+@receiver(pre_save, dispatch_uid='signed_files_extraction_pre_save_signal')
+def signed_files_extraction_pre_save_signal(sender, instance, **kwargs):
+
+    signed_files_extraction_function(sender, instance, **kwargs)

sapl/base/templatetags/common_tags.py

@@ -2,6 +2,7 @@ import re
 
 from django import template
 from django.template.defaultfilters import stringfilter
+from django.utils.dateparse import parse_datetime as django_parse_datetime
 from django.utils.safestring import mark_safe
 from webpack_loader import utils
 
@@ -12,6 +13,7 @@ from sapl.parlamentares.models import Filiacao
 from sapl.sessao.models import SessaoPlenaria
 from sapl.utils import filiacao_data, SEPARADOR_HASH_PROPOSICAO
 
+
 register = template.Library()
 
 
@@ -88,6 +90,7 @@ def format_user(user):
     else:
         return user.username
 
+
 @register.filter
 def meta_model_value(instance, attr):
     try:
@@ -398,3 +401,7 @@ def dont_break_out(value):
     _safe = mark_safe(_safe)
     return _safe
 
+
+@register.filter(expects_localtime=True)
+def parse_datetime(value):
+    return django_parse_datetime(value)

sapl/crispy_layout_mixin.py

@@ -4,6 +4,7 @@ from crispy_forms.bootstrap import FormActions
 from crispy_forms.helper import FormHelper
 from crispy_forms.layout import HTML, Div, Fieldset, Layout, Submit
 from django import template
+from django.contrib.contenttypes.models import ContentType
 from django.urls import reverse, reverse_lazy
 from django.utils import formats
 from django.utils.encoding import force_text
@@ -329,6 +330,21 @@ class CrispyLayoutFormMixin:
 
         return verbose_name, display
 
+    def widget__signs(self, obj, fieldname):
+        from sapl.base.models import Metadata
+        try:
+            md = Metadata.objects.get(
+                content_type=ContentType.objects.get_for_model(
+                    obj._meta.model),
+                object_id=obj.id,)
+            autores = md.metadata['signs'][fieldname]['autores']
+            t = template.loader.get_template('base/widget__signs.html')
+            rendered = str(t.render(context={'signs': autores}))
+        except Exception as e:
+            return '', ''
+
+        return 'Assinaturas Eletrônicas', rendered
+
     @property
     def layout_display(self):
 

sapl/materia/models.py

@@ -745,6 +745,7 @@ class Parecer(models.Model):
 
 
 class Proposicao(models.Model):
+
     autor = models.ForeignKey(
         Autor,
         null=True,

sapl/templates/base/widget__signs.html

@@ -0,0 +1,23 @@
+{% load common_tags%}
+
+{% for nome, data in signs %}
+  {% if forloop.first %}
+    <div class="col d-flex align-items-center">
+      <div class="box-assinatura-eletronica">
+          <ul class="sigs px-0">
+  {% endif %}
+          <li class="sig">
+            <span class="sig-nome">
+              {{nome}}
+            </span>
+            <small class="sig-data">
+                (Assinado em: {{data.0|parse_datetime|date:"DATETIME_FORMAT" }} -
+                <small>{{data.1}}</small>)
+            </small>
+          </li>
+  {% if forloop.last %}
+        </ul>
+      </div>
+    </div>
+  {% endif %}
+{% endfor %}

sapl/templates/materia/layouts.yaml

@@ -144,6 +144,7 @@ MateriaLegislativaDetail:
   - tipo ano numero
   - data_apresentacao  numero_protocolo  tipo_apresentacao
   - texto_original
+  - texto_original|widget__signs
   - numeracao_set
   - materia_anexada_set__materia_principal|m2m_urlize_for_detail
   - materia_principal_set__materia_anexada|m2m_urlize_for_detail