chg: [publish] Improvements to the publish module for the notifications about new vulnerabilities on Mastodon.

Author: cedricbonhomme

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## Release 1.1.0 (2025-03-07)
+
+Improvements to the publish module for the notifications about new vulnerabilities
+on Mastodon.
+
+
 ## Release 1.0.0 (2025-02-13)
 
 This release introduces the capability to report errors, warnings,

fedivuln/conf_sample.py

@@ -16,7 +16,7 @@
 # Templates used when publishing status
 templates = {
     "vulnerability": "You can now share your thoughts on vulnerability "
-    "<VULNID> in Vulnerability-Lookup:\n<LINK>\n\n#VulnerabilityLookup #Vulnerability #Cybersecurity #bot",
+    "<VULNID> in Vulnerability-Lookup:\n<LINK>\n\n<VENDOR> - <PRODUCT>\n\n#VulnerabilityLookup #Vulnerability #Cybersecurity #bot",
     "comment": "Vulnerability <VULNID> has received a comment on "
     "Vulnerability-Lookup:\n\n<TITLE>\n<LINK>\n\n#VulnerabilityLookup #Vulnerability #Cybersecurity #bot",
     "bundle": "A new bundle, <BUNDLETITLE>, has been published "

fedivuln/config.py

@@ -1,7 +1,6 @@
 #! /usr/bin/env python
 
-"""This module is responsible for loading the configuration variables.
-"""
+"""This module is responsible for loading the configuration variables."""
 
 import importlib.util
 import os

fedivuln/monitoring.py

@@ -0,0 +1,32 @@
+import time
+
+import valkey
+
+from fedivuln import config
+
+if config.heartbeat_enabled:
+    valkey_client = valkey.Valkey(config.valkey_host, config.valkey_port)
+
+
+def heartbeat(process_name) -> None:
+    """Sends a heartbeat in the Valkey datastore."""
+    if not config.heartbeat_enabled:
+        return
+    try:
+        valkey_client.set(process_name, time.time(), ex=config.expiration_period)
+    except Exception as e:
+        print(f"Heartbeat error: {e}")
+        raise  # Propagate the error to stop the process
+
+
+def log(level="warning", message="", key="process_logs_FediVuln") -> None:
+    """Reports an error or warning in the Valkey datastore."""
+    timestamp = time.time()
+    log_entry = {"timestamp": timestamp, "level": level, "message": message}
+    try:
+        # Add the log entry to a list, so multiple messages are preserved
+        valkey_client.rpush(key, str(log_entry))
+        valkey_client.expire(key, 86400)  # Expire after 24 hours
+    except Exception as e:
+        print(f"Error reporting failure: {e}")
+        raise

fedivuln/publish.py

@@ -7,7 +7,8 @@
 from mastodon import Mastodon
 
 from fedivuln import config
-from fedivuln.utils import heartbeat, report_error
+from fedivuln.monitoring import heartbeat, log
+from fedivuln.utils import get_vendor_product_cve
 
 # Set up your Mastodon instance with access credentials
 if config.mastodon_clientcred_push and config.mastodon_usercred_push:
@@ -30,24 +31,46 @@ def create_status_content(event_data: str, topic: str) -> str:
     status = config.templates.get(topic, "")
     match topic:
         case "vulnerability":
-            try:  # CVE
+            # CVE
+            try:
+                if (
+                    event_dict["cveMetadata"]["datePublished"]
+                    != event_dict["cveMetadata"]["dateUpdated"]
+                ):
+                    return ""
+                vendor, product = get_vendor_product_cve(event_dict)[0]
                 status = status.replace("<VULNID>", event_dict["cveMetadata"]["cveId"])
                 status = status.replace(
                     "<LINK>",
                     f"https://vulnerability.circl.lu/vuln/{event_dict['cveMetadata']['cveId']}",
                 )
+                status = status.replace("<VENDOR>", vendor)
+                status = status.replace("<PRODUCT>", product)
                 return status
             except Exception:
                 pass
-            try:  # GHSA, PySec
+
+            # GHSA, PySec
+            try:
+                if event_dict["published"] != event_dict["modified"]:
+                    return ""
                 status = status.replace("<VULNID>", event_dict["id"])
                 status = status.replace(
                     "<LINK>", f"https://vulnerability.circl.lu/vuln/{event_dict['id']}"
                 )
+                status = status.replace("<VENDOR>", "")
+                status = status.replace("<PRODUCT>", "")
                 return status
             except Exception:
                 pass
-            try:  # CSAF
+
+            # CSAF
+            try:
+                if (
+                    event_dict["document"]["tracking"]["initial_release_date"]
+                    != event_dict["document"]["tracking"]["current_release_date"]
+                ):
+                    return ""
                 try:
                     vuln_id = event_dict["document"]["tracking"]["id"].replace(":", "_")
                 except Exception:
@@ -56,6 +79,8 @@ def create_status_content(event_data: str, topic: str) -> str:
                 status = status.replace(
                     "<LINK>", f"https://vulnerability.circl.lu/vuln/{vuln_id}"
                 )
+                status = status.replace("<VENDOR>", "")
+                status = status.replace("<PRODUCT>", "")
                 return status
             except Exception:
                 return ""
@@ -67,7 +92,7 @@ def create_status_content(event_data: str, topic: str) -> str:
             status = status.replace("<BUNDLETITLE>", event_dict["payload"]["name"])
             status = status.replace("<LINK>", event_dict["uri"])
         case _:
-            pass
+            status = ""
     return status
 
 
@@ -117,13 +142,13 @@ def listen_to_http_event_stream(url, headers=None, params=None, topic="comment")
 
     except requests.exceptions.RequestException as req_err:
         print(f"Request error: {req_err}")
-        report_error("error", f"Request error with HTTP event stream: {req_err}")
+        log("error", f"Request error with HTTP event stream: {req_err}")
     except KeyboardInterrupt:
         print("\nStream interrupted by user. Closing connection.")
-        report_error("error", "Stream interrupted by user. Closing connection.")
+        log("error", "Stream interrupted by user. Closing connection.")
     except Exception as e:
         print(f"Unexpected error: {e}")
-        report_error("error", f"Unexpected error in listen_to_http_event_stream: {e}")
+        log("error", f"Unexpected error in listen_to_http_event_stream: {e}")
 
 
 def listen_to_valkey_stream(topic="comment"):

fedivuln/stream.py

@@ -7,7 +7,7 @@
 from pyvulnerabilitylookup import PyVulnerabilityLookup
 
 from fedivuln import config
-from fedivuln.utils import heartbeat, report_error
+from fedivuln.monitoring import heartbeat, log
 
 
 # Custom encoder for datetime
@@ -71,7 +71,7 @@ def on_direct_message(self, message):
     # Handle any errors in streaming
     def on_abort(self, err):
         print("Stream aborted with error:", err)
-        report_error("error", f"Stream aborted with error: {err}")
+        log("error", f"Stream aborted with error: {err}")
 
     def handle_heartbeat(self):
         heartbeat(process_name="process_heartbeat_FediVuln")
@@ -105,7 +105,7 @@ def push_sighting_to_vulnerability_lookup(status_uri, vulnerability_ids):
             print(
                 f"Error when sending POST request to the Vulnerability-Lookup server:\n{e}"
             )
-            report_error(
+            log(
                 "error",
                 f"Error when sending POST request to the Vulnerability-Lookup server: {e}",
             )

fedivuln/utils.py

@@ -1,32 +1,5 @@
-import time
-
-import valkey
-
-from fedivuln import config
-
-if config.heartbeat_enabled:
-    valkey_client = valkey.Valkey(config.valkey_host, config.valkey_port)
-
-
-def heartbeat(process_name) -> None:
-    """Sends a heartbeat in the Valkey datastore."""
-    if not config.heartbeat_enabled:
-        return
-    try:
-        valkey_client.set(process_name, time.time(), ex=config.expiration_period)
-    except Exception as e:
-        print(f"Heartbeat error: {e}")
-        raise  # Propagate the error to stop the process
-
-
-def report_error(level="warning", message="", key="process_logs_FediVuln") -> None:
-    """Reports an error or warning in the Valkey datastore."""
-    timestamp = time.time()
-    log_entry = {"timestamp": timestamp, "level": level, "message": message}
-    try:
-        # Add the log entry to a list, so multiple messages are preserved
-        valkey_client.rpush(key, str(log_entry))
-        valkey_client.expire(key, 86400)  # Expire after 24 hours
-    except Exception as e:
-        print(f"Error reporting failure: {e}")
-        raise
+def get_vendor_product_cve(data):
+    return [
+        (entry["vendor"], entry["product"])
+        for entry in data["containers"]["cna"]["affected"]
+    ]

poetry.lock

@@ -5,7 +5,7 @@ build-backend = "poetry.core.masonry.api"
 
 [project]
 name = "FediVuln"
-version = "1.0.0"
+version = "1.1.0"
 description = "A client to gather vulnerability-related information from the Fediverse."
 authors = [
     {name = "Cédric Bonhomme", email = "[email protected]"}
@@ -20,7 +20,6 @@ requires-python = ">=3.10,<4.0"
 dependencies = [
     "pyvulnerabilitylookup (>=2.2.0)",
     "mastodon-py (>=1.8.1)",
-    "pyvulnerabilitylookup (>=2.1.0)",
     "valkey (>=6.0.2)"
 ]
 
@@ -50,6 +49,7 @@ classifiers = [
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
     "License :: OSI Approved :: GNU General Public License v3 or later (GPLv3+)"
 ]
 include = [
@@ -85,7 +85,7 @@ warn_unreachable = true
 show_error_context = true
 pretty = true
 
-exclude = "build|dist|docs|fedivuln.egg-info"
+exclude = "build|dist|docs"
 
 [tool.isort]
 profile = "black"