Browse Source

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

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
pull/3865/head
Edward Ribeiro 3 weeks ago
parent
commit
e62a87e9d4
  1. 20
      sapl/base/media.py
  2. 28
      sapl/base/tests/test_media.py
  3. 22
      sapl/materia/views.py

20
sapl/base/media.py

@ -73,6 +73,14 @@ def _nome_protegido(request, path):
return arquivo.name[2:] if arquivo.name.startswith('./') else arquivo.name 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): def media(request, path):
partes = path.split('/') partes = path.split('/')
if any(p in ('', '.', '..') for p in partes) or '\x00' in path: if any(p in ('', '.', '..') for p in partes) or '\x00' in path:
@ -93,16 +101,12 @@ def media(request, path):
tipo_ativo(content_type): tipo_ativo(content_type):
content_type, disposicao = 'application/octet-stream', 'attachment' content_type, disposicao = 'application/octet-stream', 'attachment'
caminho = _caminho_em_media(nome)
if settings.DEBUG: if settings.DEBUG:
try: try:
media_root = os.path.realpath(settings.MEDIA_ROOT) response = FileResponse(open(caminho, 'rb'),
caminho = os.path.realpath(os.path.join(media_root, nome)) content_type=content_type)
if os.path.commonpath([media_root, caminho]) != media_root: except (FileNotFoundError, IsADirectoryError):
raise Http404
response = FileResponse(
open(caminho, 'rb'),
content_type=content_type)
except (FileNotFoundError, IsADirectoryError, ValueError):
raise Http404 raise Http404
else: else:
response = HttpResponse(content_type=content_type) response = HttpResponse(content_type=content_type)

28
sapl/base/tests/test_media.py

@ -276,6 +276,34 @@ def test_debug_serve_arquivo(rf, settings, tmp_path):
chama(path) 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) @pytest.mark.django_db(transaction=False)
def test_modelo_protegido_fora_de_private_e_public(client): def test_modelo_protegido_fora_de_private_e_public(client):
_config('O') _config('O')

22
sapl/materia/views.py

@ -16,7 +16,8 @@ from django.conf import settings
from django.contrib import messages from django.contrib import messages
from django.contrib.auth.decorators import permission_required from django.contrib.auth.decorators import permission_required
from django.contrib.auth.mixins import PermissionRequiredMixin 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 import transaction
from django.db.models import Max, Q from django.db.models import Max, Q
from django.http import HttpResponse, JsonResponse from django.http import HttpResponse, JsonResponse
@ -2348,6 +2349,12 @@ class DocumentoAcessorioEmLoteView(PermissionRequiredMixin, FilterView):
username, str(msg), request.POST['data'])) username, str(msg), request.POST['data']))
return self.get(request, self.kwargs) 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/ # Fora de MEDIA_ROOT, para não ficar acessível em /media/
with tempfile.NamedTemporaryFile(dir=get_tempfile_dir(), with tempfile.NamedTemporaryFile(dir=get_tempfile_dir(),
delete=False) as tmp: delete=False) as tmp:
@ -2380,14 +2387,11 @@ class DocumentoAcessorioEmLoteView(PermissionRequiredMixin, FilterView):
str(doc.id)) str(doc.id))
if not os.path.exists(diretorio): if not os.path.exists(diretorio):
os.makedirs(diretorio) os.makedirs(diretorio)
safe_filename = os.path.basename(arquivo.name) diretorio = os.path.normpath(diretorio)
file_path = os.path.join(diretorio, safe_filename) file_path = os.path.normpath(
real_diretorio = os.path.realpath(diretorio) os.path.join(diretorio, nome_arquivo))
real_file_path = os.path.realpath(file_path) if not file_path.startswith(diretorio + os.sep):
if os.path.commonpath([real_diretorio, real_file_path]) != real_diretorio: raise SuspiciousFileOperation(nome_arquivo)
msg = _('Nome de arquivo inválido.')
messages.add_message(request, messages.ERROR, msg)
return self.get(request, self.kwargs)
shutil.copy2(tmp.name, file_path) shutil.copy2(tmp.name, file_path)
doc.arquivo.name = file_path.split( doc.arquivo.name = file_path.split(
MEDIA_ROOT + "/")[1] # Retira MEDIA_ROOT do nome MEDIA_ROOT + "/")[1] # Retira MEDIA_ROOT do nome

Loading…
Cancel
Save