Hide Chainlit Sources when no evidence is available

openrag/app_front.py

@@ -1,10 +1,11 @@
 import json
 import os
 import secrets
+import string
 import time
 from functools import lru_cache
 from pathlib import Path
-from urllib.parse import urlparse
+from urllib.parse import quote, urlparse
 
 import chainlit as cl
 import httpx
@@ -42,12 +43,19 @@
 OPENRAG_CHAT_PROFILES_METADATA_KEY = "openrag_chat_profiles"
 OPENRAG_SESSION_COOKIE_NAME = "openrag_session"
 _OPENRAG_TOKEN_STORE: dict[str, tuple[str, float]] = {}
+_MARKDOWN_ESCAPE_TABLE = str.maketrans({char: f"\\{char}" for char in string.punctuation})
+_MARKDOWN_URL_SAFE_CHARS = ":/?#[]@!$&'+,;=%"
 
 
 class MissingOpenRAGCredentialError(RuntimeError):
     pass
 
 
+def _escape_markdown_text(value: str) -> str:
+    """Render untrusted source metadata as literal Markdown text."""
+    return value.translate(_MARKDOWN_ESCAPE_TABLE)
+
+
 def get_user_language() -> str:
     """Return the active language: env override if set, otherwise browser's Accept-Language."""
     if DEFAULT_LANGUAGE:
@@ -465,26 +473,56 @@ async def __fetch_page_content(chunk_url, headers=None):
 
 
 async def _format_sources(metadata_sources, only_txt=False, api_key=None):
-    external_url = get_external_url()  # used to override the base URL when the front-end requests a file resource
     if not metadata_sources:
-        return None, None
+        return [], []
 
     d = {}
     headers = get_headers(api_key)
+    external_url = get_external_url()  # used to override the base URL when the front-end requests a file resource
     for i, s in enumerate(metadata_sources):
+        if not isinstance(s, dict):
+            continue
+
         if s.get("source_type") == "web":
-            title = s.get("title") or s.get("url", f"Web source {i + 1}")
+            title = s.get("title", "")
             url = s.get("url", "")
             snippet = s.get("snippet", "")
-            content = f"**[{title}]({url})**\n\n{snippet}"
-            source_name = title
+            title = title.strip() if isinstance(title, str) else ""
+            url = url.strip() if isinstance(url, str) else ""
+            snippet = snippet.strip() if isinstance(snippet, str) else ""
+            parsed_url = urlparse(url)
+            if parsed_url.scheme not in {"http", "https"} or not parsed_url.netloc:
+                continue
+            try:
+                markdown_url = quote(str(httpx.URL(url)), safe=_MARKDOWN_URL_SAFE_CHARS)
+            except httpx.InvalidURL:
+                continue
+
+            source_label = title or url
+            source_name = source_label
             if source_name in d:
-                source_name = f"{title} ({i})"
+                source_name = f"{source_name} ({i})"
+            content = f"**[{_escape_markdown_text(source_label)}]({markdown_url})**"
+            if snippet:
+                content += f"\n\n{_escape_markdown_text(snippet)}"
             d[source_name] = cl.Text(content=content, name=source_name, display="side")
             continue
 
-        filename = Path(s["filename"])
-        file_url = s["file_url"]
+        filename_value = s.get("filename")
+        file_url = s.get("file_url")
+        page = s.get("page")
+        if (
+            not isinstance(filename_value, str)
+            or not filename_value.strip()
+            or not isinstance(file_url, str)
+            or not file_url.strip()
+            or page is None
+            or not str(page).strip()
+        ):
+            continue
+
+        filename = Path(filename_value.strip())
+        file_url = file_url.strip()
         file_url = file_url.replace(INTERNAL_BASE_URL, external_url)  # put the correct base url
         # Avoid leaking the credential in the URL (browser history, proxy logs,
         # Referer headers). In OIDC mode the browser already sends the
@@ -495,36 +533,49 @@ async def _format_sources(metadata_sources, only_txt=False, api_key=None):
         # authenticate the fetch.
         if api_key and (AUTH_MODE != "oidc" or _current_openrag_auth_provider() == "credentials"):
             file_url = f"{file_url}?token={api_key}"
-        page = s["page"]
         source_name = f"{filename}" + (
             f" (page: {page})" if filename.suffix in [".pdf", ".pptx", ".docx", ".doc"] else ""
         )
 
-        if only_txt:
-            chunk_content = await __fetch_page_content(chunk_url=s["chunk_url"], headers=headers)
-            elem = cl.Text(content=chunk_content, name=source_name, display="side")
-        else:
-            match filename.suffix.lower():
-                case ".pdf":
-                    elem = cl.Pdf(
-                        name=source_name,
-                        url=file_url,
-                        page=int(s["page"]),
-                        display="side",
-                    )
-                case suffix if suffix in [".png", ".jpg", ".jpeg"]:
-                    elem = cl.Image(name=source_name, url=file_url, display="side")
-                case ".mp4":
-                    elem = cl.Video(name=source_name, url=file_url, display="side")
-                case ".mp3":
-                    elem = cl.Audio(name=source_name, url=file_url, display="side")
-                case _:
-                    chunk_content = await __fetch_page_content(chunk_url=s["chunk_url"], headers=headers)
-                    elem = cl.Text(content=chunk_content, name=source_name, display="side")
+        try:
+            if only_txt:
+                chunk_url = s.get("chunk_url")
+                if not isinstance(chunk_url, str) or not chunk_url.strip():
+                    continue
+                chunk_content = await __fetch_page_content(chunk_url=chunk_url, headers=headers)
+                if not isinstance(chunk_content, str) or not chunk_content.strip():
+                    continue
+                elem = cl.Text(content=chunk_content, name=source_name, display="side")
+            else:
+                match filename.suffix.lower():
+                    case ".pdf":
+                        elem = cl.Pdf(
+                            name=source_name,
+                            url=file_url,
+                            page=int(page),
+                            display="side",
+                        )
+                    case suffix if suffix in [".png", ".jpg", ".jpeg"]:
+                        elem = cl.Image(name=source_name, url=file_url, display="side")
+                    case ".mp4":
+                        elem = cl.Video(name=source_name, url=file_url, display="side")
+                    case ".mp3":
+                        elem = cl.Audio(name=source_name, url=file_url, display="side")
+                    case _:
+                        chunk_url = s.get("chunk_url")
+                        if not isinstance(chunk_url, str) or not chunk_url.strip():
+                            continue
+                        chunk_content = await __fetch_page_content(chunk_url=chunk_url, headers=headers)
+                        if not isinstance(chunk_content, str) or not chunk_content.strip():
+                            continue
+                        elem = cl.Text(content=chunk_content, name=source_name, display="side")
+        except (httpx.HTTPError, TypeError, ValueError, AttributeError):
+            logger.warning("Skipping an unavailable source", source_index=i)
+            continue
 
         d[source_name] = elem
 
-    source_names = list(d.keys())
+    source_names = [_escape_markdown_text(name) for name in d]
     elements = list(d.values())
 
     return elements, source_names
@@ -582,7 +633,7 @@ async def on_message(message: cl.Message):
             # Show sources
             elements, source_names = await _format_sources(sources, api_key=api_key, only_txt=False)
             msg.elements = elements if elements else []
-            if source_names:
+            if elements and source_names:
                 s = "\n\n" + "-" * 50 + f"\n\n{t('sources_label')}: \n" + "\n".join(source_names)
                 await msg.stream_token(s)
                 await msg.update()

tests/unit/test_app_front_secret.py

@@ -29,6 +29,16 @@ def _load_app_front(monkeypatch, *, auth_mode: str, module_name: str):
     return module
 
 
+def _stub_chainlit_elements(module):
+    module.cl = SimpleNamespace(
+        Pdf=lambda **kwargs: SimpleNamespace(**kwargs),
+        Text=lambda **kwargs: SimpleNamespace(**kwargs),
+        Image=lambda **kwargs: SimpleNamespace(**kwargs),
+        Video=lambda **kwargs: SimpleNamespace(**kwargs),
+        Audio=lambda **kwargs: SimpleNamespace(**kwargs),
+    )
+
+
 def test_no_hardcoded_default_secret_assignment_in_source():
     """The fall-through to a literal default secret must be gone.
 
@@ -398,6 +408,146 @@ async def fake_load_model_ids(_client, api_key):
     assert module._OPENRAG_TOKEN_STORE[auth_handle][0] == "handoff-token"
 
 
+@pytest.mark.parametrize(
+    "sources",
+    [
+        None,
+        [],
+        [{}],
+        [None],
+        [{"source_type": "web", "title": "", "url": "", "snippet": ""}],
+        [{"filename": "", "file_url": "", "page": ""}],
+    ],
+)
+@pytest.mark.asyncio
+async def test_chainlit_hides_sources_when_none_are_displayable(monkeypatch, sources):
+    module = _load_app_front(monkeypatch, auth_mode="token", module_name="app_front_empty_sources_test")
+    monkeypatch.setattr(module, "get_external_url", lambda: "https://openrag.example")
+    _stub_chainlit_elements(module)
+
+    elements, source_names = await module._format_sources(sources)
+
+    assert elements == []
+    assert source_names == []
+
+
+@pytest.mark.asyncio
+async def test_chainlit_keeps_valid_web_sources(monkeypatch):
+    module = _load_app_front(monkeypatch, auth_mode="token", module_name="app_front_web_source_test")
+    monkeypatch.setattr(module, "get_external_url", lambda: "https://openrag.example")
+    _stub_chainlit_elements(module)
+
+    elements, source_names = await module._format_sources(
+        [
+            {
+                "source_type": "web",
+                "title": "Example reference",
+                "url": "https://example.test/reference",
+                "snippet": "Supporting evidence",
+            }
+        ]
+    )
+
+    assert source_names == ["Example reference"]
+    assert elements[0].name == "Example reference"
+    assert elements[0].content == ("**[Example reference](https://example.test/reference)**\n\nSupporting evidence")
+
+
+@pytest.mark.asyncio
+async def test_chainlit_escapes_untrusted_web_source_markdown(monkeypatch):
+    module = _load_app_front(monkeypatch, auth_mode="token", module_name="app_front_web_source_escaping_test")
+    monkeypatch.setattr(module, "get_external_url", lambda: "https://openrag.example")
+    _stub_chainlit_elements(module)
+
+    title = "Reference ](https://spoof.test) **trusted**"
+    snippet = "Evidence [click here](https://spoof.test) or *ignore this*."
+    url = "https://example.test/reference_(draft)"
+    elements, source_names = await module._format_sources(
+        [
+            {
+                "source_type": "web",
+                "title": title,
+                "url": url,
+                "snippet": snippet,
+            }
+        ]
+    )
+
+    assert source_names == [r"Reference \]\(https\:\/\/spoof\.test\) \*\*trusted\*\*"]
+    assert elements[0].name == title
+    assert elements[0].content == (
+        r"**[Reference \]\(https\:\/\/spoof\.test\) \*\*trusted\*\*]"
+        "(https://example.test/reference_%28draft%29)**\n\n"
+        r"Evidence \[click here\]\(https\:\/\/spoof\.test\) or \*ignore this\*\."
+    )
+
+
+@pytest.mark.asyncio
+async def test_chainlit_skips_unavailable_text_sources(monkeypatch):
+    module = _load_app_front(monkeypatch, auth_mode="token", module_name="app_front_unavailable_source_test")
+    monkeypatch.setattr(module, "get_external_url", lambda: "https://openrag.example")
+    _stub_chainlit_elements(module)
+
+    async def unavailable_chunk(*_args, **_kwargs):
+        raise httpx.ConnectError("source unavailable")
+
+    monkeypatch.setattr(module, "__fetch_page_content", unavailable_chunk)
+
+    elements, source_names = await module._format_sources(
+        [
+            {
+                "filename": "notes.txt",
+                "file_url": "http://internal:8080/static/source-id",
+                "page": "1",
+                "chunk_url": "http://internal:8080/chunks/source-id",
+            }
+        ]
+    )
+
+    assert elements == []
+    assert source_names == []
+
+
+@pytest.mark.asyncio
+async def test_chainlit_skips_text_source_with_non_object_json(monkeypatch):
+    module = _load_app_front(monkeypatch, auth_mode="token", module_name="app_front_malformed_source_test")
+    monkeypatch.setattr(module, "get_external_url", lambda: "https://openrag.example")
+    _stub_chainlit_elements(module)
+
+    class FakeResponse:
+        def raise_for_status(self):
+            return None
+
+        def json(self):
+            return [{"page_content": "unexpected list response"}]
+
+    class FakeAsyncClient:
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *_args):
+            return None
+
+        async def get(self, *_args, **_kwargs):
+            return FakeResponse()
+
+    monkeypatch.setattr(module.httpx, "AsyncClient", FakeAsyncClient)
+
+    elements, source_names = await module._format_sources(
+        [
+            {
+                "filename": "notes.txt",
+                "file_url": "http://internal:8080/static/source-id",
+                "page": "1",
+                "chunk_url": "http://internal:8080/chunks/source-id",
+            }
+        ]
+    )
+
+    assert elements == []
+    assert source_names == []
+
+
 @pytest.mark.asyncio
 async def test_oidc_token_handoff_keeps_bearer_on_static_source_urls(monkeypatch):
     module = _load_app_front(monkeypatch, auth_mode="oidc", module_name="app_front_source_token_test")
@@ -410,14 +560,8 @@ def get(self, key):
                 return SimpleNamespace(metadata={"provider": "credentials"})
             return None
 
-    module.cl = SimpleNamespace(
-        user_session=UserSession(),
-        Pdf=lambda **kwargs: SimpleNamespace(**kwargs),
-        Text=lambda **kwargs: SimpleNamespace(**kwargs),
-        Image=lambda **kwargs: SimpleNamespace(**kwargs),
-        Video=lambda **kwargs: SimpleNamespace(**kwargs),
-        Audio=lambda **kwargs: SimpleNamespace(**kwargs),
-    )
+    _stub_chainlit_elements(module)
+    module.cl.user_session = UserSession()
 
     elements, _ = await module._format_sources(
         [
@@ -445,14 +589,8 @@ def get(self, key):
                 return SimpleNamespace(metadata={"provider": "oidc"})
             return None
 
-    module.cl = SimpleNamespace(
-        user_session=UserSession(),
-        Pdf=lambda **kwargs: SimpleNamespace(**kwargs),
-        Text=lambda **kwargs: SimpleNamespace(**kwargs),
-        Image=lambda **kwargs: SimpleNamespace(**kwargs),
-        Video=lambda **kwargs: SimpleNamespace(**kwargs),
-        Audio=lambda **kwargs: SimpleNamespace(**kwargs),
-    )
+    _stub_chainlit_elements(module)
+    module.cl.user_session = UserSession()
 
     elements, _ = await module._format_sources(
         [