🎨 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

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## 1.10.6
 
+### Added
+
+- Protection against log injection attacks.
+
 ### Fixed
 
 - Column misalignment in Ripple-to-REDCap.

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,
-        instrument,
+        safe_record,
+        safe_record_for_log(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,25 @@ 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,
-        instrument,
+        safe_record,
+        safe_record_for_log(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,22 @@
 import os
 from pathlib import Path
 import platform
+import re
 import sys
 from typing import Optional, Set
+import unicodedata
+
+_ANSI_RE = re.compile(
+    r"\x1b\[[0-9;]*+[a-zA-Z]"  # CSI sequences:  ESC [ <params> <letter>
+    r"|\x1b\](?>[^\x07]*)\x07"  # OSC sequences:  ESC ] <payload> BEL
+)
+"""Possessive quantifier (*+) and atomic group ((?>...)) prevent catastrophic
+backtracking (ReDoS)."""
+
+_UNSAFE_CHARS = frozenset(("\u2028", "\u2029"))
+"""Line separator ↵ and paragraph separator ¶."""
+_UNSAFE_CATEGORIES = frozenset(("Cc", "Cf"))
+"""Control characters and format characters."""
 
 _initialized = False
 """Is logging initialized?"""
@@ -126,6 +140,44 @@ 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
+    -------
+    str
+        A sanitized copy of the string with dangerous characters removed.
+
+    """
+    result = _ANSI_RE.sub("", record)
+    return "".join(
+        ch
+        for ch in result
+        if ch not in _UNSAFE_CHARS
+        and unicodedata.category(ch) not in _UNSAFE_CATEGORIES
+    )
+
+
 def setup_tsv_logger(name="tsv_logger", filename="error_log.tsv", level=logging.ERROR):
     """Configure a logger with TSV output."""
     logger = logging.getLogger(name)