diff --git a/sapl/materia/urls.py b/sapl/materia/urls.py index f4201141c..1dc8dda06 100644 --- a/sapl/materia/urls.py +++ b/sapl/materia/urls.py @@ -56,7 +56,8 @@ from sapl.materia.views_assinatura import ( detectar_aplicacao_a3, materia_assinar_lote, docacessorio_assinar_a1, docacessorio_pdf_assinado, docacessorio_verificar_assinatura, docacessorio_remover_assinatura, - materia_verificar_documento, docacessorio_verificar_documento + materia_verificar_documento, docacessorio_verificar_documento, + docacessorio_assinar_lote ) from sapl.norma.views import NormaPesquisaSimplesView from sapl.protocoloadm.views import ( @@ -237,6 +238,10 @@ urlpatterns_materia = [ url(r'^materia/documentoacessorio/(?P\d+)/remover-assinatura/$', docacessorio_remover_assinatura, name='docacessorio_remover_assinatura'), + # Assinatura em Lote de Documentos Acessórios + url(r'^materia/documentoacessorio/assinar-em-lote/$', docacessorio_assinar_lote, + name='docacessorio_assinar_lote'), + # Verificação pública de autenticidade de Documento Acessório (sem login) url(r'^materia/documentoacessorio/(?P\d+)/verificar/$', docacessorio_verificar_documento, name='docacessorio_verificar_documento'), diff --git a/sapl/materia/views.py b/sapl/materia/views.py index 4bc6156b9..cd7ada455 100644 --- a/sapl/materia/views.py +++ b/sapl/materia/views.py @@ -2019,6 +2019,29 @@ class DocumentoAcessorioCrud(MasterDetailCrud): u.has_perm('materia.add_documentoacessorio') ) context['tipos_documento'] = TipoDocumento.objects.all() + + # Documentos acessórios pendentes de assinatura para o lote + pode_assinar_lote = u.is_authenticated and ( + u.is_superuser or + u.has_perm('materia.change_documentoacessorio') + ) + if not pode_assinar_lote: + pode_assinar_lote = u.is_authenticated and OperadorAutor.objects.filter(user=u).exists() + + if pode_assinar_lote: + materia_pk = self.kwargs.get('pk') or self.kwargs.get('root_pk') + qs_pendentes = DocumentoAcessorio.objects.filter( + materia__pk=materia_pk, + pdf_assinado='', + ).order_by('data', 'nome') + docs_lote = [ + {'id': d.pk, 'descricao': f'{d.nome} ({d.tipo}) — {d.data}'} + for d in qs_pendentes + ] + context['docs_pendentes_lote'] = docs_lote + else: + context['docs_pendentes_lote'] = [] + return context def hook_arquivo(self, obj, default, url): diff --git a/sapl/materia/views_assinatura.py b/sapl/materia/views_assinatura.py index 0a743cf57..767019fd0 100644 --- a/sapl/materia/views_assinatura.py +++ b/sapl/materia/views_assinatura.py @@ -2037,3 +2037,477 @@ def materia_assinar_lote(request): 'erros': erro_count, 'resultados': resultados, }) + + +# ============================================================================= +# Assinatura em Lote de Documentos Acessórios +# ============================================================================= + +@login_required +@csrf_exempt +@require_http_methods(["POST"]) +def docacessorio_assinar_lote(request): + """ + Assina em lote documentos acessórios pendentes com certificado A1. + + POST multipart: + - certificado: arquivo .pfx / .p12 + - senha: senha do certificado + - ids: JSON array com os PKs dos documentos acessórios ex: "[1,2,3]" + + Retorna JSON: + { + "total": 3, "sucesso": 2, "erros": 1, + "resultados": [ + {"pk": 1, "success": true, "descricao": "Despacho - PL 1/2025"}, + {"pk": 2, "success": false, "descricao": "...", "error": "..."} + ] + } + """ + _tem_perm_django = request.user.has_perm('materia.change_documentoacessorio') + try: + _autor_lote = OperadorAutor.objects.get(user=request.user).autor + except OperadorAutor.DoesNotExist: + _autor_lote = None + + if not (_tem_perm_django or _autor_lote or request.user.is_superuser): + return JsonResponse( + {'success': False, 'error': 'Sem permissao para assinar documentos acessorios.'}, + status=403 + ) + + # -- IDs dos documentos -- + ids_raw = request.POST.get('ids', '') + ids_multi = request.POST.getlist('ids[]') + + if ids_multi: + pks = [int(i) for i in ids_multi if str(i).isdigit()] + elif ids_raw: + try: + parsed = json.loads(ids_raw) + pks = [int(i) for i in parsed if str(i).isdigit() or isinstance(i, int)] + except (json.JSONDecodeError, ValueError): + return JsonResponse( + {'success': False, 'error': 'Parametro "ids" invalido. Envie um array JSON.'}, + status=400 + ) + else: + return JsonResponse({'success': False, 'error': 'Nenhum documento selecionado.'}, status=400) + + if not pks: + return JsonResponse({'success': False, 'error': 'Lista de IDs vazia.'}, status=400) + + if len(pks) > 200: + return JsonResponse( + {'success': False, 'error': 'Limite maximo de 200 documentos por lote.'}, + status=400 + ) + + # -- Certificado -- + certificado_file = request.FILES.get('certificado') + senha = request.POST.get('senha', '') + + if not certificado_file: + return JsonResponse({'success': False, 'error': 'Certificado nao informado.'}, status=400) + if not senha: + return JsonResponse({'success': False, 'error': 'Senha do certificado nao informada.'}, status=400) + + cert_bytes = certificado_file.read() + + import tempfile as tmp_module + with tmp_module.NamedTemporaryFile(delete=False, suffix='.pfx') as tmp_cert: + tmp_cert.write(cert_bytes) + tmp_cert_path = tmp_cert.name + + try: + from pyhanko.sign import signers + signer = signers.SimpleSigner.load_pkcs12( + pfx_file=tmp_cert_path, + passphrase=senha.encode('utf-8') + ) + except Exception as cert_error: + logger.error(f"[lote-doc] Erro ao carregar certificado: {cert_error}") + err_msg = str(cert_error) + if 'password' in err_msg.lower() or 'mac' in err_msg.lower(): + detail = 'Senha incorreta ou arquivo invalido.' + elif 'decode' in err_msg.lower() or 'parse' in err_msg.lower(): + detail = 'Arquivo nao e um certificado valido (.pfx/.p12).' + else: + detail = f'Detalhes: {err_msg}' + return JsonResponse({'success': False, 'error': f'Erro ao carregar certificado: {detail}'}, status=400) + finally: + if os.path.exists(tmp_cert_path): + os.unlink(tmp_cert_path) + + cert_info = signer.signing_cert + error_response = _validar_certificado(cert_info) + if error_response: + data = json.loads(error_response.content) + return JsonResponse({'success': False, 'error': data.get('error', 'Certificado invalido.')}, status=400) + + # -- Assinatura por documento -- + docs = DocumentoAcessorio.objects.filter(pk__in=pks).select_related('materia', 'tipo', 'materia__tipo') + docs_map = {d.pk: d for d in docs} + + resultados = [] + sucesso_count = 0 + erro_count = 0 + + for pk in pks: + doc = docs_map.get(pk) + if not doc: + resultados.append({'pk': pk, 'success': False, 'descricao': f'ID {pk}', 'error': 'Documento nao encontrado.'}) + erro_count += 1 + continue + + descricao = f'{doc.nome} - {doc.materia.tipo.sigla} {doc.materia.numero}/{doc.materia.ano}' + + if doc.pdf_assinado: + resultados.append({'pk': pk, 'success': False, 'descricao': descricao, 'error': 'Ja possui PDF assinado. Ignorado.'}) + erro_count += 1 + continue + + assinaturas_existentes = _normalizar_assinatura_info(doc.assinatura_info) + if any(a.get('signed_by') == request.user.username for a in assinaturas_existentes): + resultados.append({'pk': pk, 'success': False, 'descricao': descricao, 'error': 'Voce ja assinou este documento.'}) + erro_count += 1 + continue + + pdf_bytes, error = _gerar_pdf_do_docacessorio(doc, request) + if error: + resultados.append({'pk': pk, 'success': False, 'descricao': descricao, 'error': error}) + erro_count += 1 + continue + + temp_stamped_path = None + try: + from pyhanko.pdf_utils.incremental_writer import IncrementalPdfFileWriter + from PyPDF4 import PdfFileReader, PdfFileWriter as PyPDF4Writer + + nome_assinante, cargo, tipo_cert = _obter_info_assinante(request, cert_info) + data_assinatura = timezone.localtime(timezone.now()) + data_formatada = data_assinatura.strftime('%d/%m/%Y %H:%M:%S') + data_simples = data_assinatura.strftime('%d/%m/%Y %H:%M') + + codigo = _gerar_codigo_autenticacao(pdf_bytes) + url_verificacao = _construir_url_verificacao(request, 'docacessorio', pk, codigo) + + nova_assinatura_info = { + 'nome_assinante': nome_assinante, + 'cargo': cargo, + 'data_assinatura': data_simples, + } + + original_pdf = PdfFileReader(io.BytesIO(pdf_bytes)) + last_page = original_pdf.getPage(original_pdf.getNumPages() - 1) + page_box = last_page.mediaBox + page_width = float(page_box.getWidth()) + page_height = float(page_box.getHeight()) + + auth_page_bytes = _gerar_pagina_autenticacao( + [nova_assinatura_info], codigo, url_verificacao, + page_width, page_height + ) + + auth_page_pdf = PdfFileReader(io.BytesIO(auth_page_bytes)) + output_pdf = PyPDF4Writer() + for page_num in range(original_pdf.getNumPages()): + output_pdf.addPage(original_pdf.getPage(page_num)) + output_pdf.addPage(auth_page_pdf.getPage(0)) + + with tempfile.NamedTemporaryFile(delete=False, suffix='.pdf') as tmp_f: + temp_stamped_path = tmp_f.name + output_pdf.write(tmp_f) + + with open(temp_stamped_path, 'rb') as f: + stamped_bytes = f.read() + + signed_buffer = io.BytesIO() + with io.BytesIO(stamped_bytes) as inf: + w = IncrementalPdfFileWriter(inf) + meta = signers.PdfSignatureMetadata( + field_name='AssinaturaDigital', + location='Camara Municipal', + reason='Documento assinado digitalmente nos termos da MP 2.200-2/2001', + name=nome_assinante + ) + signers.sign_pdf(w, meta, signer=signer, output=signed_buffer) + + signed_buffer.seek(0) + signed_pdf_content = signed_buffer.read() + + filename = f"docacessorio_{doc.pk}_assinado_{int(timezone.now().timestamp())}.pdf" + doc.pdf_assinado.save(filename, ContentFile(signed_pdf_content), save=False) + doc.codigo_autenticacao = codigo + + nova_assinatura_record = { + 'tipo_certificado': 'A1', + 'tipo_certificado_display': f'{tipo_cert} - A1', + 'subject': str(cert_info.subject), + 'issuer': str(cert_info.issuer), + 'serial': str(cert_info.serial_number), + 'valid_from': cert_info.not_valid_before.isoformat(), + 'valid_to': cert_info.not_valid_after.isoformat(), + 'signed_by': request.user.username, + 'nome_assinante': nome_assinante, + 'cargo': cargo, + 'data_assinatura': data_formatada, + 'validade_juridica': 'Assinatura Eletronica Qualificada' + } + assinaturas_existentes.append(nova_assinatura_record) + doc.assinatura_info = assinaturas_existentes + doc.assinado_em = timezone.now() + doc.assinado_por = request.user + doc.save() + + logger.info(f"[lote-doc] DocAcessorio {pk} assinado por {request.user.username}") + resultados.append({'pk': pk, 'success': True, 'descricao': descricao}) + sucesso_count += 1 + + except Exception as e: + logger.error(f"[lote-doc] Erro ao assinar docacessorio {pk}: {e}") + resultados.append({'pk': pk, 'success': False, 'descricao': descricao, 'error': str(e)}) + erro_count += 1 + finally: + if temp_stamped_path and os.path.exists(temp_stamped_path): + os.unlink(temp_stamped_path) + + return JsonResponse({ + 'success': True, + 'total': len(pks), + 'sucesso': sucesso_count, + 'erros': erro_count, + 'resultados': resultados, + }) + + +# ============================================================================= +# Assinatura em Lote de Documentos Acessorios +# ============================================================================= + +@login_required +@csrf_exempt +@require_http_methods(["POST"]) +def docacessorio_assinar_lote(request): + """ + Assina em lote documentos acessorios pendentes com certificado A1. + + POST multipart: + - certificado: arquivo .pfx / .p12 + - senha: senha do certificado + - ids: JSON array com os PKs dos documentos acessorios + + Retorna JSON com resultado por documento. + """ + _tem_perm_django = request.user.has_perm('materia.change_documentoacessorio') + try: + _autor_lote = OperadorAutor.objects.get(user=request.user).autor + except OperadorAutor.DoesNotExist: + _autor_lote = None + + if not (_tem_perm_django or _autor_lote or request.user.is_superuser): + return JsonResponse( + {'success': False, 'error': 'Sem permissao para assinar documentos acessorios.'}, + status=403 + ) + + ids_raw = request.POST.get('ids', '') + ids_multi = request.POST.getlist('ids[]') + + if ids_multi: + pks = [int(i) for i in ids_multi if str(i).isdigit()] + elif ids_raw: + try: + parsed = json.loads(ids_raw) + pks = [int(i) for i in parsed if str(i).isdigit() or isinstance(i, int)] + except (json.JSONDecodeError, ValueError): + return JsonResponse( + {'success': False, 'error': 'Parametro "ids" invalido. Envie um array JSON.'}, + status=400 + ) + else: + return JsonResponse({'success': False, 'error': 'Nenhum documento selecionado.'}, status=400) + + if not pks: + return JsonResponse({'success': False, 'error': 'Lista de IDs vazia.'}, status=400) + + if len(pks) > 200: + return JsonResponse( + {'success': False, 'error': 'Limite maximo de 200 documentos por lote.'}, + status=400 + ) + + certificado_file = request.FILES.get('certificado') + senha = request.POST.get('senha', '') + + if not certificado_file: + return JsonResponse({'success': False, 'error': 'Certificado nao informado.'}, status=400) + if not senha: + return JsonResponse({'success': False, 'error': 'Senha do certificado nao informada.'}, status=400) + + cert_bytes = certificado_file.read() + + import tempfile as tmp_module + with tmp_module.NamedTemporaryFile(delete=False, suffix='.pfx') as tmp_cert: + tmp_cert.write(cert_bytes) + tmp_cert_path = tmp_cert.name + + try: + from pyhanko.sign import signers + signer = signers.SimpleSigner.load_pkcs12( + pfx_file=tmp_cert_path, + passphrase=senha.encode('utf-8') + ) + except Exception as cert_error: + logger.error(f"[lote-doc] Erro ao carregar certificado: {cert_error}") + err_msg = str(cert_error) + if 'password' in err_msg.lower() or 'mac' in err_msg.lower(): + detail = 'Senha incorreta ou arquivo invalido.' + elif 'decode' in err_msg.lower() or 'parse' in err_msg.lower(): + detail = 'Arquivo nao e um certificado valido (.pfx/.p12).' + else: + detail = f'Detalhes: {err_msg}' + return JsonResponse({'success': False, 'error': f'Erro ao carregar certificado: {detail}'}, status=400) + finally: + if os.path.exists(tmp_cert_path): + os.unlink(tmp_cert_path) + + cert_info = signer.signing_cert + error_response = _validar_certificado(cert_info) + if error_response: + data = json.loads(error_response.content) + return JsonResponse({'success': False, 'error': data.get('error', 'Certificado invalido.')}, status=400) + + docs = DocumentoAcessorio.objects.filter(pk__in=pks).select_related('materia', 'tipo', 'materia__tipo') + docs_map = {d.pk: d for d in docs} + + resultados = [] + sucesso_count = 0 + erro_count = 0 + + for pk in pks: + doc = docs_map.get(pk) + if not doc: + resultados.append({'pk': pk, 'success': False, 'descricao': f'ID {pk}', 'error': 'Documento nao encontrado.'}) + erro_count += 1 + continue + + descricao = f'{doc.nome} - {doc.materia.tipo.sigla} {doc.materia.numero}/{doc.materia.ano}' + + if doc.pdf_assinado: + resultados.append({'pk': pk, 'success': False, 'descricao': descricao, 'error': 'Ja possui PDF assinado. Ignorado.'}) + erro_count += 1 + continue + + assinaturas_existentes = _normalizar_assinatura_info(doc.assinatura_info) + if any(a.get('signed_by') == request.user.username for a in assinaturas_existentes): + resultados.append({'pk': pk, 'success': False, 'descricao': descricao, 'error': 'Voce ja assinou este documento.'}) + erro_count += 1 + continue + + pdf_bytes, error = _gerar_pdf_do_docacessorio(doc, request) + if error: + resultados.append({'pk': pk, 'success': False, 'descricao': descricao, 'error': error}) + erro_count += 1 + continue + + temp_stamped_path = None + try: + from pyhanko.pdf_utils.incremental_writer import IncrementalPdfFileWriter + from PyPDF4 import PdfFileReader, PdfFileWriter as PyPDF4Writer + + nome_assinante, cargo, tipo_cert = _obter_info_assinante(request, cert_info) + data_assinatura = timezone.localtime(timezone.now()) + data_formatada = data_assinatura.strftime('%d/%m/%Y %H:%M:%S') + data_simples = data_assinatura.strftime('%d/%m/%Y %H:%M') + + codigo = _gerar_codigo_autenticacao(pdf_bytes) + url_verificacao = _construir_url_verificacao(request, 'docacessorio', pk, codigo) + + nova_assinatura_info = { + 'nome_assinante': nome_assinante, + 'cargo': cargo, + 'data_assinatura': data_simples, + } + + original_pdf = PdfFileReader(io.BytesIO(pdf_bytes)) + last_page = original_pdf.getPage(original_pdf.getNumPages() - 1) + page_box = last_page.mediaBox + page_width = float(page_box.getWidth()) + page_height = float(page_box.getHeight()) + + auth_page_bytes = _gerar_pagina_autenticacao( + [nova_assinatura_info], codigo, url_verificacao, + page_width, page_height + ) + + auth_page_pdf = PdfFileReader(io.BytesIO(auth_page_bytes)) + output_pdf = PyPDF4Writer() + for page_num in range(original_pdf.getNumPages()): + output_pdf.addPage(original_pdf.getPage(page_num)) + output_pdf.addPage(auth_page_pdf.getPage(0)) + + with tempfile.NamedTemporaryFile(delete=False, suffix='.pdf') as tmp_f: + temp_stamped_path = tmp_f.name + output_pdf.write(tmp_f) + + with open(temp_stamped_path, 'rb') as f: + stamped_bytes = f.read() + + signed_buffer = io.BytesIO() + with io.BytesIO(stamped_bytes) as inf: + w = IncrementalPdfFileWriter(inf) + meta = signers.PdfSignatureMetadata( + field_name='AssinaturaDigital', + location='Camara Municipal', + reason='Documento assinado digitalmente nos termos da MP 2.200-2/2001', + name=nome_assinante + ) + signers.sign_pdf(w, meta, signer=signer, output=signed_buffer) + + signed_buffer.seek(0) + signed_pdf_content = signed_buffer.read() + + filename = f"docacessorio_{doc.pk}_assinado_{int(timezone.now().timestamp())}.pdf" + doc.pdf_assinado.save(filename, ContentFile(signed_pdf_content), save=False) + doc.codigo_autenticacao = codigo + + nova_assinatura_record = { + 'tipo_certificado': 'A1', + 'tipo_certificado_display': f'{tipo_cert} - A1', + 'subject': str(cert_info.subject), + 'issuer': str(cert_info.issuer), + 'serial': str(cert_info.serial_number), + 'valid_from': cert_info.not_valid_before.isoformat(), + 'valid_to': cert_info.not_valid_after.isoformat(), + 'signed_by': request.user.username, + 'nome_assinante': nome_assinante, + 'cargo': cargo, + 'data_assinatura': data_formatada, + 'validade_juridica': 'Assinatura Eletronica Qualificada' + } + assinaturas_existentes.append(nova_assinatura_record) + doc.assinatura_info = assinaturas_existentes + doc.assinado_em = timezone.now() + doc.assinado_por = request.user + doc.save() + + logger.info(f"[lote-doc] DocAcessorio {pk} assinado por {request.user.username}") + resultados.append({'pk': pk, 'success': True, 'descricao': descricao}) + sucesso_count += 1 + + except Exception as e: + logger.error(f"[lote-doc] Erro ao assinar docacessorio {pk}: {e}") + resultados.append({'pk': pk, 'success': False, 'descricao': descricao, 'error': str(e)}) + erro_count += 1 + finally: + if temp_stamped_path and os.path.exists(temp_stamped_path): + os.unlink(temp_stamped_path) + + return JsonResponse({ + 'success': True, + 'total': len(pks), + 'sucesso': sucesso_count, + 'erros': erro_count, + 'resultados': resultados, + }) diff --git a/sapl/templates/materia/documentoacessorio_list.html b/sapl/templates/materia/documentoacessorio_list.html index 04766b043..4ef928923 100644 --- a/sapl/templates/materia/documentoacessorio_list.html +++ b/sapl/templates/materia/documentoacessorio_list.html @@ -58,6 +58,17 @@ + {% if docs_pendentes_lote %} +
+ +
+ {% endif %} {% if pode_upload %} @@ -169,4 +180,387 @@ })(); {% endif %} + +{% if docs_pendentes_lote %} +{# ── Modal de Assinatura em Lote – Documentos Acessórios ─────────────── #} + + + +{% endif %} {% endblock %}