feat: add secret masking in logs (#21)

backend/lib/app/app.py

@@ -10,6 +10,7 @@
 import lib.utils.aiojobs as aiojobs_utils
 import lib.utils.asyncio as asyncio_utils
 import lib.utils.lifecycle_manager as lifecycle_manager_utils
+import lib.utils.logging as logging_utils
 
 logger = logging.getLogger(__name__)
 
@@ -31,9 +32,21 @@ def __init__(
     def from_settings(cls, settings: app_settings.Settings) -> typing.Self:
         # Logging
 
-        logging.basicConfig(
-            level=settings.logs.level,
-            format=settings.logs.format,
+        logging_utils.initialize(
+            config=logging_utils.create_config(
+                log_level=settings.logs.level,
+                log_format=settings.logs.format,
+                loggers={
+                    "asyncio": logging_utils.LoggerConfig(
+                        propagate=False,
+                        level=settings.logs.level,
+                    ),
+                    "gql.transport.aiohttp": logging_utils.LoggerConfig(
+                        propagate=False,
+                        level=settings.logs.level if settings.app.debug else "WARNING",
+                    ),
+                },
+            ),
         )
 
         logger.info("Initializing application")

backend/lib/app/settings.py

@@ -8,6 +8,7 @@
 import lib.task.repositories as task_repositories
 import lib.task.services as task_services
 import lib.utils.aiojobs as aiojobs_utils
+import lib.utils.logging as logging_utils
 
 
 class AppSettings(pydantic_settings.BaseSettings):
@@ -29,7 +30,7 @@ def is_debug(self) -> bool:
 
 
 class LoggingSettings(pydantic_settings.BaseSettings):
-    level: str = "INFO"
+    level: logging_utils.LogLevel = "INFO"
     format: str = "%(asctime)s | %(name)s | %(levelname)s | %(message)s"
 
 

backend/lib/task/base/secret.py

@@ -5,6 +5,7 @@
 
 import pydantic
 
+import lib.utils.logging as logging_utils
 import lib.utils.pydantic as pydantic_utils
 
 
@@ -60,7 +61,10 @@ class EnvSecretConfig(BaseSecretConfig):
 
     @property
     def value(self) -> str:
-        return os.environ[self.key]
+        value = os.environ[self.key]
+        logging_utils.register_secret(value=value, replace_value=f"REPLACED ENV SECRET {self.key}")
+
+        return value
 
 
 __all__ = [

backend/lib/task/jobs/task_processor.py

@@ -35,7 +35,7 @@ async def _process(self) -> None:
         try:
             async with self._queue_repository.acquire(topic=task_repositories.JobTopic.TASK) as task_job:
                 assert isinstance(task_job, task_job_models.TaskJob)
-                logging.debug("Processing TaskJob(%s)", task_job.id)
+                logger.debug("Processing TaskJob(%s)", task_job.id)
                 try:
                     await self._process_task(task_job=task_job)
                 except Exception:

backend/lib/utils/logging/__init__.py

@@ -0,0 +1,2 @@
+from .config import *
+from .formatters import *

backend/lib/utils/logging/config.py

@@ -0,0 +1,70 @@
+import dataclasses
+import logging.config as logging_config
+import sys
+import typing
+
+import lib.utils.logging.formatters as formatters
+
+LogLevel = typing.Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
+
+
+def initialize(
+    config: dict[str, typing.Any],
+) -> None:
+    logging_config.dictConfig(config)
+
+
+@dataclasses.dataclass
+class LoggerConfig:
+    propagate: bool
+    level: LogLevel
+
+
+def create_config(
+    log_level: LogLevel,
+    log_format: str,
+    loggers: dict[str, LoggerConfig] | None = None,
+) -> dict[str, typing.Any]:
+    config: dict[str, typing.Any] = {
+        "version": 1,
+        "disable_existing_loggers": False,
+        "handlers": {
+            "default": {
+                "stream": sys.stdout,
+                "level": log_level,
+                "formatter": "default",
+                "class": "logging.StreamHandler",
+            },
+        },
+        "formatters": {
+            "default": {
+                "()": lambda: formatters.CustomFormatter(
+                    fmt=log_format,
+                ),
+            },
+        },
+        "root": {
+            "level": log_level,
+            "handlers": ["default"],
+        },
+    }
+
+    if loggers:
+        config["loggers"] = {
+            logger_name: {
+                "handlers": ["default"],
+                "level": logger_config.level,
+                "propagate": logger_config.propagate,
+            }
+            for logger_name, logger_config in loggers.items()
+        }
+
+    return config
+
+
+__all__ = [
+    "LogLevel",
+    "LoggerConfig",
+    "create_config",
+    "initialize",
+]

backend/lib/utils/logging/formatters.py

@@ -0,0 +1,31 @@
+import logging
+import typing
+
+_SECRETS: dict[str, str] = dict()  # value: replace_value
+
+
+def register_secret(value: str, replace_value: str) -> None:
+    _SECRETS[value] = replace_value
+
+
+class CustomFormatter(logging.Formatter):
+    def __init__(
+        self,
+        *args: typing.Any,
+        **kwargs: typing.Any,
+    ) -> None:
+        super().__init__(*args, **kwargs)
+
+    def format(self, record: logging.LogRecord) -> str:
+        log_message = super().format(record)
+
+        for value, replace_value in _SECRETS.items():
+            log_message = log_message.replace(value, f"***{replace_value}***")
+
+        return log_message
+
+
+__all__ = [
+    "CustomFormatter",
+    "register_secret",
+]