Add OSS subscriber for Open Source Messaging banner

Fetches a markdown message from the DaaS-published GCS bucket, renders
the bleached headline and optional expanded section through the existing
additional_banners template loop. Cached for 1h; any fetch/parse failure
silently yields no banner. No Django settings introduced — disabling the
banner requires forking.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: Maffooch

dojo/announcements/__init__.py

@@ -0,0 +1,119 @@
+import logging
+
+import bleach
+import markdown
+import requests
+from django.core.cache import cache
+
+logger = logging.getLogger(__name__)
+
+BUCKET_URL = "https://storage.googleapis.com/defectdojo-os-messages-prod/open_source_message.md"
+CACHE_SECONDS = 3600
+HTTP_TIMEOUT_SECONDS = 2
+CACHE_KEY = "os_message:v1"
+
+INLINE_TAGS = ["strong", "em", "a"]
+INLINE_ATTRS = {"a": ["href", "title"]}
+
+# Keep BLOCK_TAGS / BLOCK_ATTRS in sync with the DaaS publisher's
+# MARKDOWNIFY["default"]["WHITELIST_TAGS"] / WHITELIST_ATTRS so previews
+# on DaaS and rendering in OSS stay byte-identical.
+BLOCK_TAGS = [
+    "p", "ul", "ol", "li", "a", "strong", "em", "code", "pre",
+    "blockquote", "h2", "h3", "h4", "hr", "br", "b", "i",
+    "abbr", "acronym",
+]
+BLOCK_ATTRS = {
+    "a": ["href", "title"],
+    "abbr": ["title"],
+    "acronym": ["title"],
+}
+
+_MISS = object()
+
+
+def fetch_os_message():
+    cached = cache.get(CACHE_KEY, default=_MISS)
+    if cached is not _MISS:
+        return cached
+
+    try:
+        response = requests.get(BUCKET_URL, timeout=HTTP_TIMEOUT_SECONDS)
+    except Exception:
+        logger.debug("os_message: fetch failed", exc_info=True)
+        cache.set(CACHE_KEY, None, CACHE_SECONDS)
+        return None
+
+    if response.status_code != 200 or not response.text.strip():
+        cache.set(CACHE_KEY, None, CACHE_SECONDS)
+        return None
+
+    cache.set(CACHE_KEY, response.text, CACHE_SECONDS)
+    return response.text
+
+
+def _strip_outer_p(html):
+    stripped = html.strip()
+    if stripped.startswith("<p>") and stripped.endswith("</p>"):
+        return stripped[3:-4]
+    return stripped
+
+
+def parse_os_message(text):
+    lines = text.splitlines()
+
+    headline_source = None
+    body_start = None
+    for index, line in enumerate(lines):
+        if line.startswith("# "):
+            headline_source = line[2:].strip()
+            body_start = index + 1
+            break
+
+    if not headline_source:
+        return None
+
+    headline_source = headline_source[:100]
+    headline_rendered = markdown.markdown(headline_source)
+    headline_cleaned = bleach.clean(
+        headline_rendered,
+        tags=INLINE_TAGS,
+        attributes=INLINE_ATTRS,
+        strip=True,
+    )
+    headline_html = _strip_outer_p(headline_cleaned)
+
+    expanded_html = None
+    expanded_marker = "## Expanded Message"
+    expanded_body_lines = None
+    for offset, line in enumerate(lines[body_start:], start=body_start):
+        if line.strip() == expanded_marker:
+            expanded_body_lines = lines[offset + 1:]
+            break
+
+    if expanded_body_lines is not None:
+        expanded_source = "\n".join(expanded_body_lines).strip()
+        if expanded_source:
+            expanded_rendered = markdown.markdown(
+                expanded_source,
+                extensions=["extra", "fenced_code"],
+            )
+            expanded_html = bleach.clean(
+                expanded_rendered,
+                tags=BLOCK_TAGS,
+                attributes=BLOCK_ATTRS,
+                strip=True,
+            )
+
+    return {"message": headline_html, "expanded_html": expanded_html}
+
+
+def get_os_banner():
+    try:
+        text = fetch_os_message()
+        if not text:
+            return None
+        return parse_os_message(text)
+    except Exception:
+        logger.debug("os_message: get_os_banner failed", exc_info=True)
+        return None

dojo/announcements/os_message.py

@@ -5,13 +5,14 @@
 from django.conf import settings
 from django.contrib import messages
 
+from dojo.announcements.os_message import get_os_banner
 from dojo.labels import get_labels
 from dojo.models import Alerts, System_Settings, UserAnnouncement
 
 
 def globalize_vars(request):
     # return the value you want as a dictionnary. you may add multiple values in there.
-    return {
+    context = {
         "SHOW_LOGIN_FORM": settings.SHOW_LOGIN_FORM,
         "FORGOT_PASSWORD": settings.FORGOT_PASSWORD,
         "FORGOT_USERNAME": settings.FORGOT_USERNAME,
@@ -39,6 +40,16 @@ def globalize_vars(request):
         # V3 Feature Flags
         "V3_FEATURE_LOCATIONS": settings.V3_FEATURE_LOCATIONS,
     }
+    os_banner = get_os_banner()
+    if os_banner is not None:
+        context["additional_banners"] = [{
+            "message": os_banner["message"],
+            "style": "info",
+            "url": "",
+            "link_text": "",
+            "expanded_html": os_banner["expanded_html"],
+        }]
+    return context
 
 
 def bind_system_settings(request):

dojo/context_processors.py

@@ -1126,6 +1126,25 @@ div.custom-search-form {
     border-radius: 0px 0px 4px 4px;
 }
 
+.os-banner-toggle {
+    margin-left: 6px;
+    cursor: pointer;
+    color: inherit;
+    text-decoration: none;
+}
+
+.os-banner-toggle .fa-caret-down {
+    transition: transform 0.15s ease-in-out;
+}
+
+.os-banner-toggle:not(.collapsed) .fa-caret-down {
+    transform: rotate(180deg);
+}
+
+.os-banner-expanded {
+    margin-top: 8px;
+}
+
 @media (min-width: 795px) {
     div.custom-search-form {
         height: 21px;

dojo/static/dojo/css/dojo.css

@@ -672,7 +672,17 @@
                 {% endif %}
                 {% for banner in additional_banners %}
                     <div role="alert" class="announcement-banner alert alert-{{ banner.style }} show">
-                        {{ banner.message }} <a href="{{ banner.url }}">{{ banner.link_text }}</a>
+                        {{ banner.message|safe }}{% if banner.url %} <a href="{{ banner.url }}">{{ banner.link_text }}</a>{% endif %}
+                        {% if banner.expanded_html %}
+                            <a role="button" data-toggle="collapse" href="#os-banner-expanded-{{ forloop.counter }}"
+                               aria-expanded="false" aria-controls="os-banner-expanded-{{ forloop.counter }}"
+                               class="os-banner-toggle collapsed">
+                                <i class="fa-solid fa-caret-down"></i>
+                            </a>
+                            <div id="os-banner-expanded-{{ forloop.counter }}" class="collapse os-banner-expanded">
+                                {{ banner.expanded_html|safe }}
+                            </div>
+                        {% endif %}
                     </div>
                 {% endfor %}
                 <div class="container-fluid">

dojo/templates/base.html

@@ -0,0 +1,198 @@
+from unittest.mock import patch
+
+import requests
+from django.core.cache import cache
+from django.template import Context, Template
+from django.test import RequestFactory, SimpleTestCase, override_settings
+
+from dojo import context_processors
+from dojo.announcements import os_message
+
+
+class _Resp:
+    def __init__(self, status_code=200, text=""):
+        self.status_code = status_code
+        self.text = text
+
+
+@override_settings(CACHES={"default": {"BACKEND": "django.core.cache.backends.locmem.LocMemCache"}})
+class TestParseOsMessage(SimpleTestCase):
+
+    def setUp(self):
+        cache.clear()
+
+    def test_valid_doc_with_expanded(self):
+        text = (
+            "# DefectDojo v3.0 is available\n"
+            "\n"
+            "## Expanded Message\n"
+            "\n"
+            "- Major feature A\n"
+            "- Major feature B\n"
+        )
+        result = os_message.parse_os_message(text)
+        self.assertEqual(result["message"], "DefectDojo v3.0 is available")
+        self.assertIn("<li>Major feature A</li>", result["expanded_html"])
+        self.assertIn("<li>Major feature B</li>", result["expanded_html"])
+
+    def test_missing_headline_returns_none(self):
+        text = "No headline here\n## Expanded Message\nbody\n"
+        self.assertIsNone(os_message.parse_os_message(text))
+
+    def test_headline_inline_markdown(self):
+        text = "# Read the **release notes** at [link](https://example.com)\n"
+        result = os_message.parse_os_message(text)
+        self.assertIn("<strong>release notes</strong>", result["message"])
+        self.assertIn('<a href="https://example.com">link</a>', result["message"])
+        self.assertIsNone(result["expanded_html"])
+
+    def test_headline_strips_disallowed_html(self):
+        text = "# Headline <script>alert(1)</script> tail\n"
+        result = os_message.parse_os_message(text)
+        self.assertNotIn("<script", result["message"])
+        self.assertNotIn("</script>", result["message"])
+        self.assertIn("Headline", result["message"])
+
+    def test_missing_expanded_section(self):
+        text = "# Just a headline\n"
+        result = os_message.parse_os_message(text)
+        self.assertEqual(result["message"], "Just a headline")
+        self.assertIsNone(result["expanded_html"])
+
+    def test_expanded_with_fenced_code(self):
+        text = (
+            "# Headline\n"
+            "## Expanded Message\n"
+            "```python\n"
+            "print('hi')\n"
+            "```\n"
+        )
+        result = os_message.parse_os_message(text)
+        self.assertIn("<pre>", result["expanded_html"])
+        self.assertIn("<code>", result["expanded_html"])
+        self.assertIn("print('hi')", result["expanded_html"])
+
+    def test_expanded_strips_script_tag(self):
+        text = (
+            "# Headline\n"
+            "## Expanded Message\n"
+            "<script>alert(1)</script>\n"
+            "Body paragraph\n"
+        )
+        result = os_message.parse_os_message(text)
+        self.assertNotIn("<script", result["expanded_html"])
+        self.assertNotIn("</script>", result["expanded_html"])
+        self.assertIn("Body paragraph", result["expanded_html"])
+
+    def test_headline_outer_p_is_stripped(self):
+        text = "# Plain headline\n"
+        result = os_message.parse_os_message(text)
+        self.assertFalse(result["message"].startswith("<p>"))
+        self.assertFalse(result["message"].endswith("</p>"))
+
+    def test_headline_truncated_to_100_chars(self):
+        long_headline = "x" * 200
+        text = f"# {long_headline}\n"
+        result = os_message.parse_os_message(text)
+        self.assertLessEqual(len(result["message"]), 100)
+
+
+@override_settings(CACHES={"default": {"BACKEND": "django.core.cache.backends.locmem.LocMemCache"}})
+class TestFetchOsMessage(SimpleTestCase):
+
+    def setUp(self):
+        cache.clear()
+
+    def test_200_with_body_caches_body(self):
+        body = "# headline\n"
+        with patch("dojo.announcements.os_message.requests.get", return_value=_Resp(200, body)) as mock_get:
+            result = os_message.fetch_os_message()
+        self.assertEqual(result, body)
+        self.assertEqual(cache.get(os_message.CACHE_KEY), body)
+        mock_get.assert_called_once()
+
+    def test_404_caches_none(self):
+        with patch("dojo.announcements.os_message.requests.get", return_value=_Resp(404, "not found")):
+            result = os_message.fetch_os_message()
+        self.assertIsNone(result)
+        self.assertIsNone(cache.get(os_message.CACHE_KEY, default="sentinel"))
+
+    def test_timeout_caches_none(self):
+        with patch("dojo.announcements.os_message.requests.get", side_effect=requests.exceptions.Timeout):
+            result = os_message.fetch_os_message()
+        self.assertIsNone(result)
+        self.assertIsNone(cache.get(os_message.CACHE_KEY, default="sentinel"))
+
+    def test_connection_error_caches_none(self):
+        with patch("dojo.announcements.os_message.requests.get", side_effect=requests.exceptions.ConnectionError):
+            result = os_message.fetch_os_message()
+        self.assertIsNone(result)
+        self.assertIsNone(cache.get(os_message.CACHE_KEY, default="sentinel"))
+
+    def test_empty_body_caches_none(self):
+        with patch("dojo.announcements.os_message.requests.get", return_value=_Resp(200, "   \n\n")):
+            result = os_message.fetch_os_message()
+        self.assertIsNone(result)
+        self.assertIsNone(cache.get(os_message.CACHE_KEY, default="sentinel"))
+
+    def test_second_call_hits_cache(self):
+        with patch("dojo.announcements.os_message.requests.get", return_value=_Resp(200, "# h\n")) as mock_get:
+            os_message.fetch_os_message()
+            os_message.fetch_os_message()
+        self.assertEqual(mock_get.call_count, 1)
+
+    def test_second_call_after_failure_also_hits_cache(self):
+        with patch("dojo.announcements.os_message.requests.get", side_effect=requests.exceptions.Timeout) as mock_get:
+            os_message.fetch_os_message()
+            os_message.fetch_os_message()
+        self.assertEqual(mock_get.call_count, 1)
+
+
+@override_settings(CACHES={"default": {"BACKEND": "django.core.cache.backends.locmem.LocMemCache"}})
+class TestGetOsBanner(SimpleTestCase):
+
+    def setUp(self):
+        cache.clear()
+
+    def test_returns_none_when_fetch_returns_none(self):
+        with patch("dojo.announcements.os_message.fetch_os_message", return_value=None):
+            self.assertIsNone(os_message.get_os_banner())
+
+    def test_swallows_parse_exception(self):
+        with patch("dojo.announcements.os_message.fetch_os_message", return_value="# ok\n"), \
+             patch("dojo.announcements.os_message.parse_os_message", side_effect=RuntimeError("boom")):
+            self.assertIsNone(os_message.get_os_banner())
+
+
+@override_settings(CACHES={"default": {"BACKEND": "django.core.cache.backends.locmem.LocMemCache"}})
+class TestGlobalizeVarsOsBanner(SimpleTestCase):
+
+    def setUp(self):
+        cache.clear()
+        self.request = RequestFactory().get("/")
+
+    def test_additional_banners_populated_when_banner_present(self):
+        banner = {"message": "<strong>Hi</strong>", "expanded_html": "<p>body</p>"}
+        with patch.object(context_processors, "get_os_banner", return_value=banner):
+            result = context_processors.globalize_vars(self.request)
+        self.assertIn("additional_banners", result)
+        entry = result["additional_banners"][0]
+        self.assertEqual(entry["message"], "<strong>Hi</strong>")
+        self.assertEqual(entry["expanded_html"], "<p>body</p>")
+        self.assertEqual(entry["style"], "info")
+        self.assertEqual(entry["url"], "")
+        self.assertEqual(entry["link_text"], "")
+
+    def test_additional_banners_absent_when_no_banner(self):
+        with patch.object(context_processors, "get_os_banner", return_value=None):
+            result = context_processors.globalize_vars(self.request)
+        self.assertNotIn("additional_banners", result)
+
+    def test_template_renders_bleached_message(self):
+        banner = {"message": "<strong>Hi</strong>", "expanded_html": None}
+        with patch.object(context_processors, "get_os_banner", return_value=banner):
+            ctx = context_processors.globalize_vars(self.request)
+        rendered = Template(
+            "{% for b in additional_banners %}{{ b.message|safe }}{% endfor %}",
+        ).render(Context(ctx))
+        self.assertIn("<strong>Hi</strong>", rendered)