feat: enhance WebHook reporter to support additional lifecycle events (#6965)

* feat: enhance WebHook reporter to support additional lifecycle events

* [MegaLinter] Apply linters fixes

---------

Co-authored-by: nvuillam <[email protected]>

Author: nvuillam

.trivyignore

@@ -176,6 +176,8 @@ CVE-2025-66506
 CVE-2025-66564
 # https://avd.aquasec.com/nvd/cve-2025-68156 : github.com/expr-lang/expr: Expr: Denial of Service via uncontrolled recursion in expression evaluation, not applicable in MegaLinter context
 CVE-2025-68156
+# https://avd.aquasec.com/nvd/cve-2026-22701 : filelock Time-of-Check-Time-of-Use (TOCTOU) race condition leading to privilege escalation, not applicable in MegaLinter context
+CVE-2026-22701
 # Dockerfile
 DS001
 DS002

CHANGELOG.md

@@ -31,6 +31,7 @@ Note: Can be used with `oxsecurity/megalinter@beta` in your GitHub Action mega-l
 - Reporters
   - Add a link inviting to star MegaLinter
   - Display in the console reporter the working directory from which the commands are executed by @bdovaz
+  - Update WebHook reporter so it can send more events for a better integration with UI
 
 - Doc
   - JSON Schema: add default values for file extensions and file names variables + improve descriptions

megalinter/reporters/WebHookLinterReporter.py

@@ -1,13 +1,16 @@
 #!/usr/bin/env python3
 """
 Web Hook linter reporter
-Post linter results to a Web Hook
+Post linter lifecycle events to a Web Hook endpoint
 """
 import logging
 
-import requests
 from megalinter import Reporter, config
-from megalinter.utils_reporter import build_linter_reporter_external_result
+from megalinter.utils_reporter import (
+    build_linter_reporter_external_result,
+    build_linter_reporter_start_message,
+    post_webhook_message,
+)
 
 
 class WebHookLinterReporter(Reporter):
@@ -18,7 +21,7 @@ class WebHookLinterReporter(Reporter):
     web_hook_data: object | None = None
 
     def __init__(self, params=None):
-        # Deactivate GitHub Status by default
+        # Deactivate WebHook Linter Reporter by default
         self.is_active = False
         self.processing_order = 20  # Run after text reporter
         super().__init__(params)
@@ -35,39 +38,14 @@ def manage_activation(self):
                     "You need to define WEBHOOK_REPORTER_URL to use WebHookReporter"
                 )
 
-    # Snd webHook to remote server
+    # Send message when linter is about to start
+    def initialize(self):
+        start_message = build_linter_reporter_start_message(self)
+        post_webhook_message(self.hook_url, start_message, self, "linter start event")
+
+    # Send message when linter is completed
     def produce_report(self):
         self.web_hook_data = build_linter_reporter_external_result(self)
-        headers = {
-            "accept": "application/json",
-            "content-type": "application/json",
-        }
-        if config.exists(self.master.request_id, "WEBHOOK_REPORTER_BEARER_TOKEN"):
-            headers["authorization"] = (
-                f"Bearer {config.get(self.master.request_id, 'WEBHOOK_REPORTER_BEARER_TOKEN')}"
-            )
-        try:
-            response = requests.post(
-                self.hook_url, headers=headers, json=self.web_hook_data
-            )
-            if 200 <= response.status_code < 299:
-                logging.debug(
-                    f"[WebHook Reporter] Successfully posted Web Hook for {self.master.descriptor_id}"
-                    f" with {self.master.linter_name}"
-                )
-            else:
-                logging.warning(
-                    f"[WebHook Reporter] Error posting Status for {self.master.descriptor_id}"
-                    f" with {self.master.linter_name}: {response.status_code}\n"
-                    f"API response: {response.text}"
-                )
-        except ConnectionError as e:
-            logging.warning(
-                f"[WebHook Reporter] Error posting Web Hook for {self.master.descriptor_id}"
-                f" with {self.master.linter_name}: Connection error {str(e)}"
-            )
-        except Exception as e:
-            logging.warning(
-                f"[WebHook Reporter] Error posting Web Hook for {self.master.descriptor_id}"
-                f" with {self.master.linter_name}: Error {str(e)}"
-            )
+        post_webhook_message(
+            self.hook_url, self.web_hook_data, self, "linter complete event"
+        )

megalinter/reporters/WebHookReporter.py

@@ -0,0 +1,53 @@
+#!/usr/bin/env python3
+"""
+Web Hook reporter
+Post MegaLinter lifecycle events to a Web Hook endpoint
+"""
+import logging
+
+from megalinter import Reporter, config
+from megalinter.utils_reporter import (
+    build_reporter_external_result,
+    build_reporter_start_message,
+    post_webhook_message,
+)
+
+
+class WebHookReporter(Reporter):
+    name = "WEBHOOK_REPORTER"
+    scope = "mega-linter"
+
+    hook_url: str | None = None
+    web_hook_data: object | None = None
+
+    def __init__(self, params=None):
+        # Deactivate WebHook reporter by default
+        self.is_active = False
+        self.processing_order = 20  # Run after text reporter
+        super().__init__(params)
+
+    def manage_activation(self):
+        if config.get(self.master.request_id, "WEBHOOK_REPORTER", "false") == "true":
+            if config.exists(self.master.request_id, "WEBHOOK_REPORTER_URL"):
+                self.is_active = True
+                self.hook_url = config.get(
+                    self.master.request_id, "WEBHOOK_REPORTER_URL"
+                )
+            else:
+                logging.error(
+                    "You need to define WEBHOOK_REPORTER_URL to use WebHookReporter"
+                )
+
+    # Send message when MegaLinter is about to start
+    def initialize(self):
+        start_message = build_reporter_start_message(self)
+        post_webhook_message(
+            self.hook_url, start_message, self, "MegaLinter start event"
+        )
+
+    # Send message when MegaLinter is completed
+    def produce_report(self):
+        self.web_hook_data = build_reporter_external_result(self)
+        post_webhook_message(
+            self.hook_url, self.web_hook_data, self, "MegaLinter complete event"
+        )

megalinter/utils_reporter.py

@@ -6,6 +6,7 @@
 import subprocess
 import time
 
+import requests
 from megalinter import config, utils
 from megalinter.constants import (
     DEFAULT_RELEASE,
@@ -466,6 +467,46 @@ def manage_redis_stream(result, redis_stream):
     return result
 
 
+def build_webhook_headers(request_id: str) -> dict:
+    headers = {
+        "accept": "application/json",
+        "content-type": "application/json",
+    }
+    if config.exists(request_id, "WEBHOOK_REPORTER_BEARER_TOKEN"):
+        headers["authorization"] = (
+            f"Bearer {config.get(request_id, 'WEBHOOK_REPORTER_BEARER_TOKEN')}"
+        )
+    return headers
+
+
+def post_webhook_message(hook_url: str, payload: object, reporter, success_label: str):
+    context = ""
+    if reporter.scope == "linter":
+        context = (
+            f" for {reporter.master.descriptor_id}"
+            f" with {reporter.master.linter_name}"
+        )
+    try:
+        response = requests.post(
+            hook_url,
+            headers=build_webhook_headers(reporter.master.request_id),
+            json=payload,
+        )
+        if 200 <= response.status_code < 299:
+            logging.debug(
+                f"[WebHook Reporter] Successfully posted {success_label}{context}"
+            )
+        else:
+            logging.warning(
+                f"[WebHook Reporter] Error posting {success_label}{context}: {response.status_code}\n"
+                f"API response: {response.text}"
+            )
+    except requests.exceptions.RequestException as e:
+        logging.warning(
+            f"[WebHook Reporter] Error posting {success_label}{context}: {str(e)}"
+        )
+
+
 def send_redis_message(reporter_self, message_data):
     try:
         redis = Redis(