Zugangsdaten in URLs im Log unkenntlich machen

packages/helpermodules/logger.py

@@ -26,6 +26,9 @@
     (r'"{field}":\s*"(.*?)"', r'"{field}": "***REDACTED***"'),  # "field": "value", JSON formatted data
     (r'\'{field}\':\s*\'(.*?)\'', r"'{field}': '***REDACTED***'")  # 'field': 'value', JSON formatted data
 ]
+# Credentials in a URL (scheme://user:password@host/) carry no field name for REDACTION_PATTERNS
+# to key on. The user name is kept, it is useful when diagnosing authentication problems.
+URL_CREDENTIALS_PATTERN = (r'(\w+://[^/\s:@]*):[^/\s]+@', r'\1:***REDACTED***@')
 
 
 def redact_sensitive_info(message: str, additional_fields: list = None) -> str:
@@ -37,13 +40,17 @@ def redact_sensitive_info(message: str, additional_fields: list = None) -> str:
     redacted are defined in the KNOWN_SENSITIVE_FIELDS list. The function uses
     predefined patterns to identify and replace the sensitive information.
 
+    Passwords given as URL credentials (scheme://user:password@host/) are redacted as well,
+    those carry no field name to key on.
+
     Args:
         message (str): The log message to be redacted.
 
     Returns:
         str: The redacted log message.
     """
     fields_to_redact = KNOWN_SENSITIVE_FIELDS + (additional_fields or [])
+    message = re.sub(URL_CREDENTIALS_PATTERN[0], URL_CREDENTIALS_PATTERN[1], message)
     for field in fields_to_redact:
         for pattern, replacement in REDACTION_PATTERNS:
             pattern = pattern.replace('{field}', field)
@@ -344,9 +351,9 @@ def threading_excepthook(args):
         with open(thread_errors_path, "a") as f:
             f.write("Uncaught exception in thread:\n")
             f.write(f"Type: {args.exc_type}\n")
-            f.write(f"Value: {args.exc_value}\n")
+            f.write(redact_sensitive_info(f"Value: {args.exc_value}\n"))
             import traceback
-            traceback.print_tb(args.exc_traceback, file=f)
+            f.write(redact_sensitive_info("".join(traceback.format_tb(args.exc_traceback))))
     threading.excepthook = threading_excepthook
 
     def handle_unhandled_exception(exc_type, exc_value, exc_traceback):
@@ -356,7 +363,7 @@ def handle_unhandled_exception(exc_type, exc_value, exc_traceback):
         with open(thread_errors_path, "a") as f:
             f.write("Uncaught exception:\n")
             f.write(f"Type: {exc_type}\n")
-            f.write(f"Value: {exc_value}\n")
+            f.write(redact_sensitive_info(f"Value: {exc_value}\n"))
             f.write(f"Traceback:{exc_traceback}\n")
     sys.excepthook = handle_unhandled_exception
 

packages/helpermodules/logger_test.py

@@ -0,0 +1,46 @@
+import pytest
+
+from helpermodules.logger import redact_sensitive_info
+
+
+@pytest.mark.parametrize("message, expected", [
+    pytest.param("connection to ws://openwb:secret@ocpp.example.com/v16/ established",
+                 "connection to ws://openwb:***REDACTED***@ocpp.example.com/v16/ established",
+                 id="websocket url"),
+    pytest.param("wss://openwb:secret@ocpp.example.com:443/v16/",
+                 "wss://openwb:***REDACTED***@ocpp.example.com:443/v16/",
+                 id="port is kept"),
+    pytest.param('{"url": "ws://openwb:secret@ocpp.example.com/v16/", "version": "1.6"}',
+                 '{"url": "ws://openwb:***REDACTED***@ocpp.example.com/v16/", "version": "1.6"}',
+                 id="url embedded in json"),
+    pytest.param("ws://user:p@ssword@host/",
+                 "ws://user:***REDACTED***@host/",
+                 id="at sign within password"),
+    pytest.param("ws://openwb:***REDACTED***@ocpp.example.com/v16/",
+                 "ws://openwb:***REDACTED***@ocpp.example.com/v16/",
+                 id="already redacted"),
+])
+def test_redact_url_credentials(message, expected):
+    assert redact_sensitive_info(message) == expected
+
+
+@pytest.mark.parametrize("message", [
+    pytest.param("http://192.168.1.5:8080/api?value=1", id="url without credentials"),
+    pytest.param("https://api.example.com/mail?to=someone@example.com", id="at sign in query"),
+    pytest.param("http://host:8080 unreachable, contact someone@example.com", id="mail address after url"),
+])
+def test_keep_url_without_credentials(message):
+    assert redact_sensitive_info(message) == message
+
+
+def test_redact_url_credentials_and_known_field():
+    message = '{"url": "ws://openwb:secret@ocpp.example.com/v16/", "password": "abc123"}'
+    expected = '{"url": "ws://openwb:***REDACTED***@ocpp.example.com/v16/", "password": "***REDACTED***"}'
+
+    assert redact_sensitive_info(message) == expected
+
+
+def test_redact_url_credentials_with_sensitive_field_as_user_name():
+    # If the user name matches an entry of KNOWN_SENSITIVE_FIELDS, the field pattern applies on
+    # top and truncates the url. The password is removed in that case as well.
+    assert redact_sensitive_info("ws://token:secret@ocpp.example.com/v16/") == "ws://token=***REDACTED***"