diff --git a/docker/config/nginx/sapl.conf b/docker/config/nginx/sapl.conf index 015538c96..44b339173 100644 --- a/docker/config/nginx/sapl.conf +++ b/docker/config/nginx/sapl.conf @@ -49,8 +49,12 @@ server { alias /var/interlegis/sapl/collected_static/; } - location /media/ { - alias /var/interlegis/sapl/media/; + # /media/ não é servido direto do disco: cai em location / e o Django + # (sapl.base.media) autoriza e devolve X-Accel-Redirect para cá. + location /media-interna/ { + internal; + alias /var/interlegis/sapl/media/; + add_header X-Content-Type-Options nosniff always; } location / { diff --git a/sapl/base/media.py b/sapl/base/media.py new file mode 100644 index 000000000..394380c63 --- /dev/null +++ b/sapl/base/media.py @@ -0,0 +1,96 @@ +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 +from sapl.protocoloadm.models import (DocumentoAcessorioAdministrativo, + DocumentoAdministrativo, + pode_ver_documento_adm) +from sapl.utils import EXTENSOES_ATIVAS + +X_ACCEL_PREFIX = '/media-interna/' + +# Arquivos que dependem de autorização. Inclui as pastas public/ onde +# ficaram os arquivos enviados antes da criação de private/. +PROTEGIDO_RE = re.compile( + r'^sapl/(?:private|public/(?:documentoadministrativo|' + r'documentoacessorioadministrativo|proposicao))/') + + +def _pode_ver_proposicao(user, proposicao): + if proposicao.data_recebimento: + return True + if not user.is_authenticated: + return False + if proposicao.autor and proposicao.autor.operadores.filter( + id=user.id).exists(): + return True + return bool(proposicao.data_envio) and user.has_module_perms( + 'protocoloadm') + + +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), +} + + +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 media(request, path): + partes = path.split('/') + if any(p in ('', '.', '..') for p in partes) or '\x00' in path: + raise Http404 + + protegido = bool(PROTEGIDO_RE.match(path)) + nome = _nome_protegido(request, path) if protegido else path + nome_arquivo = os.path.basename(nome) + + if os.path.splitext(nome_arquivo)[1].lower() in EXTENSOES_ATIVAS: + content_type, disposicao = 'application/octet-stream', 'attachment' + else: + content_type = (mimetypes.guess_type(nome_arquivo)[0] or + 'application/octet-stream') + disposicao = 'inline' + + if settings.DEBUG: + try: + response = FileResponse( + open(os.path.join(settings.MEDIA_ROOT, nome), '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: + 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..397d2720f --- /dev/null +++ b/sapl/base/tests/test_media.py @@ -0,0 +1,247 @@ +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 media +from sapl.base.models import AppConfig, Autor +from sapl.materia.models import Proposicao +from sapl.protocoloadm.models import (DocumentoAcessorioAdministrativo, + DocumentoAdministrativo) + +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 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, perm=None): + user = baker.make(get_user_model(), username=username) + if perm: + 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, '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') + + +def _proposicao(enviada=False, recebida=False): + autor = baker.make(Autor) + operador = baker.make(get_user_model(), username='operador') + autor.operadores.add(operador) + agora = timezone.now() + baker.make(Proposicao, autor=autor, texto_original=PROPOSICAO_PATH, + data_envio=agora if enviada or recebida else None, + data_recebimento=agora if recebida else None) + return operador + + +@pytest.mark.django_db(transaction=False) +@pytest.mark.parametrize('estado,perfil,pode_ver', [ + ('rascunho', 'operador', True), + ('rascunho', 'outro', False), + ('rascunho', 'protocolo', False), + ('rascunho', None, False), + ('enviada', 'protocolo', True), + ('enviada', 'outro', False), + ('enviada', None, False), + ('recebida', None, True), +]) +def test_proposicao(client, estado, perfil, pode_ver): + operador = _proposicao(enviada=estado == 'enviada', + recebida=estado == 'recebida') + if perfil == 'operador': + client.force_login(operador) + elif perfil == 'outro': + _login(client, 'outro') + elif perfil == 'protocolo': + _login(client, 'protocolo', + 'protocoloadm.detail_documentoadministrativo') + + response = _get(client, PROPOSICAO_PATH) + + assert response.status_code == (200 if pode_ver else 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']) +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) 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 9bd6231dc..615fb2f11 100644 --- a/sapl/utils.py +++ b/sapl/utils.py @@ -601,6 +601,12 @@ class FilterOverridesMetaMixin: } +# Extensões que o navegador pode executar como página na origem do SAPL +EXTENSOES_ATIVAS = { + '.html', '.htm', '.xhtml', '.shtml', '.svg', '.svgz', '.xml', '.xsl', + '.js', '.mht', '.mhtml', +} + TIPOS_TEXTO_PERMITIDOS = ( 'application/vnd.oasis.opendocument.text', 'application/x-vnd.oasis.opendocument.text',