🎨 Make record logging safer

Potential fix for pull request finding 'CodeQL / Log Injection'

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>

Author: shnizzedy

python_jobs/src/hbnmigration/from_redcap/to_curious.py

@@ -25,6 +25,7 @@
     initialize_logging,
     new_curious_account,
     redcap_api_push,
+    safe_record_for_log,
 )
 from .config import Fields, Values
 from .from_redcap import fetch_data, RedcapRecord
@@ -506,20 +507,21 @@ async def redcap_to_curious_webhook(
     3. Set URL to: ``https://your-domain.com/webhook/redcap-to-curious``
 
     """
+    safe_record = safe_record_for_log(record)
     logger.info(
         "Received REDCap trigger for record %s (instrument: %s)",
-        record,
+        safe_record,
         instrument,
     )
     if ready_to_send_to_curious != "1":
-        logger.debug("Ready flag not set for record %s, ignoring trigger", record)
+        logger.debug("Ready flag not set for record %s, ignoring trigger", safe_record)
         return {
             "status": "ignored",
             "message": "Ready flag not set to '1', ignoring trigger",
             "record_id": record,
         }
     background_tasks.add_task(process_record_for_curious, record)
-    logger.info("Queued record %s for Curious push", record)
+    logger.info("Queued record %s for Curious push", safe_record)
     return {
         "status": "accepted",
         "message": f"Trigger accepted for record {record}",

python_jobs/src/hbnmigration/from_redcap/to_redcap.py

@@ -20,7 +20,7 @@
 
 from .._config_variables import redcap_variables
 from ..exceptions import NoData
-from ..utility_functions import initialize_logging, redcap_api_push
+from ..utility_functions import initialize_logging, redcap_api_push, safe_record_for_log
 from .config import Fields, Values
 from .from_redcap import fetch_data, RedcapRecord
 
@@ -443,23 +443,24 @@ async def redcap_to_intake_webhook(
     3. Set URL to: ``https://your-domain.com/webhook/redcap-to-intake``
 
     """
+    safe_record = safe_record_for_log(record)
     logger.info(
         "Received REDCap trigger for record %s (instrument: %s)",
-        record,
+        safe_record,
         instrument,
     )
     if intake_ready != "1":
-        logger.debug("Ready flag not set for record %s, ignoring trigger", record)
+        logger.debug("Ready flag not set for record %s, ignoring trigger", safe_record)
         return {
             "status": "ignored",
             "message": "Ready flag not set to '1', ignoring trigger",
             "record_id": record,
         }
     background_tasks.add_task(process_record_for_redcap_operations, record)
-    logger.info("Queued record %s for Intake REDCap push", record)
+    logger.info("Queued record %s for Intake REDCap push", safe_record)
     return {
         "status": "accepted",
-        "message": f"Trigger accepted for record {record}",
+        "message": f"Trigger accepted for record {safe_record}",
         "record_id": record,
     }
 

python_jobs/src/hbnmigration/utility_functions/__init__.py

@@ -63,7 +63,7 @@
     ValueClass,
     ValueField,
 )
-from .logging import initialize_logging, setup_tsv_logger
+from .logging import initialize_logging, safe_record_for_log, setup_tsv_logger
 from .secrets import ImportWithFallback
 from .typescript import tsx
 
@@ -125,6 +125,7 @@
     "print_module_variables",
     "read_vars_file_as_module",
     "redcap_api_push",
+    "safe_record_for_log",
     "setup_tsv_logger",
     "today",
     "tsx",

python_jobs/src/hbnmigration/utility_functions/logging.py

@@ -9,8 +9,10 @@
 import os
 from pathlib import Path
 import platform
+import re
 import sys
 from typing import Optional, Set
+import unicodedata
 
 _initialized = False
 """Is logging initialized?"""
@@ -126,6 +128,59 @@ def initialize_logging(
     return logging.getLogger(name)
 
 
+def safe_record_for_log(record: str) -> str:
+    r"""
+    Sanitize a string for safe inclusion in log output.
+
+    Mitigates CRLF log injection attacks by removing characters that could
+    be used to forge log entries, evade log analysis, or exploit log viewers.
+
+    Strips the following:
+        - CR/LF (\r, \n) — prevents basic log line injection
+        - Other ASCII control characters (\x00-\x1f, \x7f) — removes tabs,
+          backspaces, escape sequences, null bytes, etc.
+        - Unicode line separators (U+2028, U+2029) — prevents injection via
+          lesser-known Unicode newline characters
+        - ANSI escape sequences (ESC[...m, etc.) — prevents terminal escape
+          attacks in log viewers that interpret color/cursor codes
+        - Unicode control category characters (Cc, Cf) — catches remaining
+          control and formatting characters across all of Unicode
+
+    Parameters
+    ----------
+    record
+        The untrusted string to sanitize before logging.
+
+    Returns
+    -------
+    result
+        A sanitized copy of the string with dangerous characters removed.
+
+    Example:
+        >>> safe_record_for_log("admin\nINFO: Forged entry")
+        'adminINFO: Forged entry'
+        >>> safe_record_for_log("normal\x1b[31m RED ALERT")
+        'normal RED ALERT'
+        >>> safe_record_for_log("benign\u2028Injected line")
+        'benignInjected line'
+
+    """
+    # Strip ANSI escape sequences (e.g. ESC[31m, ESC[0J, ESC]0;title\x07)
+    result = re.sub(r"\x1b\[[0-9;]*[a-zA-Z]", "", record)
+    result = re.sub(r"\x1b\][^\x07]*\x07", "", result)
+
+    # Remove characters in Unicode categories Cc (control) and Cf (format)
+    # This covers \r, \n, \t, \x00, null bytes, soft hyphens, direction
+    # overrides, zero-width characters, and more.
+    result = "".join(
+        ch for ch in result if unicodedata.category(ch) not in ("Cc", "Cf")
+    )
+
+    # Explicitly strip Unicode line/paragraph separators (category Zl/Zp)
+    # in case the Unicode standard ever reclassifies them
+    return result.replace("\u2028", "").replace("\u2029", "")
+
+
 def setup_tsv_logger(name="tsv_logger", filename="error_log.tsv", level=logging.ERROR):
     """Configure a logger with TSV output."""
     logger = logging.getLogger(name)