Browse Source

Restringe acesso a arquivos de mídia (#3865)

* Verifica permissão de acesso em documentos administrativos

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Serve arquivos de mídia via Django com X-Accel-Redirect

Todo /media/ passa pelo Django, que verifica a permissão dos arquivos de
documentos administrativos e proposições e define Content-Type e
Content-Disposition. O nginx entrega o arquivo por uma location interna.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Recusa upload de arquivos com conteúdo executável pelo navegador

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Oculta sessões não iniciadas também na API e nos arquivos de mídia

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Valida arquivo antes de gravar no cadastro de acessórios em lote

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Identifica conteúdo ativo também pelo tipo MIME da extensão

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Não redireciona arquivos de mídia para troca de senha

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Ajusta regras de acesso a arquivos de documentos e proposições

Documentos administrativos e acessórios ficam ocultos ao público quando a
configuração é restritiva ou o documento é restrito. O texto original da
proposição nunca é público. Arquivos de sessões plenárias são públicos.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Aplica regras da página de detalhe e de sessões ocultas aos arquivos

No modo restritivo, os arquivos de documentos administrativos exigem a
mesma permissão da página de detalhe. Arquivos de sessões ocultas e de
seus oradores e justificativas não são servidos ao público.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Usa permissões por estado no acesso ao texto original da proposição

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Potential fix for pull request finding 'CodeQL / Uncontrolled data used in path expression'

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>

* Potential fix for pull request finding 'CodeQL / Uncontrolled data used in path expression'

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>

* Centraliza validação de caminho dos arquivos de mídia

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Ajusta teste de sessão oculta à regra de data da sessão

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Update sapl/materia/views.py

Co-authored-by: joaohortsenado <100957576+joaohortsenado@users.noreply.github.com>

* Update sapl/base/media.py

Co-authored-by: joaohortsenado <100957576+joaohortsenado@users.noreply.github.com>

* Update sapl/base/media.py

Co-authored-by: joaohortsenado <100957576+joaohortsenado@users.noreply.github.com>

* Exige permissão de proposição para receber e confirmar proposições

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Corrige redirecionamento do PDF de acessórios de documento administrativo

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Usa a mesma configuração na página e nos arquivos de documento administrativo

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Remove XML da lista de tipos de texto aceitos no upload

O upload de XML já era recusado pela checagem de conteúdo ativo.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Envia ao Django só os arquivos de mídia que dependem de autorização

Os demais arquivos de /media/ voltam a ser servidos direto pelo nginx,
com nosniff e download forçado para extensões ativas.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Detecta extensões ativas sem depender de /etc/mime.types

A imagem python:3.12-slim não tem /etc/mime.types, então mimetypes não
reconhecia .xht, .xslt, .kml etc. e o upload/entrega deixavam passar.
EXTENSOES_ATIVAS passa a cobrir as mesmas extensões do nginx (mais .kml e
.xspf), com teste que mantém as duas listas iguais. media-types entra na
imagem para o tipo_ativo reconhecer os demais +xml.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Serve arquivos públicos só com tipos conhecidos

Em vez de depender do mime.types global, que entrega tipos +xml (.kml,
.xspf) inline, a location /media/ lista os tipos que podem abrir no
navegador e manda o resto como application/octet-stream.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: Edward <9326037+edwardoliveira@users.noreply.github.com>
Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>
Co-authored-by: joaohortsenado <100957576+joaohortsenado@users.noreply.github.com>
pull/3867/head
Edward 6 days ago
committed by GitHub
parent
commit
9c94aa398d
No known key found for this signature in database GPG Key ID: B5690EEEBB952194
  1. 2
      docker/Dockerfile
  2. 62
      docker/config/nginx/sapl.conf
  3. 28
      sapl/api/views_sessao.py
  4. 124
      sapl/base/media.py
  5. 384
      sapl/base/tests/test_media.py
  6. 19
      sapl/materia/models.py
  7. 64
      sapl/materia/tests/test_acessorio_em_lote.py
  8. 98
      sapl/materia/tests/test_proposicao_confirmar.py
  9. 72
      sapl/materia/views.py
  10. 4
      sapl/middleware/check_password.py
  11. 10
      sapl/protocoloadm/models.py
  12. 163
      sapl/protocoloadm/tests/test_docadm_acesso.py
  13. 32
      sapl/protocoloadm/views.py
  14. 94
      sapl/sessao/tests/test_sessao_oculta_api.py
  15. 94
      sapl/test_utils.py
  16. 10
      sapl/urls.py
  17. 44
      sapl/utils.py

2
docker/Dockerfile

@ -51,7 +51,7 @@ ENV LANG=C.UTF-8 LC_ALL=C.UTF-8 PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 \
RUN set -eux; \
apt-get update; \
apt-get install -y --no-install-recommends \
curl jq bash tzdata fontconfig tini libmagic1 \
curl jq bash tzdata fontconfig tini libmagic1 media-types \
libcairo2 libpango-1.0-0 libpangocairo-1.0-0 libgdk-pixbuf-2.0-0 \
libharfbuzz0b libfreetype6 libjpeg62-turbo zlib1g fonts-dejavu-core; \
if [ "$WITH_GRAPHVIZ" = "1" ]; then apt-get install -y --no-install-recommends graphviz; fi; \

62
docker/config/nginx/sapl.conf

@ -49,8 +49,70 @@ server {
alias /var/interlegis/sapl/collected_static/;
}
# Arquivos que dependem de autorização vão ao Django (sapl.base.media),
# que devolve X-Accel-Redirect para /media-interna/. Manter a lista em
# sincronia com PROTEGIDO_RE e MODELOS_DE_SESSAO (teste em
# sapl/base/tests/test_media.py). Os dois regex ficam no nível do server
# e antes de location /media/: um regex aninhado em /media/ teria
# precedência sobre este.
location ~* ^/media/sapl/(private/|[^/]+/(documentoadministrativo|documentoacessorioadministrativo|proposicao|sessaoplenaria|orador|oradorexpediente|oradorordemdia|justificativaausencia)/) {
proxy_set_header X-Request-ID $req_id;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_set_header Host $http_host;
proxy_redirect off;
proxy_pass http://sapl_server;
}
# Demais arquivos são públicos e servidos direto do disco; conteúdo que o
# navegador executaria como página é forçado a download.
location ~* ^/media/.+\.(html?|xhtml|xht|shtml|svgz?|xml|xsl|xslt|rdf|js|mjs|mht|mhtml|rss|atom|mml|kml|xspf)$ {
root /var/interlegis/sapl;
types { }
default_type application/octet-stream;
add_header Content-Disposition attachment always;
add_header X-Content-Type-Options nosniff always;
}
# Só os tipos abaixo saem com Content-Type próprio; o resto (inclusive
# +xml que o mime.types global entregaria inline) vai como octet-stream.
location /media/ {
root /var/interlegis/sapl;
types {
application/pdf pdf;
image/jpeg jpg jpeg jpe;
image/png png;
image/gif gif;
image/webp webp;
image/bmp bmp;
image/tiff tif tiff;
image/x-icon ico;
text/plain txt;
text/csv csv;
audio/mpeg mp3;
audio/ogg ogg;
video/mp4 mp4;
video/webm webm;
application/rtf rtf;
application/zip zip;
application/msword doc;
application/vnd.ms-excel xls;
application/vnd.ms-powerpoint ppt;
application/vnd.openxmlformats-officedocument.wordprocessingml.document docx;
application/vnd.openxmlformats-officedocument.spreadsheetml.sheet xlsx;
application/vnd.openxmlformats-officedocument.presentationml.presentation pptx;
application/vnd.oasis.opendocument.text odt;
application/vnd.oasis.opendocument.spreadsheet ods;
application/vnd.oasis.opendocument.presentation odp;
}
default_type application/octet-stream;
add_header X-Content-Type-Options nosniff always;
}
location /media-interna/ {
internal;
alias /var/interlegis/sapl/media/;
add_header X-Content-Type-Options nosniff always;
}
location / {

28
sapl/api/views_sessao.py

@ -7,7 +7,8 @@ from drfautoapi.drfautoapi import ApiViewSetConstrutor, \
customize, wrapper_queryset_response_for_drf_action
from sapl.api.serializers import ChoiceSerializer,\
SessaoPlenariaECidadaniaSerializer
from sapl.sessao.models import SessaoPlenaria, ExpedienteSessao
from sapl.sessao.models import SessaoPlenaria, ExpedienteSessao,\
restringe_sessoes_visiveis
from sapl.utils import choice_anos_com_sessaoplenaria
@ -21,6 +22,13 @@ ApiViewSetConstrutor.build_class(
@customize(SessaoPlenaria)
class _SessaoPlenariaViewSet:
def get_queryset(self):
qs = super().get_queryset()
# actions com wrapper_queryset_response_for_drf_action trocam o model
if qs.model is SessaoPlenaria:
qs = restringe_sessoes_visiveis(qs, self.request.user)
return qs
@action(detail=False)
def years(self, request, *args, **kwargs):
years = choice_anos_com_sessaoplenaria()
@ -30,6 +38,7 @@ class _SessaoPlenariaViewSet:
@action(detail=True)
def expedientes(self, request, *args, **kwargs):
self.get_object()
return self.get_expedientes()
@wrapper_queryset_response_for_drf_action(model=ExpedienteSessao)
@ -45,3 +54,20 @@ class _SessaoPlenariaViewSet:
def ecidadania_list(self, request, *args, **kwargs):
self.serializer_class = SessaoPlenariaECidadaniaSerializer
return self.list(request, *args, **kwargs)
class _RestringeSessoesVisiveisMixin:
def get_queryset(self):
qs = super().get_queryset()
if self.request.user.is_authenticated:
return qs
return qs.filter(sessao_plenaria__in=restringe_sessoes_visiveis(
SessaoPlenaria.objects.all(), self.request.user))
for _model in apps.get_app_config('sessao').get_models():
if _model is not SessaoPlenaria and any(
f.name == 'sessao_plenaria' for f in _model._meta.fields):
customize(_model)(type('_%sViewSet' % _model.__name__,
(_RestringeSessoesVisiveisMixin,), {}))

124
sapl/base/media.py

@ -0,0 +1,124 @@
import mimetypes
import os
import re
from urllib.parse import quote
from django.conf import settings
from django.http import FileResponse, Http404, HttpResponse
from sapl.materia.models import Proposicao, pode_ver_proposicao
from sapl.protocoloadm.models import (DocumentoAcessorioAdministrativo,
DocumentoAdministrativo,
pode_ver_documento_adm)
from sapl.sessao.models import (JustificativaAusencia, Orador,
OradorExpediente, OradorOrdemDia,
SessaoPlenaria, restringe_sessoes_visiveis)
from sapl.utils import EXTENSOES_ATIVAS, tipo_ativo
X_ACCEL_PREFIX = '/media-interna/'
# O nginx do container só envia para cá os caminhos de PROTEGIDO_RE e de
# MODELOS_DE_SESSAO; o resto de /media/ sai direto do disco. Ao incluir um
# modelo aqui, incluí-lo também em docker/config/nginx/sapl.conf.
# Arquivos que dependem de autorização, em private/ ou em qualquer outra
# pasta (public/ guarda os enviados antes da criação de private/).
PROTEGIDO_RE = re.compile(
r'^sapl/[^/]+/(?:documentoadministrativo|'
r'documentoacessorioadministrativo|proposicao)/')
RESOLVEDORES = {
'documentoadministrativo': (
DocumentoAdministrativo, 'texto_integral', pode_ver_documento_adm),
'documentoacessorioadministrativo': (
DocumentoAcessorioAdministrativo, 'arquivo',
lambda user, obj: pode_ver_documento_adm(user, obj.documento)),
'proposicao': (
Proposicao, 'texto_original', pode_ver_proposicao),
}
# Arquivos gravados em sapl/public/<model>/<pk>/..., que seguem a
# visibilidade da sessão plenária a que pertencem.
MODELOS_DE_SESSAO = {m._meta.model_name: m for m in (
SessaoPlenaria, Orador, OradorExpediente, OradorOrdemDia,
JustificativaAusencia)}
def _de_sessao_oculta(request, partes):
model = MODELOS_DE_SESSAO.get(partes[2])
if request.user.is_authenticated or not model or len(partes) < 5 or \
partes[1] != 'public' or not re.fullmatch(r'[0-9]+', partes[3]):
return False
if model is SessaoPlenaria:
sessoes = SessaoPlenaria.objects.filter(pk=partes[3])
else:
sessoes = SessaoPlenaria.objects.filter(
pk__in=model.objects.filter(pk=partes[3]).values(
'sessao_plenaria_id'))
return sessoes.exists() and not restringe_sessoes_visiveis(
sessoes, request.user).exists()
def _nome_protegido(request, path):
"""Devolve o nome do arquivo gravado no objeto dono, se o usuário puder
vê-lo. O caminho servido vem do banco, nunca da URL."""
resolvedor = RESOLVEDORES.get(path.split('/')[2])
if not resolvedor:
raise Http404
model, campo, pode_ver = resolvedor
obj = model.objects.filter(
**{campo + '__in': [path, './' + path]}).first()
if not obj:
raise Http404
arquivo = getattr(obj, campo)
if not arquivo or not pode_ver(request.user, obj):
raise Http404
return arquivo.name[2:] if arquivo.name.startswith('./') else arquivo.name
def _caminho_em_media(nome):
raiz = os.path.realpath(settings.MEDIA_ROOT)
caminho = os.path.realpath(os.path.join(raiz, nome))
if not caminho.startswith(raiz + os.sep):
raise Http404
return caminho
def media(request, path):
partes = path.split('/')
if any(p in ('', '.', '..') for p in partes) or '\x00' in path:
raise Http404
protegido = path.startswith('sapl/private/') or bool(
PROTEGIDO_RE.match(path))
nome = _nome_protegido(request, path) if protegido else path
if len(partes) > 2 and _de_sessao_oculta(request, partes):
raise Http404
nome_arquivo = os.path.basename(nome)
content_type = (mimetypes.guess_type(nome_arquivo)[0] or
'application/octet-stream')
disposicao = 'inline'
if os.path.splitext(nome_arquivo)[1].lower() in EXTENSOES_ATIVAS or \
tipo_ativo(content_type):
content_type, disposicao = 'application/octet-stream', 'attachment'
caminho = _caminho_em_media(nome)
if settings.DEBUG:
try:
response = FileResponse(open(caminho, 'rb'),
content_type=content_type)
except (FileNotFoundError, IsADirectoryError):
raise Http404
else:
response = HttpResponse(content_type=content_type)
response['X-Accel-Redirect'] = X_ACCEL_PREFIX + quote(nome)
response['Content-Disposition'] = "%s; filename*=UTF-8''%s" % (
disposicao, quote(nome_arquivo))
response['X-Content-Type-Options'] = 'nosniff'
if protegido or (len(partes) > 2 and partes[2] in MODELOS_DE_SESSAO):
response['Cache-Control'] = 'private'
return response

384
sapl/base/tests/test_media.py

@ -0,0 +1,384 @@
import os
import re
from django.conf import settings
from django.contrib.auth import get_user_model
from django.contrib.auth.models import AnonymousUser
from django.contrib.auth.models import Permission
from django.contrib.contenttypes.models import ContentType
from django.core.cache import cache
from django.http import Http404
from django.utils import timezone
from model_bakery import baker
import pytest
from sapl.base.media import MODELOS_DE_SESSAO, RESOLVEDORES, media
from sapl.base.models import AppConfig, Autor
from sapl.materia.models import Proposicao
from sapl.protocoloadm.models import (DocumentoAcessorioAdministrativo,
DocumentoAdministrativo)
from sapl.utils import EXTENSOES_ATIVAS
DOC_PATH = 'sapl/private/documentoadministrativo/2026/1/texto.pdf'
ACESSORIO_PATH = 'sapl/public/documentoacessorioadministrativo/2026/1/a.pdf'
PROPOSICAO_PATH = 'sapl/private/proposicao/2026/1/proposicao.pdf'
@pytest.fixture(autouse=True)
def limpa_cache_appconfig():
# o cache em arquivo sobrevive ao rollback do teste
cache.clear()
yield
cache.clear()
@pytest.fixture(autouse=True)
def sem_debug(settings):
settings.DEBUG = False
def _config(comportamento):
cache.clear()
AppConfig.objects.all().delete()
baker.make(AppConfig, documentos_administrativos=comportamento)
def _login(client, username, *perms):
user = baker.make(get_user_model(), username=username)
for perm in perms:
app_label, codename = perm.split('.')
perm_obj = Permission.objects.filter(
content_type__app_label=app_label, codename=codename).first()
if not perm_obj:
perm_obj = Permission.objects.create(
codename=codename, name=codename,
content_type=ContentType.objects.get_for_model(
DocumentoAdministrativo))
user.user_permissions.add(perm_obj)
client.force_login(user)
return user
def _get(client, path):
return client.get('/media/' + path)
def _assert_servido(response, path):
assert response.status_code == 200
assert response['X-Accel-Redirect'] == '/media-interna/' + path
assert response['X-Content-Type-Options'] == 'nosniff'
assert response['Cache-Control'] == 'private'
@pytest.mark.django_db(transaction=False)
@pytest.mark.parametrize('comportamento,restrito,perfil,pode_ver', [
('O', False, None, True),
('O', True, None, False),
('O', True, 'logado', True),
('R', False, None, False),
('R', False, 'logado', False),
('R', True, None, False),
('R', True, 'logado', False),
('R', True, 'com_permissao', True),
])
def test_documento_administrativo(client, comportamento, restrito, perfil,
pode_ver):
_config(comportamento)
baker.make(DocumentoAdministrativo, restrito=restrito,
texto_integral=DOC_PATH)
if perfil == 'logado':
_login(client, 'logado')
elif perfil == 'com_permissao':
_login(client, 'perm',
'protocoloadm.detail_documentoadministrativo')
response = _get(client, DOC_PATH)
if pode_ver:
_assert_servido(response, DOC_PATH)
assert response['Content-Type'] == 'application/pdf'
else:
assert response.status_code == 404
assert not response.has_header('X-Accel-Redirect')
@pytest.mark.django_db(transaction=False)
@pytest.mark.parametrize('restrito,pode_ver', [(False, True), (True, False)])
def test_documento_acessorio_segue_documento(client, restrito, pode_ver):
_config('O')
documento = baker.make(DocumentoAdministrativo, restrito=restrito)
baker.make(DocumentoAcessorioAdministrativo, documento=documento,
arquivo=ACESSORIO_PATH)
response = _get(client, ACESSORIO_PATH)
assert response.status_code == (200 if pode_ver else 404)
@pytest.mark.django_db(transaction=False)
def test_caminho_legado_e_prefixo_ponto_barra(client):
_config('O')
legado = 'sapl/public/documentoadministrativo/2010/7/antigo.pdf'
baker.make(DocumentoAdministrativo, texto_integral=legado)
baker.make(DocumentoAdministrativo, texto_integral='./' + DOC_PATH)
_assert_servido(_get(client, legado), legado)
_assert_servido(_get(client, DOC_PATH), DOC_PATH)
@pytest.mark.django_db(transaction=False)
@pytest.mark.parametrize('path', [
'sapl/private/documentoadministrativo/2026/1/inexistente.pdf',
'sapl/private/outromodelo/2026/1/texto.pdf',
'sapl/private/documentoadministrativo/../../public/x.pdf',
'sapl/private/documentoadministrativo/%2e%2e/x.pdf',
])
def test_caminho_invalido(client, path):
_config('O')
baker.make(DocumentoAdministrativo, texto_integral=DOC_PATH)
assert _get(client, path).status_code == 404
@pytest.mark.django_db(transaction=False)
def test_extensao_ativa_e_baixada(client):
_config('O')
path = 'sapl/private/documentoadministrativo/2026/1/x.html'
baker.make(DocumentoAdministrativo, texto_integral=path)
response = _get(client, path)
_assert_servido(response, path)
assert response['Content-Type'] == 'application/octet-stream'
assert response['Content-Disposition'].startswith('attachment;')
@pytest.mark.django_db(transaction=False)
def test_nome_nao_ascii_codificado(client):
_config('O')
path = 'sapl/private/documentoadministrativo/2026/1/ofício_nº1.pdf'
baker.make(DocumentoAdministrativo, texto_integral=path)
response = _get(client, path)
assert response.status_code == 200
assert response['X-Accel-Redirect'] == (
'/media-interna/sapl/private/documentoadministrativo/2026/1/'
'of%C3%ADcio_n%C2%BA1.pdf')
ESTADOS_PROPOSICAO = ('rascunho', 'enviada', 'devolvida', 'incorporada')
PERMS_DOCADM = ['protocoloadm.add_documentoadministrativo',
'protocoloadm.change_documentoadministrativo',
'protocoloadm.detail_documentoadministrativo']
def _proposicao(estado):
autor = baker.make(Autor)
operador = baker.make(get_user_model(), username='operador')
autor.operadores.add(operador)
agora = timezone.now()
# mesmos campos gravados pelo envio, devolução e incorporação
baker.make(Proposicao, autor=autor, texto_original=PROPOSICAO_PATH,
data_envio=agora if estado in ('enviada', 'incorporada')
else None,
data_recebimento=agora if estado == 'incorporada' else None,
data_devolucao=agora if estado == 'devolvida' else None)
return operador
def _login_proposicao(client, perfil, operador):
if perfil == 'operador':
client.force_login(operador)
elif perfil == 'docadm':
user = _login(client, 'docadm', *PERMS_DOCADM)
assert user.has_module_perms('protocoloadm')
elif perfil == 'outro':
_login(client, 'outro')
elif perfil:
_login(client, perfil, 'materia.detail_proposicao_' + perfil)
def _pode_ver_proposicao(estado, perfil):
return perfil == 'operador' or perfil == estado != 'rascunho'
GRADE_PROPOSICAO = [
(estado, perfil, _pode_ver_proposicao(estado, perfil))
for estado in ESTADOS_PROPOSICAO
for perfil in (None, 'operador', 'outro', 'docadm', 'enviada',
'devolvida', 'incorporada')]
@pytest.mark.django_db(transaction=False)
@pytest.mark.parametrize('estado,perfil,pode_ver', GRADE_PROPOSICAO)
def test_proposicao(client, estado, perfil, pode_ver):
_login_proposicao(client, perfil, _proposicao(estado))
response = _get(client, PROPOSICAO_PATH)
if pode_ver:
_assert_servido(response, PROPOSICAO_PATH)
else:
assert response.status_code == 404
@pytest.mark.django_db(transaction=False)
def test_arquivo_publico_passa_por_x_accel(client):
path = 'sapl/public/materialegislativa/2026/1/materia.pdf'
response = _get(client, path)
assert response.status_code == 200
assert response['X-Accel-Redirect'] == '/media-interna/' + path
assert response['Content-Type'] == 'application/pdf'
assert response['X-Content-Type-Options'] == 'nosniff'
assert not response.has_header('Cache-Control')
@pytest.mark.django_db(transaction=False)
@pytest.mark.parametrize('nome', ['x.svg', 'x.html', 'x.XML', 'x.js',
'x.xht', 'x.rdf', 'x.xslt', 'x.kml'])
def test_arquivo_publico_ativo_e_baixado(client, nome):
response = _get(client, 'sapl/public/normajuridica/2026/1/' + nome)
assert response['Content-Type'] == 'application/octet-stream'
assert response['Content-Disposition'].startswith('attachment;')
@pytest.mark.django_db(transaction=False)
@pytest.mark.parametrize('path', [
'sapl/public/../private/documentoadministrativo/2026/1/texto.pdf',
'sapl//private/documentoadministrativo/2026/1/texto.pdf',
'sapl/./private/documentoadministrativo/2026/1/texto.pdf',
'sapl/public/normajuridica/',
])
def test_caminho_nao_normalizado(client, path):
assert _get(client, path).status_code == 404
@pytest.mark.django_db(transaction=False)
def test_debug_serve_arquivo(rf, settings, tmp_path):
settings.DEBUG = True
settings.MEDIA_ROOT = str(tmp_path)
_config('R')
publico = tmp_path / 'sapl/public/normajuridica/1/n.pdf'
publico.parent.mkdir(parents=True)
publico.write_bytes(b'%PDF-1.4 norma')
baker.make(DocumentoAdministrativo, texto_integral=DOC_PATH)
def chama(path):
request = rf.get('/media/' + path)
request.user = AnonymousUser()
return media(request, path)
response = chama('sapl/public/normajuridica/1/n.pdf')
assert b''.join(response.streaming_content) == b'%PDF-1.4 norma'
assert not response.has_header('X-Accel-Redirect')
for path in (DOC_PATH, 'sapl/public/normajuridica/1/nao.pdf'):
with pytest.raises(Http404):
chama(path)
@pytest.mark.django_db(transaction=False)
@pytest.mark.parametrize('debug', [False, True])
def test_link_simbolico_para_fora_de_media(rf, settings, tmp_path, debug):
settings.DEBUG = debug
media_root = tmp_path / 'media'
fora = tmp_path / 'fora'
(media_root / 'sapl/public').mkdir(parents=True)
fora.mkdir()
(fora / 'segredo.txt').write_bytes(b'segredo')
(media_root / 'sapl/public/link').symlink_to(fora)
settings.MEDIA_ROOT = str(media_root)
request = rf.get('/media/sapl/public/link/segredo.txt')
request.user = AnonymousUser()
with pytest.raises(Http404):
media(request, 'sapl/public/link/segredo.txt')
@pytest.mark.django_db(transaction=False)
@pytest.mark.parametrize('path', ['/etc/passwd', 'sapl/public//etc/passwd'])
def test_caminho_absoluto(rf, path):
request = rf.get('/media/' + path)
request.user = AnonymousUser()
with pytest.raises(Http404):
media(request, path)
@pytest.mark.django_db(transaction=False)
def test_modelo_protegido_fora_de_private_e_public(client):
_config('O')
path = 'sapl/outra/documentoadministrativo/2026/1/texto.pdf'
baker.make(DocumentoAdministrativo, restrito=True, texto_integral=path)
assert _get(client, path).status_code == 404
@pytest.mark.django_db(transaction=False)
def test_senha_fraca_nao_redireciona_media(client):
_login(client, 'fraca')
session = client.session
session['weak_password'] = True
session.save()
response = _get(client, 'sapl/public/casalegislativa/logotipo.png')
assert response.status_code == 200
assert response.has_header('X-Accel-Redirect')
@pytest.mark.django_db(transaction=False)
@pytest.mark.parametrize('estado,perfil,pode_ver', GRADE_PROPOSICAO)
def test_proposicao_texto(client, settings, tmp_path, estado, perfil,
pode_ver):
settings.MEDIA_ROOT = str(tmp_path)
arquivo = tmp_path / PROPOSICAO_PATH
arquivo.parent.mkdir(parents=True)
arquivo.write_bytes(b'%PDF-1.4 proposicao')
_login_proposicao(client, perfil, _proposicao(estado))
pk = Proposicao.objects.get(texto_original=PROPOSICAO_PATH).pk
response = client.get('/proposicao/texto/%s' % pk)
if pode_ver:
assert response.status_code == 200
assert response.content == b'%PDF-1.4 proposicao'
else:
assert response.status_code == 302
assert response['Location'] == '/proposicao/%s' % pk
def test_nginx_envia_ao_django_todos_os_caminhos_protegidos():
caminho = os.path.join(settings.PROJECT_DIR, 'docker', 'config',
'nginx', 'sapl.conf')
with open(caminho) as f:
conf = f.read()
regex = re.search(r'location ~\*? \^/media/sapl/\(private/\|'
r'\[\^/\]\+/\(([a-z|]+)\)/\)', conf)
assert regex, 'location de arquivos protegidos não encontrada'
no_nginx = set(regex.group(1).split('|'))
assert set(RESOLVEDORES) | set(MODELOS_DE_SESSAO) <= no_nginx
assert conf.index(regex.group(0)) < conf.index('location /media/ {')
def test_nginx_forca_download_das_extensoes_ativas():
caminho = os.path.join(settings.PROJECT_DIR, 'docker', 'config',
'nginx', 'sapl.conf')
with open(caminho) as f:
conf = f.read()
regex = re.search(r'location ~\*? \^/media/\.\+\\\.\(([a-z?|]+)\)\$',
conf)
assert regex, 'location de extensões ativas não encontrada'
no_nginx = set()
for ext in regex.group(1).split('|'):
if ext.endswith('?'):
no_nginx |= {ext[:-2], ext[:-1]}
else:
no_nginx.add(ext)
assert {'.' + e for e in no_nginx} == EXTENSOES_ATIVAS

19
sapl/materia/models.py

@ -1154,6 +1154,25 @@ class Proposicao(models.Model):
update_fields=update_fields)
def pode_ver_proposicao(user, proposicao):
"""Texto original da proposição: nunca público. Operadores do autor veem
sempre; demais usuários seguem as permissões por estado da página de
detalhe (ProposicaoCrud.DetailView.dispatch)."""
if not user.is_authenticated:
return False
if proposicao.autor and proposicao.autor.operadores.filter(
id=user.id).exists():
return True
# devolução limpa data_envio: checar antes de tratar como rascunho
if proposicao.data_devolucao:
return user.has_perm('materia.detail_proposicao_devolvida')
if not proposicao.data_envio:
return False
if not proposicao.data_recebimento:
return user.has_perm('materia.detail_proposicao_enviada')
return user.has_perm('materia.detail_proposicao_incorporada')
class HistoricoProposicao(models.Model):
STATUS_PROPOSICAO = Choices(('E', 'ENVIADA', _('Enviada')),
('R', 'RECEBIDA', _('Recebida')),

64
sapl/materia/tests/test_acessorio_em_lote.py

@ -0,0 +1,64 @@
import os
from django.core.files.uploadedfile import SimpleUploadedFile
from django.urls import reverse
from model_bakery import baker
import pytest
from sapl.materia.models import (DocumentoAcessorio, MateriaLegislativa,
TipoDocumento)
@pytest.fixture
def dirs(settings, tmp_path, monkeypatch):
media = tmp_path / 'media'
temp = tmp_path / 'temp'
media.mkdir()
temp.mkdir()
settings.MEDIA_ROOT = str(media)
monkeypatch.setattr('sapl.materia.views.MEDIA_ROOT', str(media))
monkeypatch.setattr('sapl.materia.views.get_tempfile_dir',
lambda: str(temp))
return media, temp
def _post(client, nome, conteudo):
materias = baker.make(MateriaLegislativa, _quantity=2)
tipo = baker.make(TipoDocumento, descricao='Parecer')
client.post(reverse('sapl.materia:acessorio_em_lote'), {
'materia_id': [m.pk for m in materias],
'tipo': tipo.descricao,
'nome': 'Parecer',
'autor': 'Comissão',
'data': '12/10/2026',
'ementa': 'Ementa',
'arquivo': SimpleUploadedFile(nome, conteudo),
})
return materias
@pytest.mark.django_db(transaction=False)
def test_arquivo_invalido_nao_grava_nada(admin_client, dirs):
media, temp = dirs
_post(admin_client, 'x.html', b'%PDF-1.4\n<html><script></script>')
assert not DocumentoAcessorio.objects.exists()
assert os.listdir(media) == []
assert os.listdir(temp) == []
@pytest.mark.django_db(transaction=False)
def test_arquivo_valido_copiado_para_cada_materia(admin_client, dirs):
media, temp = dirs
materias = _post(admin_client, 'parecer.pdf', b'%PDF-1.4\n%%EOF')
docs = DocumentoAcessorio.objects.filter(materia__in=materias)
assert docs.count() == 2
for doc in docs:
assert doc.arquivo.name == (
'sapl/public/documentoacessorio/2026/%s/parecer.pdf' % doc.pk)
assert (media / doc.arquivo.name).read_bytes() == b'%PDF-1.4\n%%EOF'
assert os.listdir(media) == ['sapl']
assert os.listdir(temp) == []

98
sapl/materia/tests/test_proposicao_confirmar.py

@ -0,0 +1,98 @@
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Permission
from django.contrib.contenttypes.models import ContentType
from django.urls import reverse
from django.utils import timezone
from model_bakery import baker
import pytest
from sapl.base.models import AppConfig
from sapl.materia.models import (Proposicao, TipoMateriaLegislativa,
TipoProposicao)
from sapl.utils import SEPARADOR_HASH_PROPOSICAO, gerar_hash_arquivo
PROPOSICAO_PATH = 'sapl/private/proposicao/2026/1/proposicao.pdf'
PERMS_DOCADM = ['add_documentoadministrativo',
'change_documentoadministrativo',
'detail_documentoadministrativo']
def _login(client, app_label, *codenames):
user = baker.make(get_user_model(), username=app_label)
user.user_permissions.add(*Permission.objects.filter(
content_type__app_label=app_label, codename__in=codenames))
client.force_login(user)
return user
def _login_docadm(client):
user = _login(client, 'protocoloadm', *PERMS_DOCADM)
assert user.has_module_perms('protocoloadm')
def _login_protocolo(client):
_login(client, 'materia', 'detail_proposicao_enviada')
def _proposicao_enviada(settings, tmp_path):
settings.MEDIA_ROOT = str(tmp_path)
baker.make(AppConfig)
arquivo = tmp_path / PROPOSICAO_PATH
arquivo.parent.mkdir(parents=True)
arquivo.write_bytes(b'%PDF-1.4 proposicao')
tipo = baker.make(
TipoProposicao,
content_type=ContentType.objects.get_for_model(TipoMateriaLegislativa),
object_id=baker.make(TipoMateriaLegislativa).pk)
proposicao = baker.make(Proposicao, tipo=tipo,
texto_original=PROPOSICAO_PATH,
data_envio=timezone.now())
hasher = gerar_hash_arquivo(str(arquivo), str(proposicao.pk))
return proposicao, hasher[1:].split(SEPARADOR_HASH_PROPOSICAO)[0]
def _url_confirmar(proposicao, hash_):
return reverse('sapl.materia:proposicao-confirmar',
kwargs={'hash': hash_, 'pk': proposicao.pk})
@pytest.mark.django_db(transaction=False)
def test_receber_exige_permissao_de_proposicao(client):
_login_docadm(client)
url = reverse('sapl.materia:receber-proposicao')
assert client.get(url).status_code == 403
@pytest.mark.django_db(transaction=False)
def test_receber_operador_de_protocolo(client):
_login_protocolo(client)
url = reverse('sapl.materia:receber-proposicao')
assert client.get(url).status_code == 200
@pytest.mark.django_db(transaction=False)
def test_confirmar_exige_permissao_de_proposicao(client, settings, tmp_path):
proposicao, hash_ = _proposicao_enviada(settings, tmp_path)
_login_docadm(client)
response = client.get(_url_confirmar(proposicao, hash_))
assert response.status_code == 403
@pytest.mark.django_db(transaction=False)
def test_confirmar_operador_de_protocolo_ve_texto(client, settings,
tmp_path):
settings.DEBUG = False
proposicao, hash_ = _proposicao_enviada(settings, tmp_path)
_login_protocolo(client)
response = client.get(_url_confirmar(proposicao, hash_))
assert response.status_code == 200
assert proposicao.texto_original.url.encode() in response.content
texto = client.get(proposicao.texto_original.url)
assert texto.status_code == 200
assert texto['X-Accel-Redirect'] == '/media-interna/' + PROPOSICAO_PATH

72
sapl/materia/views.py

@ -5,6 +5,7 @@ import logging
import os
from random import choice
import shutil
import tempfile
from string import ascii_letters, digits
import time
import zipfile
@ -15,7 +16,8 @@ from django.conf import settings
from django.contrib import messages
from django.contrib.auth.decorators import permission_required
from django.contrib.auth.mixins import PermissionRequiredMixin
from django.core.exceptions import ObjectDoesNotExist, MultipleObjectsReturned, ValidationError
from django.core.exceptions import ObjectDoesNotExist, MultipleObjectsReturned, \
SuspiciousFileOperation, ValidationError
from django.db import transaction
from django.db.models import Max, Q
from django.http import HttpResponse, JsonResponse
@ -52,11 +54,12 @@ from sapl.materia.forms import (AnexadaForm, AutoriaForm, AutoriaMultiCreateForm
from sapl.norma.models import LegislacaoCitada
from sapl.parlamentares.models import Legislatura
from sapl.protocoloadm.models import Protocolo
from sapl.settings import MAX_DOC_UPLOAD_SIZE, MEDIA_ROOT, RATE_LIMITER_RATE
from sapl.settings import MEDIA_ROOT, RATE_LIMITER_RATE
from sapl.utils import (autor_label, autor_modal, gerar_hash_arquivo, get_base_url,
get_client_ip, get_mime_type_from_file_extension, lista_anexados,
mail_service_configured, montar_row_autor, SEPARADOR_HASH_PROPOSICAO,
show_results_filter_set, get_tempfile_dir,
restringe_tipos_de_arquivo_txt, validar_arquivo,
google_recaptcha_configured, MultiFormatOutputMixin, ratelimit_ip)
from .forms import (AcessorioEmLoteFilterSet, AcompanhamentoMateriaForm,
@ -72,7 +75,7 @@ from .forms import (AcessorioEmLoteFilterSet, AcompanhamentoMateriaForm,
ReceberProposicaoForm, RelatoriaForm,
TramitacaoEmLoteFilterSet, TramitacaoEmLoteForm,
UnidadeTramitacaoForm, StatusTramitacaoFilterSet)
from .models import (AcompanhamentoMateria, Anexada, AssuntoMateria, Autoria, DespachoInicial,
from .models import (pode_ver_proposicao, AcompanhamentoMateria, Anexada, AssuntoMateria, Autoria, DespachoInicial,
DocumentoAcessorio, MateriaAssunto, MateriaLegislativa, Numeracao, Orgao,
Origem, Proposicao, RegimeTramitacao, Relatoria, StatusTramitacao,
TipoDocumento, TipoFimRelatoria, TipoMateriaLegislativa, TipoProposicao,
@ -104,14 +107,10 @@ def proposicao_texto(request, pk):
username = request.user.username
logger.debug('user=' + username +
'. Tentando obter objeto Proposicao com pk = {}.'.format(pk))
proposicao = Proposicao.objects.get(pk=pk)
proposicao = get_object_or_404(Proposicao, pk=pk)
if proposicao.texto_original:
if (not proposicao.data_recebimento and
not proposicao.autor.operadores.filter(
id=request.user.id
).exists()
):
if not pode_ver_proposicao(request.user, proposicao):
logger.error("user=" + username + ". Usuário ({}) não tem permissão para acessar o texto original."
.format(request.user.id))
messages.error(request, _(
@ -577,8 +576,8 @@ class ProposicaoRecebida(PermissionRequiredMixin, ListView):
return context
class ReceberProposicao(PermissionRequiredForAppCrudMixin, FormView):
app_label = sapl.protocoloadm.apps.AppConfig.label
class ReceberProposicao(PermissionRequiredMixin, FormView):
permission_required = ('materia.detail_proposicao_enviada', )
template_name = "crud/form.html"
form_class = ReceberProposicaoForm
@ -661,8 +660,8 @@ class RetornarProposicao(UpdateView):
request, *args, **kwargs)
class ConfirmarProposicao(PermissionRequiredForAppCrudMixin, UpdateView):
app_label = sapl.protocoloadm.apps.AppConfig.label
class ConfirmarProposicao(PermissionRequiredMixin, UpdateView):
permission_required = ('materia.detail_proposicao_enviada', )
template_name = "materia/confirmar_proposicao.html"
model = Proposicao
form_class = ConfirmarProposicaoForm, DevolverProposicaoForm
@ -2330,17 +2329,15 @@ class DocumentoAcessorioEmLoteView(PermissionRequiredMixin, FilterView):
messages.add_message(request, messages.ERROR, msg)
return self.get(request, self.kwargs)
if request.FILES['arquivo'].size > MAX_DOC_UPLOAD_SIZE:
msg = _("O arquivo Anexo de Texto Integral deve ser menor que {0:.1f} MB, \
o tamanho atual desse arquivo é {1:.1f} MB"
.format((MAX_DOC_UPLOAD_SIZE / 1024) / 1024, (request.FILES['arquivo'].size / 1024) / 1024))
messages.add_message(request, messages.ERROR, msg)
arquivo = request.FILES['arquivo']
try:
validar_arquivo(arquivo, 'Arquivo')
restringe_tipos_de_arquivo_txt(arquivo)
except ValidationError as e:
for m in e.messages:
messages.add_message(request, messages.ERROR, m)
return self.get(request, self.kwargs)
tmp_name = os.path.join(MEDIA_ROOT, request.FILES['arquivo'].name)
with open(tmp_name, 'wb') as destination:
for chunk in request.FILES['arquivo'].chunks():
destination.write(chunk)
try:
doc_data = tz.localize(datetime.strptime(
request.POST['data'], "%d/%m/%Y"))
@ -2350,9 +2347,20 @@ class DocumentoAcessorioEmLoteView(PermissionRequiredMixin, FilterView):
messages.add_message(request, messages.ERROR, msg)
self.logger.error("User={}. {}. Data inserida: {}".format(
username, str(msg), request.POST['data']))
os.remove(tmp_name)
return self.get(request, self.kwargs)
nome_arquivo = os.path.basename(arquivo.name)
if nome_arquivo in ('', '.', '..'):
messages.add_message(request, messages.ERROR,
_('Nome de arquivo inválido.'))
return self.get(request, self.kwargs)
# Fora de MEDIA_ROOT, para não ficar acessível em /media/
with tempfile.NamedTemporaryFile(dir=get_tempfile_dir(),
delete=False) as tmp:
for chunk in arquivo.chunks():
tmp.write(chunk)
try:
for materia_id in marcadas:
doc = DocumentoAcessorio()
doc.materia_id = materia_id
@ -2361,9 +2369,8 @@ class DocumentoAcessorioEmLoteView(PermissionRequiredMixin, FilterView):
doc.data = doc_data
doc.autor = request.POST['autor']
doc.ementa = request.POST['ementa']
doc.arquivo.name = tmp_name
try:
doc.clean_fields()
doc.clean_fields(exclude=['arquivo'])
except ValidationError as e:
for m in ['%s: %s' % (DocumentoAcessorio()._meta.get_field(k).verbose_name, '</br>'.join(v))
for k, v in e.message_dict.items()]:
@ -2371,8 +2378,7 @@ class DocumentoAcessorioEmLoteView(PermissionRequiredMixin, FilterView):
# 'verbose_name do nome do campo': 'mensagem de erro'
messages.add_message(request, messages.ERROR, m)
self.logger.error("User={}. {}. Nome do arquivo: {}.".format(
username, str(msg), request.FILES['arquivo'].name))
os.remove(tmp_name)
username, str(m), arquivo.name))
return self.get(request, self.kwargs)
doc.save()
diretorio = os.path.join(MEDIA_ROOT,
@ -2381,13 +2387,17 @@ class DocumentoAcessorioEmLoteView(PermissionRequiredMixin, FilterView):
str(doc.id))
if not os.path.exists(diretorio):
os.makedirs(diretorio)
file_path = os.path.join(diretorio,
request.FILES['arquivo'].name)
shutil.copy2(tmp_name, file_path)
diretorio = os.path.normpath(diretorio)
file_path = os.path.normpath(
os.path.join(diretorio, nome_arquivo))
if not file_path.startswith(diretorio + os.sep):
raise SuspiciousFileOperation(nome_arquivo)
shutil.copyfile(tmp.name, file_path)
doc.arquivo.name = file_path.split(
MEDIA_ROOT + "/")[1] # Retira MEDIA_ROOT do nome
doc.save()
os.remove(tmp_name)
finally:
os.remove(tmp.name)
msg = _('Documento(s) criado(s).')
messages.add_message(request, messages.SUCCESS, msg)

4
sapl/middleware/check_password.py

@ -1,5 +1,6 @@
import logging
from django.conf import settings
from django.shortcuts import redirect
from django.urls import reverse
@ -14,7 +15,8 @@ class CheckWeakPasswordMiddleware:
if request.user.is_authenticated and \
request.session.get('weak_password', False) and \
request.path != reverse('sapl.base:alterar_senha') and \
request.path != reverse('sapl.base:logout'):
request.path != reverse('sapl.base:logout') and \
not request.path.startswith(settings.MEDIA_URL):
logging.warning(f"Usuário {request.user.username} possui senha fraca.")
return redirect('sapl.base:alterar_senha')

10
sapl/protocoloadm/models.py

@ -6,7 +6,8 @@ from django.utils.functional import cached_property
from django.utils.translation import ugettext_lazy as _
from model_utils import Choices
from sapl.base.models import Autor, AppConfig as SaplAppConfig
from sapl.base.models import Autor, AppConfig as SaplAppConfig, \
DOC_ADM_OSTENSIVO
from sapl.materia.models import TipoMateriaLegislativa, UnidadeTramitacao,\
MateriaLegislativa
from sapl.utils import (RANGE_ANOS, YES_NO_CHOICES, texto_upload_path,
@ -344,6 +345,13 @@ class DocumentoAdministrativo(models.Model):
update_fields=update_fields)
def pode_ver_documento_adm(user, documento):
"""Mesmas regras da página de detalhe do documento (CRUD)."""
if SaplAppConfig.attr('documentos_administrativos') == DOC_ADM_OSTENSIVO:
return not (documento.restrito and user.is_anonymous)
return user.has_perm('protocoloadm.detail_documentoadministrativo')
class DocumentoAcessorioAdministrativo(models.Model):
documento = models.ForeignKey(DocumentoAdministrativo,
on_delete=models.PROTECT)

163
sapl/protocoloadm/tests/test_docadm_acesso.py

@ -0,0 +1,163 @@
from io import BytesIO
from PyPDF4 import PdfFileWriter
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Permission
from django.contrib.contenttypes.models import ContentType
from django.core.cache import cache
from django.core.files.uploadedfile import SimpleUploadedFile
from django.urls import reverse
from model_bakery import baker
import pytest
from sapl.base.models import AppConfig
from sapl.protocoloadm.models import (DocumentoAcessorioAdministrativo,
DocumentoAdministrativo)
@pytest.fixture(autouse=True)
def limpa_cache_appconfig():
# o cache em arquivo sobrevive ao rollback do teste
cache.clear()
yield
cache.clear()
@pytest.fixture
def media_tmp(settings, tmp_path):
settings.MEDIA_ROOT = str(tmp_path)
return tmp_path
def _config(comportamento):
cache.clear()
AppConfig.objects.all().delete()
baker.make(AppConfig, documentos_administrativos=comportamento)
def _documento(restrito):
documento = baker.make(DocumentoAdministrativo, restrito=restrito)
documento.texto_integral = SimpleUploadedFile(
'texto.pdf', b'%PDF-1.4 conteudo', content_type='application/pdf')
documento.save()
return documento
def _client(client, perfil):
if perfil == 'anonimo':
return client
user = baker.make(get_user_model(), username=perfil)
if perfil == 'com_permissao':
ct = ContentType.objects.get_for_model(DocumentoAdministrativo)
perm, _ = Permission.objects.get_or_create(
codename='detail_documentoadministrativo', content_type=ct,
defaults={'name': 'detail'})
user.user_permissions.add(perm)
client.force_login(user)
return client
ACESSO = [
# (comportamento, restrito, perfil, pode_ver)
('O', False, 'anonimo', True),
('O', False, 'sem_permissao', True),
('O', False, 'com_permissao', True),
('O', True, 'anonimo', False),
('O', True, 'sem_permissao', True),
('O', True, 'com_permissao', True),
('R', False, 'anonimo', False),
('R', False, 'sem_permissao', False),
('R', False, 'com_permissao', True),
('R', True, 'anonimo', False),
('R', True, 'sem_permissao', False),
('R', True, 'com_permissao', True),
]
@pytest.mark.django_db(transaction=False)
@pytest.mark.parametrize('comportamento,restrito,perfil,pode_ver', ACESSO)
def test_doc_texto_integral_acesso(client, media_tmp, comportamento,
restrito, perfil, pode_ver):
_config(comportamento)
documento = _documento(restrito)
client = _client(client, perfil)
response = client.get(reverse('sapl.protocoloadm:doc_texto_integral',
kwargs={'pk': documento.pk}))
assert response.status_code == (200 if pode_ver else 404)
def _pdf():
writer = PdfFileWriter()
writer.addBlankPage(72, 72)
data = BytesIO()
writer.write(data)
return data.getvalue()
@pytest.mark.django_db(transaction=False)
@pytest.mark.parametrize('comportamento,restrito,perfil,pode_ver', ACESSO)
def test_pdf_docacessorios_acesso(client, media_tmp, monkeypatch,
comportamento, restrito, perfil, pode_ver):
# create_pdf_docacessorios usa o MEDIA_ROOT importado de sapl.settings
monkeypatch.setattr('sapl.protocoloadm.views.MEDIA_ROOT', str(media_tmp))
_config(comportamento)
documento = baker.make(DocumentoAdministrativo, restrito=restrito)
acessorio = baker.make(DocumentoAcessorioAdministrativo,
documento=documento)
acessorio.arquivo = SimpleUploadedFile(
'acessorio.pdf', _pdf(), content_type='application/pdf')
acessorio.save()
client = _client(client, perfil)
response = client.get(reverse('sapl.protocoloadm:merge_docacessorios',
kwargs={'pk': documento.pk}))
if pode_ver:
assert response.status_code == 200
assert response['Content-Type'] == 'application/pdf'
assert response.content.startswith(b'%PDF')
else:
assert response.status_code == 404
@pytest.mark.django_db(transaction=False)
def test_pdf_docacessorios_sem_acessorios_volta_para_lista(client):
_config('O')
documento = baker.make(DocumentoAdministrativo, restrito=False)
response = client.get(reverse('sapl.protocoloadm:merge_docacessorios',
kwargs={'pk': documento.pk}))
assert response.status_code == 302
assert response['Location'] == reverse(
'sapl.protocoloadm:documentoacessorioadministrativo_list',
kwargs={'pk': documento.pk})
@pytest.mark.django_db(transaction=False)
@pytest.mark.parametrize('url_name', ['doc_texto_integral',
'merge_docacessorios'])
def test_docadm_pk_inexistente(client, url_name):
_config('O')
response = client.get(reverse('sapl.protocoloadm:' + url_name,
kwargs={'pk': 999999}))
assert response.status_code == 404
@pytest.mark.django_db(transaction=False)
def test_detalhe_e_arquivo_usam_a_mesma_configuracao(client, media_tmp):
# com mais de uma linha em AppConfig, página e arquivo leem a mesma
_config('O')
baker.make(AppConfig, documentos_administrativos='R')
documento = _documento(False)
detalhe = client.get(reverse(
'sapl.protocoloadm:documentoadministrativo_detail',
kwargs={'pk': documento.pk}))
texto = client.get(reverse('sapl.protocoloadm:doc_texto_integral',
kwargs={'pk': documento.pk}))
assert detalhe.status_code == 302
assert texto.status_code == 404

32
sapl/protocoloadm/views.py

@ -32,7 +32,8 @@ from django_filters.views import FilterView
import sapl
from sapl.base.email_utils import do_envia_email_confirmacao
from sapl.base.models import Autor, CasaLegislativa, AppConfig
from sapl.base.models import (Autor, CasaLegislativa, AppConfig,
DOC_ADM_OSTENSIVO)
from sapl.comissoes.models import Comissao
from sapl.crud.base import (Crud, CrudAux, MasterDetailCrud, make_pagination,
RP_LIST, RP_DETAIL)
@ -42,7 +43,7 @@ from sapl.parlamentares.models import Legislatura, Parlamentar
from sapl.protocoloadm.forms import VinculoDocAdminMateriaForm,\
VinculoDocAdminMateriaEmLoteFilterSet
from sapl.protocoloadm.models import Protocolo, DocumentoAdministrativo,\
VinculoDocAdminMateria
VinculoDocAdminMateria, pode_ver_documento_adm
from sapl.relatorios.views import relatorio_doc_administrativos
from sapl.utils import (create_barcode, get_base_url, get_client_ip,
get_mime_type_from_file_extension, lista_anexados,
@ -101,15 +102,9 @@ def recuperar_materia_protocolo(request):
def doc_texto_integral(request, pk):
can_see = True
documento = get_object_or_404(DocumentoAdministrativo, pk=pk)
if not request.user.is_authenticated:
app_config = AppConfig.objects.last()
if app_config and app_config.documentos_administrativos == 'R':
can_see = False
if can_see:
documento = DocumentoAdministrativo.objects.get(pk=pk)
if pode_ver_documento_adm(request.user, documento):
if documento.texto_integral:
arquivo = documento.texto_integral
@ -127,6 +122,8 @@ def doc_texto_integral(request, pk):
def get_pdf_docacessorios(request, pk):
documento_administrativo = get_object_or_404(DocumentoAdministrativo, pk=pk)
if not pode_ver_documento_adm(request.user, documento_administrativo):
raise Http404
logger = logging.getLogger(__name__)
username = 'Usuário anônimo' if request.user.is_anonymous else request.user.username
try:
@ -137,20 +134,23 @@ def get_pdf_docacessorios(request, pk):
logger.error("user= {}.Não há arquivos cadastrados".format(username))
msg = _('Não há arquivos cadastrados nesses documentos acessórios.')
messages.add_message(request, messages.ERROR, msg)
return redirect(reverse('sapl.materia:documentoacessorio_list',
return redirect(reverse(
'sapl.protocoloadm:documentoacessorioadministrativo_list',
kwargs={'pk': pk}))
except Exception as e:
logger.error("user= {}.Um erro inesperado ocorreu na criação do pdf de documentos acessorios: {}"
.format(username, str(e)))
msg = _('Um erro inesperado ocorreu. Entre em contato com o suporte do SAPL.')
messages.add_message(request, messages.ERROR, msg)
return redirect(reverse('sapl.materia:documentoacessorio_list',
return redirect(reverse(
'sapl.protocoloadm:documentoacessorioadministrativo_list',
kwargs={'pk': pk}))
if not data:
msg = _('Não há nenhum documento acessório PDF cadastrado.')
messages.add_message(request, messages.ERROR, msg)
return redirect(reverse('sapl.materia:documentoacessorio_list',
return redirect(reverse(
'sapl.protocoloadm:documentoacessorioadministrativo_list',
kwargs={'pk': pk}))
response = HttpResponse(data, content_type='application/pdf')
@ -401,8 +401,8 @@ class AcompanhamentoDocumentoView(CreateView):
class DocumentoAdministrativoMixin:
def has_permission(self):
app_config = AppConfig.objects.last()
if app_config and app_config.documentos_administrativos == 'O':
# mesmo acessor de pode_ver_documento_adm, usado nos arquivos
if AppConfig.attr('documentos_administrativos') == DOC_ADM_OSTENSIVO:
return True
return super().has_permission()
@ -498,7 +498,7 @@ class DocumentoAdministrativoCrud(Crud):
def get(self, *args, **kwargs):
pk = self.kwargs['pk']
documento = DocumentoAdministrativo.objects.get(id=pk)
documento = get_object_or_404(DocumentoAdministrativo, pk=pk)
if documento.restrito and self.request.user.is_anonymous:
return redirect('/')
return super(Crud.DetailView, self).get(args, kwargs)

94
sapl/sessao/tests/test_sessao_oculta_api.py

@ -0,0 +1,94 @@
from datetime import timedelta
from django.contrib.auth import get_user_model
from django.core.cache import cache
from django.utils import timezone
from model_bakery import baker
import pytest
from sapl.sessao.models import Orador, SessaoPlenaria
@pytest.fixture(autouse=True)
def limpa_cache_appconfig():
# o cache em arquivo sobrevive ao rollback do teste
cache.clear()
yield
cache.clear()
@pytest.fixture
def sessoes():
# só a sessão futura, não iniciada e sem pauta fica oculta do público
previa = baker.make(SessaoPlenaria, iniciada=False, publicar_pauta=False,
data_inicio=timezone.localdate() + timedelta(days=7))
iniciada = baker.make(SessaoPlenaria, iniciada=True, publicar_pauta=False)
baker.make(Orador, sessao_plenaria=previa)
baker.make(Orador, sessao_plenaria=iniciada)
return previa, iniciada
def _listados(client, endpoint, campo, sessoes):
"""Sessões de `sessoes` referenciadas na listagem completa do endpoint."""
valores, page = set(), 1
while page:
response = client.get('/api/sessao/%s/?page_size=100&page=%s' % (
endpoint, page))
assert response.status_code == 200
dados = response.json()
valores |= {r[campo] for r in dados['results']}
page = dados['pagination']['next_page']
return valores & {s.pk for s in sessoes}
@pytest.mark.django_db(transaction=False)
def test_api_oculta_sessao_previa_do_anonimo(client, sessoes):
previa, iniciada = sessoes
assert _listados(client, 'sessaoplenaria', 'id', sessoes) == {
iniciada.pk}
assert client.get('/api/sessao/sessaoplenaria/%s/' % previa.pk
).status_code == 404
assert client.get('/api/sessao/sessaoplenaria/%s/ecidadania/' % previa.pk
).status_code == 404
assert client.get('/api/sessao/sessaoplenaria/%s/expedientes/' % previa.pk
).status_code == 404
assert client.get('/api/sessao/sessaoplenaria/%s/expedientes/'
% iniciada.pk).status_code == 200
assert _listados(client, 'orador', 'sessao_plenaria', sessoes) == {
iniciada.pk}
@pytest.mark.django_db(transaction=False)
def test_api_mostra_sessao_previa_ao_usuario_autenticado(client, sessoes):
previa, iniciada = sessoes
client.force_login(baker.make(get_user_model()))
assert _listados(client, 'sessaoplenaria', 'id', sessoes) == {
previa.pk, iniciada.pk}
assert client.get('/api/sessao/sessaoplenaria/%s/' % previa.pk
).status_code == 200
assert _listados(client, 'orador', 'sessao_plenaria', sessoes) == {
previa.pk, iniciada.pk}
@pytest.mark.django_db(transaction=False)
def test_media_de_sessao_oculta_do_anonimo(client, settings, sessoes):
settings.DEBUG = False
previa, iniciada = sessoes
orador_previa = Orador.objects.get(sessao_plenaria=previa)
orador_iniciada = Orador.objects.get(sessao_plenaria=iniciada)
def get(path):
return client.get('/media/sapl/public/' + path).status_code
assert get('sessaoplenaria/%s/pauta/p.pdf' % previa.pk) == 404
assert get('sessaoplenaria/%s/pauta/p.pdf' % iniciada.pk) == 200
assert get('orador/%s/anexo/a.pdf' % orador_previa.pk) == 404
assert get('orador/%s/anexo/a.pdf' % orador_iniciada.pk) == 200
client.force_login(baker.make(get_user_model()))
assert get('sessaoplenaria/%s/pauta/p.pdf' % previa.pk) == 200
assert get('orador/%s/anexo/a.pdf' % orador_previa.pk) == 200

94
sapl/test_utils.py

@ -1,4 +1,9 @@
from .utils import listify
from django.core.exceptions import ValidationError
import pytest
from .utils import (listify, rejeita_conteudo_ativo,
restringe_tipos_de_arquivo_txt, tipo_ativo,
validar_arquivo)
def test_listify():
@ -8,3 +13,90 @@ def test_listify():
yield 1
yield 2
assert [1, 2] == gen()
def _upload(nome, conteudo):
import io
from django.core.files.uploadedfile import InMemoryUploadedFile
return InMemoryUploadedFile(io.BytesIO(conteudo), 'arquivo', nome, None,
len(conteudo), None)
def _odt():
import io
import zipfile
buf = io.BytesIO()
with zipfile.ZipFile(buf, 'w') as z:
z.writestr(zipfile.ZipInfo('mimetype'),
'application/vnd.oasis.opendocument.text')
z.writestr('content.xml', '<office:document-content/>')
return buf.getvalue()
PNG = (b'\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01'
b'\x08\x06\x00\x00\x00\x1f\x15\xc4\x89')
ATIVOS = [
('x.html', b'%PDF-1.4\n<html><script>alert(1)</script></html>'),
('x.html', b'<html><script>alert(1)</script></html>'),
('x.svg', b'<svg xmlns="http://www.w3.org/2000/svg"></svg>'),
('x.XHTML', b'%PDF-1.4'),
('x.xht', b'%PDF-1.4\n<html><script>alert(1)</script></html>'),
('x.rdf', b'%PDF-1.4'),
('x.kml', b'<kml><script>alert(1)</script></kml>'),
('x.pdf', b'<!DOCTYPE html><html><script>alert(1)</script></html>'),
]
@pytest.mark.parametrize('nome,conteudo', ATIVOS)
def test_validar_arquivo_rejeita_conteudo_ativo(nome, conteudo):
with pytest.raises(ValidationError):
validar_arquivo(_upload(nome, conteudo), 'Arquivo')
@pytest.mark.parametrize('nome,conteudo', ATIVOS)
def test_restringe_tipos_rejeita_conteudo_ativo(nome, conteudo):
with pytest.raises(ValidationError):
restringe_tipos_de_arquivo_txt(_upload(nome, conteudo))
@pytest.mark.parametrize('nome,conteudo', [
('a.pdf', b'%PDF-1.4\n%%EOF'),
('a.csv', b'nome;valor\nx;1\n'),
('a.odt', _odt()),
('a.png', PNG),
('a.xlsx', b'PK\x03\x04' + b'\x00' * 64),
])
def test_validar_arquivo_aceita_tipos_comuns(nome, conteudo):
validar_arquivo(_upload(nome, conteudo), 'Arquivo')
@pytest.mark.parametrize('nome,conteudo', [
('a.pdf', b'%PDF-1.4\n%%EOF'),
('a.csv', b'nome;valor\nx;1\n'),
('a.odt', _odt()),
])
def test_restringe_tipos_aceita_documentos(nome, conteudo):
restringe_tipos_de_arquivo_txt(_upload(nome, conteudo))
def test_validar_arquivo_ignora_arquivo_ja_gravado():
from django.db.models.fields.files import FieldFile
from sapl.protocoloadm.models import DocumentoAdministrativo
campo = DocumentoAdministrativo._meta.get_field('texto_integral')
antigo = FieldFile(None, campo,
'sapl/private/documentoadministrativo/x.html')
rejeita_conteudo_ativo(antigo)
@pytest.mark.parametrize('mime,ativo', [
('text/html', True), ('application/xhtml+xml', True),
('application/rdf+xml', True), ('text/xml', True),
('application/javascript', True), ('image/svg+xml', True),
('application/pdf', False), ('image/png', False), ('text/csv', False),
('application/vnd.openxmlformats-officedocument.wordprocessingml.document',
False),
(None, False),
])
def test_tipo_ativo(mime, ativo):
assert tipo_ativo(mime) is ativo

10
sapl/urls.py

@ -19,7 +19,6 @@ from django.conf.urls.static import static
from django.contrib import admin
from django.urls import path
from django.views.generic.base import RedirectView, TemplateView
from django.views.static import serve as view_static_server
import sapl.api.urls
import sapl.audiencia.urls
@ -36,11 +35,14 @@ import sapl.redireciona_urls.urls
import sapl.relatorios.urls
import sapl.sessao.urls
from sapl.base.media import media
from sapl.api.views_health import AppzVersionView, HealthzView, ReadyzView
urlpatterns = []
urlpatterns += [
url(r'^media/(?P<path>.+)$', media, name='media'),
url(r'^message$', TemplateView.as_view(template_name='base.html')),
url(r'^admin/', admin.site.urls),
@ -95,12 +97,6 @@ if settings.DEBUG:
urlpatterns += static(settings.STATIC_URL,
document_root=settings.STATIC_ROOT)
urlpatterns += [
url(r'^media/(?P<path>.*)$', view_static_server, {
'document_root': settings.MEDIA_ROOT,
}),
]
# Make the rate limiter return 429 (Too Many Requests) instead of 403 (Forbidden Access)
def custom_permission_denied_view(request, exception=None):

44
sapl/utils.py

@ -41,6 +41,7 @@ from django.utils.translation import ugettext_lazy as _
import django_filters
from easy_thumbnails import source_generators
import magic
import mimetypes
import requests
from unipath.path import Path
from xlsxwriter.workbook import Workbook
@ -133,6 +134,7 @@ def num_materias_por_tipo(qs, attr_tipo='tipo'):
def validar_arquivo(arquivo, nome_campo):
rejeita_conteudo_ativo(arquivo)
if len(arquivo.name) > 200:
raise ValidationError(
"Certifique-se de que o nome do arquivo no "
@ -601,6 +603,45 @@ class FilterOverridesMetaMixin:
}
# Extensões que o navegador pode executar como página na origem do SAPL.
# Não depender só de mimetypes: sem /etc/mime.types (imagem slim) várias
# destas não são reconhecidas. Manter em sincronia com sapl.conf.
EXTENSOES_ATIVAS = {
'.html', '.htm', '.xhtml', '.xht', '.shtml', '.svg', '.svgz', '.xml',
'.xsl', '.xslt', '.rdf', '.js', '.mjs', '.mht', '.mhtml', '.rss', '.atom',
'.mml', '.kml', '.xspf',
}
MIMES_ATIVOS = {
'text/html', 'application/xhtml+xml', 'image/svg+xml', 'text/xml',
'application/xml', 'application/javascript', 'text/javascript',
}
def tipo_ativo(mime):
"""MIME que o navegador renderiza como documento capaz de rodar script."""
if not mime:
return False
mime = mime.lower()
return (mime in MIMES_ATIVOS or 'html' in mime or 'javascript' in mime
or mime.endswith('/xml') or mime.endswith('+xml'))
def rejeita_conteudo_ativo(arquivo):
"""Recusa uploads que o navegador executaria como página, pela extensão
ou pelo conteúdo. Arquivos já gravados não são checados, para não impedir
a edição de registros antigos."""
if not isinstance(arquivo, UploadedFile):
return
if os.path.splitext(arquivo.name)[1].lower() in EXTENSOES_ATIVAS or \
tipo_ativo(mimetypes.guess_type(arquivo.name)[0]):
raise ValidationError(_('Tipo de arquivo não permitido.'))
arquivo.seek(0)
mime = magic.from_buffer(arquivo.read(2048), mime=True)
arquivo.seek(0)
if tipo_ativo(mime):
raise ValidationError(_('Tipo de arquivo não permitido.'))
TIPOS_TEXTO_PERMITIDOS = (
'application/vnd.oasis.opendocument.text',
'application/x-vnd.oasis.opendocument.text',
@ -621,8 +662,6 @@ TIPOS_TEXTO_PERMITIDOS = (
'widetext/paragraph',
'application/msword',
'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
'application/xml',
'text/xml',
# Formatos editáveis de planilha/CSV (atende à gravação de relatórios do PNTP).
'text/csv',
'application/csv',
@ -657,6 +696,7 @@ TIPOS_IMG_PERMITIDOS = (
def fabrica_validador_de_tipos_de_arquivo(lista, nome):
def restringe_tipos_de_arquivo(value):
rejeita_conteudo_ativo(value)
filename = value.name if type(value) in (
InMemoryUploadedFile, TemporaryUploadedFile

Loading…
Cancel
Save