chore: Remediate Vulnerabilites across all repos with shared utilities

Author: mobious999

.hooks/check-compiled-requirements.py

@@ -2,11 +2,22 @@
 if they are out of sync with requirements.in and requirements-dev.in.
 """
 
-import subprocess  # nosec B404
+import subprocess  # nosec B404 - subprocess is used safely with controlled arguments
 import sys
 from pathlib import Path
 from shutil import which
-from typing import Optional
+
+
+def get_pip_compile_path() -> str:
+    """Resolve the full path to pip-compile."""
+    pip_compile = which("pip-compile")
+    if not pip_compile:
+        print("[Error] pip-compile not found in PATH.")
+        sys.exit(1)
+    return pip_compile
+
+
+PIP_COMPILE = get_pip_compile_path()
 
 
 def recompile(in_file: str, out_file: str) -> bool:
@@ -21,8 +32,8 @@ def recompile(in_file: str, out_file: str) -> bool:
     """
     print(f"[Fix] Recompiling {in_file} -> {out_file}")
     try:
-        subprocess.run(  # nosec B603
-            ["pip-compile", in_file, "--resolver=backtracking", "--output-file", out_file],
+        subprocess.run(  # nosec B603 - fully qualified pip-compile path and safe args
+            [PIP_COMPILE, in_file, "--resolver=backtracking", "--output-file", out_file],
             check=True,
         )
         return True
@@ -49,8 +60,8 @@ def check_file(in_file: str, out_file: str, autofix: bool = True) -> bool:
     tmp_out = Path(out_file + ".tmp")
 
     try:
-        subprocess.run(  # nosec B603
-            ["pip-compile", in_file, "--resolver=backtracking", "--output-file", str(tmp_out)],
+        subprocess.run(  # nosec B603 - safe, static argument list
+            [PIP_COMPILE, in_file, "--resolver=backtracking", "--output-file", str(tmp_out)],
             check=True,
             stdout=subprocess.DEVNULL,
             stderr=subprocess.DEVNULL,
@@ -73,10 +84,6 @@ def check_file(in_file: str, out_file: str, autofix: bool = True) -> bool:
 
 def main() -> int:
     """Main entry point for the hook."""
-    if not which("pip-compile"):
-        print("[Error] pip-compile not found in PATH.")
-        return 1
-
     ok1 = check_file("requirements.in", "requirements.txt", autofix=True)
     ok2 = check_file("requirements-dev.in", "requirements-dev.txt", autofix=True)
 

src/app/queue_handler.py

@@ -18,13 +18,19 @@
 logger = setup_logger(__name__)
 shutdown_event = threading.Event()
 
+REDACT_SENSITIVE_LOGS = config.get_config_value("REDACT_SENSITIVE_LOGS", "true").lower() == "true"
+
+
+def safe_log(msg: str) -> str:
+    """Standardized redacted log message."""
+    return f"{msg}: [REDACTED]" if REDACT_SENSITIVE_LOGS else msg
+
 
 def consume_messages(callback: Callable[[list[dict]], None]) -> None:
     """Start the queue listener for the configured queue type.
 
     Args:
         callback: A function that takes a list of messages and processes them.
-
     """
     signal.signal(signal.SIGINT, _graceful_shutdown)
     signal.signal(signal.SIGTERM, _graceful_shutdown)
@@ -35,7 +41,7 @@ def consume_messages(callback: Callable[[list[dict]], None]) -> None:
     elif queue_type == "sqs":
         _start_sqs_listener(callback)
     else:
-        raise ValueError(f"Unsupported QUEUE_TYPE: {queue_type}")
+        raise ValueError("Unsupported QUEUE_TYPE: [REDACTED]")
 
 
 def _graceful_shutdown(signum, frame) -> None:
@@ -50,7 +56,6 @@ def _start_rabbitmq_listener(callback: Callable[[list[dict]], None]) -> None:
 
     Args:
         callback: Function to process received messages.
-
     """
     connection = pika.BlockingConnection(
         pika.ConnectionParameters(
@@ -76,11 +81,11 @@ def on_message(ch: BlockingChannel, method, properties, body: bytes) -> None:
             callback([message])
             ch.basic_ack(delivery_tag=method.delivery_tag)
             logger.debug("✅ RabbitMQ message processed and acknowledged.")
-        except Exception as e:
-            logger.error(f"❌ Error processing RabbitMQ message: {e}")
+        except Exception:
+            logger.error("❌ RabbitMQ message processing failed (details redacted)")
             ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
 
-    logger.info("🚀 Consuming RabbitMQ messages from queue: %s", queue_name)
+    logger.info(safe_log("🚀 Consuming RabbitMQ messages from queue"))
 
     try:
         channel.basic_qos(prefetch_count=config.get_batch_size())
@@ -99,12 +104,11 @@ def _start_sqs_listener(callback: Callable[[list[dict]], None]) -> None:
 
     Args:
         callback: Function to process a batch of messages.
-
     """
     sqs = boto3.client("sqs", region_name=config.get_sqs_region())
     queue_url = config.get_sqs_queue_url()
 
-    logger.info("🚀 Polling SQS queue: %s", queue_url)
+    logger.info(safe_log("🚀 Polling SQS queue"))
 
     while not shutdown_event.is_set():
         try:
@@ -125,17 +129,17 @@ def _start_sqs_listener(callback: Callable[[list[dict]], None]) -> None:
                     payload = json.loads(msg["Body"])
                     payloads.append(payload)
                     receipt_handles.append(msg["ReceiptHandle"])
-                except Exception as e:
-                    logger.warning(f"⚠️ Failed to parse SQS message: {e}")
+                except Exception:
+                    logger.warning("⚠️ Failed to parse SQS message body (redacted)")
 
             if payloads:
                 callback(payloads)
                 for handle in receipt_handles:
                     sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=handle)
                 logger.debug("✅ SQS: Processed and deleted %d message(s)", len(payloads))
 
-        except (BotoCoreError, NoCredentialsError) as e:
-            logger.error("❌ SQS error: %s", e)
+        except (BotoCoreError, NoCredentialsError):
+            logger.error("❌ SQS error encountered (details redacted)")
             time.sleep(5)
 
     logger.info("🛑 SQS polling stopped.")

src/app/queue_sender.py

@@ -20,6 +20,13 @@
 
 logger = setup_logger(__name__)
 
+REDACT_SENSITIVE_LOGS = config_shared.get_config_value("REDACT_SENSITIVE_LOGS", "true").lower() == "true"
+
+
+def safe_log_message(data: dict[str, Any]) -> str:
+    """Return redacted or full string for logging."""
+    return "[REDACTED]" if REDACT_SENSITIVE_LOGS else json.dumps(data, ensure_ascii=False)
+
 
 def publish_to_queue(payload: list[dict[str, Any]]) -> None:
     """Publish processed results to RabbitMQ or SQS.
@@ -28,8 +35,11 @@ def publish_to_queue(payload: list[dict[str, Any]]) -> None:
     ----------
     payload : list[dict[str, Any]]
         A list of message payloads to publish.
-
     """
+    if not isinstance(payload, list):
+        logger.error("❌ Invalid payload type: expected list, got %s", type(payload).__name__)
+        return
+
     queue_type = config_shared.get_queue_type().lower()
 
     for message in payload:
@@ -38,7 +48,8 @@ def publish_to_queue(payload: list[dict[str, Any]]) -> None:
         elif queue_type == "sqs":
             _send_to_sqs(message)
         else:
-            logger.error("❌ Invalid QUEUE_TYPE specified: %s", queue_type)
+            redacted_type = "[REDACTED]" if REDACT_SENSITIVE_LOGS else queue_type
+            logger.error("❌ Invalid QUEUE_TYPE specified. Value may be misconfigured or missing.")
 
 
 @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=2, max=10))
@@ -56,7 +67,6 @@ def _send_to_rabbitmq(data: dict[str, Any]) -> None:
         If connection to RabbitMQ fails.
     Exception
         If message publishing fails.
-
     """
     try:
         credentials = pika.PlainCredentials(
@@ -77,7 +87,9 @@ def _send_to_rabbitmq(data: dict[str, Any]) -> None:
                 routing_key=config_shared.get_rabbitmq_routing_key(),
                 body=json.dumps(data, ensure_ascii=False),
             )
-        logger.info("✅ Published message to RabbitMQ.")
+
+        logger.info("✅ Published message to RabbitMQ: %s", safe_log_message(data))  # nosec
+
     except AMQPConnectionError as e:
         logger.exception("❌ RabbitMQ publish connection error: %s", e)
         raise
@@ -103,7 +115,6 @@ def _send_to_sqs(data: dict[str, Any]) -> None:
         If credentials are not available.
     Exception
         If message publishing fails.
-
     """
     sqs_url = config_shared.get_sqs_queue_url()
     region = config_shared.get_sqs_region()
@@ -114,7 +125,8 @@ def _send_to_sqs(data: dict[str, Any]) -> None:
             QueueUrl=sqs_url,
             MessageBody=json.dumps(data, ensure_ascii=False),
         )
-        logger.info("✅ Published message to SQS (MessageId: %s)", response.get("MessageId"))
+        logger.info("✅ Published message to SQS: %s", safe_log_message(data))  # nosec
+
     except (BotoCoreError, NoCredentialsError) as e:
         logger.exception("❌ Failed to initialize SQS client: %s", e)
         raise