diff --git a/sapl/base/media.py b/sapl/base/media.py index 0b7435116..17f8ea518 100644 --- a/sapl/base/media.py +++ b/sapl/base/media.py @@ -73,6 +73,14 @@ def _nome_protegido(request, path): 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: @@ -93,16 +101,12 @@ def media(request, path): tipo_ativo(content_type): content_type, disposicao = 'application/octet-stream', 'attachment' + caminho = _caminho_em_media(nome) if settings.DEBUG: try: - media_root = os.path.realpath(settings.MEDIA_ROOT) - caminho = os.path.realpath(os.path.join(media_root, nome)) - if os.path.commonpath([media_root, caminho]) != media_root: - raise Http404 - response = FileResponse( - open(caminho, 'rb'), - content_type=content_type) - except (FileNotFoundError, IsADirectoryError, ValueError): + response = FileResponse(open(caminho, 'rb'), + content_type=content_type) + except (FileNotFoundError, IsADirectoryError): raise Http404 else: response = HttpResponse(content_type=content_type) diff --git a/sapl/base/tests/test_media.py b/sapl/base/tests/test_media.py index 63366442c..24ceb7156 100644 --- a/sapl/base/tests/test_media.py +++ b/sapl/base/tests/test_media.py @@ -276,6 +276,34 @@ def test_debug_serve_arquivo(rf, settings, tmp_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) def test_modelo_protegido_fora_de_private_e_public(client): _config('O') diff --git a/sapl/materia/views.py b/sapl/materia/views.py index 99f0b4a3c..16376279d 100644 --- a/sapl/materia/views.py +++ b/sapl/materia/views.py @@ -16,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 @@ -2348,6 +2349,12 @@ class DocumentoAcessorioEmLoteView(PermissionRequiredMixin, FilterView): username, str(msg), request.POST['data'])) 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: @@ -2380,14 +2387,11 @@ class DocumentoAcessorioEmLoteView(PermissionRequiredMixin, FilterView): str(doc.id)) if not os.path.exists(diretorio): os.makedirs(diretorio) - safe_filename = os.path.basename(arquivo.name) - file_path = os.path.join(diretorio, safe_filename) - real_diretorio = os.path.realpath(diretorio) - real_file_path = os.path.realpath(file_path) - if os.path.commonpath([real_diretorio, real_file_path]) != real_diretorio: - msg = _('Nome de arquivo inválido.') - messages.add_message(request, messages.ERROR, msg) - return self.get(request, self.kwargs) + 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.copy2(tmp.name, file_path) doc.arquivo.name = file_path.split( MEDIA_ROOT + "/")[1] # Retira MEDIA_ROOT do nome