diff --git a/docker/Dockerfile b/docker/Dockerfile index 831627ec8..a1dc565f4 100644 --- a/docker/Dockerfile +++ b/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; \ diff --git a/docker/config/nginx/sapl.conf b/docker/config/nginx/sapl.conf index 015538c96..ce5171a00 100644 --- a/docker/config/nginx/sapl.conf +++ b/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/ { - alias /var/interlegis/sapl/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 / { diff --git a/sapl/api/views_sessao.py b/sapl/api/views_sessao.py index 75807f8cd..29c31bf29 100644 --- a/sapl/api/views_sessao.py +++ b/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,), {})) diff --git a/sapl/base/media.py b/sapl/base/media.py new file mode 100644 index 000000000..9b91027c2 --- /dev/null +++ b/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///..., 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 diff --git a/sapl/base/tests/test_media.py b/sapl/base/tests/test_media.py new file mode 100644 index 000000000..de4ac7c94 --- /dev/null +++ b/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 diff --git a/sapl/materia/models.py b/sapl/materia/models.py index bdeb402e3..1c61ba0db 100644 --- a/sapl/materia/models.py +++ b/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')), diff --git a/sapl/materia/tests/test_acessorio_em_lote.py b/sapl/materia/tests/test_acessorio_em_lote.py new file mode 100644 index 000000000..a56e8bcc6 --- /dev/null +++ b/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') + + 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) == [] diff --git a/sapl/materia/tests/test_proposicao_confirmar.py b/sapl/materia/tests/test_proposicao_confirmar.py new file mode 100644 index 000000000..327b65d84 --- /dev/null +++ b/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 diff --git a/sapl/materia/views.py b/sapl/materia/views.py index 698ad6d33..4ac8a2432 100644 --- a/sapl/materia/views.py +++ b/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,44 +2347,57 @@ 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) - for materia_id in marcadas: - doc = DocumentoAcessorio() - doc.materia_id = materia_id - doc.tipo = tipo - doc.nome = request.POST['nome'] - doc.data = doc_data - doc.autor = request.POST['autor'] - doc.ementa = request.POST['ementa'] - doc.arquivo.name = tmp_name - try: - doc.clean_fields() - except ValidationError as e: - for m in ['%s: %s' % (DocumentoAcessorio()._meta.get_field(k).verbose_name, '
'.join(v)) - for k, v in e.message_dict.items()]: - # Insere as mensagens de erro no formato: - # '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) - return self.get(request, self.kwargs) - doc.save() - diretorio = os.path.join(MEDIA_ROOT, - 'sapl/public/documentoacessorio', - str(doc_data.year), - 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) - doc.arquivo.name = file_path.split( - MEDIA_ROOT + "/")[1] # Retira MEDIA_ROOT do nome - doc.save() - os.remove(tmp_name) + 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 + doc.tipo = tipo + doc.nome = request.POST['nome'] + doc.data = doc_data + doc.autor = request.POST['autor'] + doc.ementa = request.POST['ementa'] + try: + doc.clean_fields(exclude=['arquivo']) + except ValidationError as e: + for m in ['%s: %s' % (DocumentoAcessorio()._meta.get_field(k).verbose_name, '
'.join(v)) + for k, v in e.message_dict.items()]: + # Insere as mensagens de erro no formato: + # '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(m), arquivo.name)) + return self.get(request, self.kwargs) + doc.save() + diretorio = os.path.join(MEDIA_ROOT, + 'sapl/public/documentoacessorio', + str(doc_data.year), + str(doc.id)) + if not os.path.exists(diretorio): + os.makedirs(diretorio) + 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() + finally: + os.remove(tmp.name) msg = _('Documento(s) criado(s).') messages.add_message(request, messages.SUCCESS, msg) diff --git a/sapl/middleware/check_password.py b/sapl/middleware/check_password.py index 329e24909..90bf40635 100644 --- a/sapl/middleware/check_password.py +++ b/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') diff --git a/sapl/protocoloadm/models.py b/sapl/protocoloadm/models.py index 1a169733a..29fd29644 100644 --- a/sapl/protocoloadm/models.py +++ b/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) diff --git a/sapl/protocoloadm/tests/test_docadm_acesso.py b/sapl/protocoloadm/tests/test_docadm_acesso.py new file mode 100644 index 000000000..5f5bf9d0a --- /dev/null +++ b/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 diff --git a/sapl/protocoloadm/views.py b/sapl/protocoloadm/views.py index 345382438..575495f4e 100755 --- a/sapl/protocoloadm/views.py +++ b/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,21 +134,24 @@ 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', - kwargs={'pk': pk})) + 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', - kwargs={'pk': pk})) + 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', - kwargs={'pk': pk})) + return redirect(reverse( + 'sapl.protocoloadm:documentoacessorioadministrativo_list', + kwargs={'pk': pk})) response = HttpResponse(data, content_type='application/pdf') response['Content-Disposition'] = ('attachment; filename="%s"' @@ -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) diff --git a/sapl/sessao/tests/test_sessao_oculta_api.py b/sapl/sessao/tests/test_sessao_oculta_api.py new file mode 100644 index 000000000..efd66806e --- /dev/null +++ b/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 diff --git a/sapl/test_utils.py b/sapl/test_utils.py index 925b0ce8a..1bb724056 100644 --- a/sapl/test_utils.py +++ b/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', '') + 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'), + ('x.html', b''), + ('x.svg', b''), + ('x.XHTML', b'%PDF-1.4'), + ('x.xht', b'%PDF-1.4\n'), + ('x.rdf', b'%PDF-1.4'), + ('x.kml', b''), + ('x.pdf', b''), +] + + +@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 diff --git a/sapl/urls.py b/sapl/urls.py index d2967e927..7c1acc3e5 100644 --- a/sapl/urls.py +++ b/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.+)$', 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.*)$', 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): diff --git a/sapl/utils.py b/sapl/utils.py index 588494bd5..532594801 100644 --- a/sapl/utils.py +++ b/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