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>

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,), {}))

sapl/base/media.py

@@ -10,6 +10,7 @@ from sapl.materia.models import Proposicao
 from sapl.protocoloadm.models import (DocumentoAcessorioAdministrativo,
                                       DocumentoAdministrativo,
                                       pode_ver_documento_adm)
+from sapl.sessao.models import SessaoPlenaria, restringe_sessoes_visiveis
 from sapl.utils import EXTENSOES_ATIVAS
 
 X_ACCEL_PREFIX = '/media-interna/'
@@ -61,6 +62,14 @@ def _nome_protegido(request, path):
     return arquivo.name[2:] if arquivo.name.startswith('./') else arquivo.name
 
 
+def _sessao_oculta(request, partes):
+    if request.user.is_authenticated or not partes[3].isdigit():
+        return False
+    qs = SessaoPlenaria.objects.filter(pk=partes[3])
+    return qs.exists() and not restringe_sessoes_visiveis(
+        qs, request.user).exists()
+
+
 def media(request, path):
     partes = path.split('/')
     if any(p in ('', '.', '..') for p in partes) or '\x00' in path:
@@ -68,6 +77,10 @@ def media(request, path):
 
     protegido = bool(PROTEGIDO_RE.match(path))
     nome = _nome_protegido(request, path) if protegido else path
+
+    if path.startswith('sapl/public/sessaoplenaria/') and len(partes) > 4 \
+            and _sessao_oculta(request, partes):
+        raise Http404
     nome_arquivo = os.path.basename(nome)
 
     if os.path.splitext(nome_arquivo)[1].lower() in EXTENSOES_ATIVAS:

sapl/base/tests/test_media.py

@@ -19,6 +19,14 @@ 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

sapl/protocoloadm/tests/test_docadm_acesso.py

@@ -11,6 +11,14 @@ from sapl.base.models import AppConfig
 from sapl.protocoloadm.models import 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)

sapl/sessao/tests/test_sessao_oculta_api.py

@@ -0,0 +1,85 @@
+from django.contrib.auth import get_user_model
+from django.core.cache import cache
+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():
+    previa = baker.make(SessaoPlenaria, iniciada=False, publicar_pauta=False)
+    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_previa_oculta_do_anonimo(client, settings, sessoes):
+    settings.DEBUG = False
+    previa, iniciada = sessoes
+
+    def get(sessao):
+        return client.get('/media/sapl/public/sessaoplenaria/%s/pauta/p.pdf'
+                          % sessao.pk)
+
+    assert get(previa).status_code == 404
+    assert get(iniciada).status_code == 200
+
+    client.force_login(baker.make(get_user_model()))
+    assert get(previa).status_code == 200