From 8d5b713bf19020340860e496bdf556673ac492fd Mon Sep 17 00:00:00 2001 From: Christian Krakau-Louis Date: Sun, 29 Mar 2026 20:59:52 +0200 Subject: [PATCH] Potential fix for code scanning alert no. 119: Log Injection Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com> --- backend/app/services/dns_resolver.py | 23 ++++++++++++++++++++--- 1 file changed, 20 insertions(+), 3 deletions(-) diff --git a/backend/app/services/dns_resolver.py b/backend/app/services/dns_resolver.py index c3b8283..a2a235f 100644 --- a/backend/app/services/dns_resolver.py +++ b/backend/app/services/dns_resolver.py @@ -15,6 +15,16 @@ from typing import List, Optional, Tuple logger = logging.getLogger(__name__) +def _sanitize_for_log(value: str) -> str: + """ + Remove characters that could be used for log injection from a string. + + Currently strips carriage returns and newlines to prevent forged log lines. + """ + return value.replace("\r", "").replace("\n", "") + + + def _sanitize_for_log(value: str) -> str: """ Remove newline and carriage return characters from log values to prevent @@ -83,7 +93,8 @@ class BaseDNSProvider(ABC): # ------------------------------------------------------------------ # High-level record checks built on top of lookup_txt # ------------------------------------------------------------------ - + safe_domain = _sanitize_for_log(domain) + logger.debug("DMARC lookup failed for %s: %s", safe_domain, exc) async def check_dmarc(self, domain: str) -> Tuple[bool, Optional[str]]: """Return *(found, record_string)* for the domain's DMARC TXT record.""" try: @@ -94,7 +105,8 @@ class BaseDNSProvider(ABC): except LookupError as exc: logger.debug("DMARC lookup failed for %s: %s", _sanitize_for_log(domain), exc) return False, None - + safe_domain = _sanitize_for_log(domain) + logger.debug("SPF lookup failed for %s: %s", safe_domain, exc) async def check_spf(self, domain: str) -> Tuple[bool, Optional[str]]: """Return *(found, record_string)* for the domain's SPF TXT record.""" try: @@ -108,8 +120,13 @@ class BaseDNSProvider(ABC): async def check_dkim( self, domain: str, selectors: List[str] + safe_selector = _sanitize_for_log(selector) + safe_domain = _sanitize_for_log(domain) ) -> Tuple[bool, Optional[str], Optional[str]]: - """Return *(found, selector, record_string)* for the first working DKIM selector.""" + "DKIM lookup failed for selector=%s domain=%s: %s", + safe_selector, + safe_domain, + exc, for selector in selectors: try: records = await self.lookup_txt(f"{selector}._domainkey.{domain}")