fix(logging): keep credentials out of the logs

src/batcontrol/forecastsolar/fcsolar.py

@@ -9,7 +9,7 @@
 import json
 import logging
 import requests
-from ..value_utils import get_config_value
+from ..value_utils import get_config_value, mask_sensitive_headers, redact_secrets
 from .baseclass import ForecastSolarBaseclass, ProviderError, RateLimitException
 
 
@@ -132,27 +132,30 @@ def get_raw_data_from_provider(self, pvinstallation_name) -> dict:
                 logger.warning(
                     'Forecast solar API rate limit exceeded [%s]. '
                     'Retry after %d seconds at %s',
-                    response.text,
+                    redact_secrets(response.text, apikey),
                     retry_seconds,
                     retry_after_timestamp
                 )
             else:
                 logger.warning(
                     'Forecast solar API rate limit exceeded [%s]. '
                     'No retry after information available, dumping headers',
-                    response.text
+                    redact_secrets(response.text, apikey)
                 )
-                for header, value in response.headers.items():
-                    logger.debug('Header: %s = %s', header, value)
+                for header, value in mask_sensitive_headers(
+                        response.headers).items():
+                    logger.debug('Header: %s = %s', header,
+                                 redact_secrets(value, apikey))
             raise RateLimitException(
                 'Forecast solar API rate limit exceeded')
 
         else:
+            safe_text = redact_secrets(response.text, apikey)
             logger.warning(
                 'Forecast solar API returned %s - %s',
-                response.status_code, response.text)
+                response.status_code, safe_text)
             raise ProviderError(
-                f'Forecast solar API returned {response.status_code} - {response.text}')
+                f'Forecast solar API returned {response.status_code} - {safe_text}')
 
 
 if __name__ == '__main__':

src/batcontrol/inverter/fronius.py

@@ -27,6 +27,7 @@
 import requests
 from packaging import version
 from cachetools import TTLCache
+from ..value_utils import mask_sensitive_headers
 from .baseclass import DEFAULT_MAX_SOC, DEFAULT_MIN_SOC, InverterBaseclass
 
 logger = logging.getLogger(__name__)
@@ -852,10 +853,14 @@ def send_request(self, path, method='GET', payload="", params=None, headers=None
                 self.__retrieve_auth_from_response(response)
                 self.login()
             else:
+                # __send_one_http_request() sets headers['Authorization'] on
+                # the dict passed in, so by now it holds the digest derived
+                # from the inverter password.
                 raise RuntimeError(
                     f"[Inverter] Request {i} failed with {response.status_code}-"
                     f"{response.reason}. \n"
-                    f"\t path:{path}, \n\tparams:{params} \n\theaders {headers} \n"
+                    f"\t path:{path}, \n\tparams:{params} \n"
+                    f"\theaders {mask_sensitive_headers(headers)} \n"
                     f"\tnonce {self.nonce} \n"
                     f"\tpayload {payload}"
                 )
@@ -878,7 +883,7 @@ def __send_one_http_request(self, path, method='GET', payload="",
             headers['Authorization'] = self.get_auth_header(
                 method=method, path=fullpath)
             logger_auth.debug("Requesting %s , header: %s",
-                              fullpath, headers)
+                              fullpath, mask_sensitive_headers(headers))
 
         for i in range(3):
             # 3 retries if connection can't be established
@@ -904,7 +909,8 @@ def __send_one_http_request(self, path, method='GET', payload="",
 
         logger.error('Request failed without response.')
         raise RuntimeError(
-            f"\turl:{url}, \n\tparams:{params} \n\theaders {headers} \n"
+            f"\turl:{url}, \n\tparams:{params} \n"
+            f"\theaders {mask_sensitive_headers(headers)} \n"
             f"\tnonce {self.nonce} \n"
             f"\tpayload {payload}"
         )

src/batcontrol/value_utils.py

@@ -28,6 +28,81 @@ def reset_alias_warnings() -> None:
     _warned_aliases.clear()
 
 
+# Header names whose value carries a credential and must never be logged
+# in full. Lower case, compared case-insensitively.
+SENSITIVE_HEADERS = frozenset({
+    'authorization',
+    'proxy-authorization',
+    'cookie',
+    'set-cookie',
+    'api-key',
+    'x-api-key',
+    'x-auth-token',
+    'x-access-token',
+})
+
+
+def mask_secret(value, keep: int = 3) -> str:
+    """Shorten a credential so it can be logged.
+
+    Keeps the first and last ``keep`` characters, which is enough to tell
+    two credentials apart or to recognise the configured one, without
+    putting the secret itself into the log. Values short enough for the
+    remainder to give the secret away are masked completely.
+
+    Args:
+        value: The credential, of any type
+        keep: Number of characters to keep at each end
+
+    Returns:
+        The shortened representation, e.g. 'eyJ...9fQ'
+    """
+    if value is None:
+        return '<unset>'
+    text = str(value)
+    if not text:
+        return '<empty>'
+    if len(text) <= keep * 2:
+        return '*' * len(text)
+    return f'{text[:keep]}...{text[-keep:]}'
+
+
+def mask_sensitive_headers(headers) -> dict:
+    """Copy an HTTP header mapping with every credential value shortened
+
+    Logging a request's headers is useful for debugging, but an
+    Authorization header holds a credential - for digest auth a hash of the
+    password, which is replayable while the nonce is valid. Only the header
+    values in SENSITIVE_HEADERS are shortened, so the remaining headers stay
+    readable.
+    """
+    if not headers:
+        return {}
+    return {
+        key: mask_secret(value) if str(key).lower() in SENSITIVE_HEADERS else value
+        for key, value in headers.items()
+    }
+
+
+def redact_secrets(text, *secrets, keep: int = 3) -> str:
+    """Replace every occurrence of a secret in ``text`` by its masked form
+
+    Needed wherever a response is logged that was fetched with a credential
+    in the request URL: forecast.solar takes the API key as the first path
+    segment, so an error body or header that echoes the request path would
+    put the key into the log. Replacing instead of truncating keeps the
+    message itself readable, which is the point of logging it.
+    """
+    result = str(text)
+    for secret in secrets:
+        if not secret:
+            continue
+        secret = str(secret)
+        if secret in result:
+            result = result.replace(secret, mask_secret(secret, keep))
+    return result
+
+
 def get_config_value(
         config: dict,
         key: str,

tests/batcontrol/forecastsolar/test_fcsolar.py

@@ -9,7 +9,7 @@
 import requests
 
 from batcontrol.fetcher.relaxed_caching import CacheMissError, RelaxedCaching
-from batcontrol.forecastsolar.baseclass import ProviderError
+from batcontrol.forecastsolar.baseclass import ProviderError, RateLimitException
 from batcontrol.forecastsolar.fcsolar import FCSolar
 
 
@@ -47,6 +47,95 @@ def test_request_error_is_wrapped_without_sensitive_details(instance, mocker):
     assert 'secret-key' not in formatted_error
 
 
+@pytest.fixture
+def instance_with_key():
+    """Provider with an API key - forecast.solar takes it as URL path segment."""
+    return FCSolar(
+        [{
+            'name': 'roof',
+            'lat': 52.17,
+            'lon': 21.25,
+            'declination': 30,
+            'azimuth': 34,
+            'kWp': 10.0,
+            'api_token': 'secret-key-1234567890',
+        }],
+        pytz.timezone('Europe/Warsaw'),
+        min_time_between_api_calls=900,
+    )
+
+
+def _response(mocker, status_code, text, headers=None):
+    """Build a minimal requests-like response."""
+    response = mocker.Mock()
+    response.status_code = status_code
+    response.text = text
+    response.headers = headers if headers is not None else {}
+    return response
+
+
+def test_rate_limit_log_does_not_contain_the_api_key(
+        instance_with_key, mocker, caplog):
+    """Code scanning py/clear-text-logging-sensitive-data: the API key is the
+    first path segment of the request URL, so an error body echoing the path
+    would put it into the log."""
+    echoed = ('{"message":{"code":429,"text":"rate limit",'
+              '"info":{"url":"/secret-key-1234567890/estimate"}}}')
+    mocker.patch(
+        'batcontrol.forecastsolar.fcsolar.requests.get',
+        return_value=_response(mocker, 429, echoed),
+    )
+
+    with caplog.at_level('DEBUG'):
+        with pytest.raises(RateLimitException):
+            instance_with_key.get_raw_data_from_provider('roof')
+
+    assert 'secret-key-1234567890' not in caplog.text
+    assert 'sec...890' in caplog.text
+    # the message itself stays readable
+    assert 'rate limit' in caplog.text
+
+
+def test_rate_limit_header_dump_does_not_contain_the_api_key(
+        instance_with_key, mocker, caplog):
+    """The header dump of the no-retry-after branch is logged at DEBUG."""
+    headers = {
+        'X-Echo-Path': '/secret-key-1234567890/estimate',
+        'Authorization': 'Bearer another-secret-value',
+        'Content-Type': 'application/json',
+    }
+    mocker.patch(
+        'batcontrol.forecastsolar.fcsolar.requests.get',
+        return_value=_response(mocker, 429, '{"message":"rate limit"}', headers),
+    )
+
+    with caplog.at_level('DEBUG'):
+        with pytest.raises(RateLimitException):
+            instance_with_key.get_raw_data_from_provider('roof')
+
+    assert 'secret-key-1234567890' not in caplog.text
+    assert 'another-secret-value' not in caplog.text
+    assert 'Content-Type' in caplog.text
+
+
+def test_error_response_does_not_leak_the_api_key(
+        instance_with_key, mocker, caplog):
+    """The non-200 branch logs the body and puts it into the ProviderError."""
+    echoed = 'Forbidden for /secret-key-1234567890/estimate'
+    mocker.patch(
+        'batcontrol.forecastsolar.fcsolar.requests.get',
+        return_value=_response(mocker, 403, echoed),
+    )
+
+    with caplog.at_level('WARNING'):
+        with pytest.raises(ProviderError) as error_info:
+            instance_with_key.get_raw_data_from_provider('roof')
+
+    assert 'secret-key-1234567890' not in str(error_info.value)
+    assert 'sec...890' in str(error_info.value)
+    assert 'secret-key-1234567890' not in caplog.text
+
+
 def test_refresh_keeps_cached_data_on_request_error(instance, mocker, caplog):
     """A transient request failure must leave the last good response cached."""
     cached_response = {'result': 'last-known-good'}

tests/batcontrol/inverter/test_fronius_auth.py

@@ -1,9 +1,13 @@
 """Tests for Fronius inverter authentication functionality."""
 import unittest
+from unittest.mock import Mock, patch
 import hashlib
 from packaging import version
 
-from batcontrol.inverter.fronius import hash_utf8, get_api_config, FroniusApiConfig
+from batcontrol.inverter.fronius import (
+    hash_utf8, get_api_config, FroniusApiConfig, FroniusWR
+)
+from batcontrol.value_utils import mask_sensitive_headers
 
 
 class TestFroniusAuthentication(unittest.TestCase):
@@ -83,4 +87,97 @@ def test_hash_utf8_with_bytes_input(self):
 
 
 if __name__ == '__main__':
-    unittest.main()
+    unittest.main()
+
+
+class TestAuthHeaderLogging(unittest.TestCase):
+    """The digest Authorization header must not be logged in full.
+
+    Code scanning alert py/clear-text-logging-sensitive-data: the header is
+    built from the inverter password (password -> a1 -> ha1 -> response) and
+    the auth debug logger used to receive it unchanged. The digest is
+    replayable for as long as the server nonce is valid.
+    """
+
+    def setUp(self):
+        self.password = 'testpass'
+        self.user = 'customer'
+        realm = 'Webinterface area'
+        ha1 = hash_utf8(f'{self.user}:{realm}:{self.password}', 'SHA256')
+        ha2 = hash_utf8('GET:/commands/Login', 'SHA256')
+        self.respdig = hash_utf8(f'{ha1}:nonce:00000001:cnonce:auth:{ha2}', 'SHA256')
+        self.header = (
+            f'Digest username="{self.user}", realm="{realm}", nonce="nonce", '
+            f'uri="/commands/Login", algorithm="SHA256", qop=auth, nc=00000001, '
+            f'cnonce="cnonce", response="{self.respdig}"'
+        )
+
+    def test_password_derived_digest_is_not_logged(self):
+        logged = mask_sensitive_headers({'Authorization': self.header})
+
+        self.assertNotIn(self.respdig, logged['Authorization'])
+        self.assertNotIn(self.password, logged['Authorization'])
+
+    def test_header_stays_recognisable(self):
+        logged = mask_sensitive_headers({'Authorization': self.header})
+
+        self.assertEqual(
+            logged['Authorization'], f'Dig...{self.header[-3:]}')
+
+    def test_non_sensitive_headers_are_kept(self):
+        headers = {'Authorization': self.header, 'Content-Type': 'application/json'}
+
+        logged = mask_sensitive_headers(headers)
+
+        self.assertEqual(logged['Content-Type'], 'application/json')
+
+
+class TestSendRequestErrorMasksHeaders(unittest.TestCase):
+    """send_request() must not put the digest into its RuntimeError.
+
+    __send_one_http_request() assigns headers['Authorization'] on the dict it
+    was handed, so the caller's dict carries the password-derived digest by
+    the time send_request() reports a failing response. CodeQL did not flag
+    this second sink because the taint travels through the mutated dict.
+    """
+
+    def setUp(self):
+        self.digest = ('Digest username="customer", realm="Webinterface area", '
+                       'nonce="nonce", response="0123456789abcdef"')
+
+    def _run_failing_request(self, headers):
+        inverter = object.__new__(FroniusWR)
+        inverter.nonce = 'servernonce'
+        digest = self.digest
+
+        def fake_request(_self, path, method, payload, params, request_headers, auth):  # pylint: disable=unused-argument
+            # mirror the mutation done by the real method
+            request_headers['Authorization'] = digest
+            response = Mock()
+            response.status_code = 500
+            response.reason = 'Internal Server Error'
+            return response
+
+        with patch.object(FroniusWR, '_FroniusWR__send_one_http_request',
+                          fake_request):
+            with self.assertRaises(RuntimeError) as context:
+                inverter.send_request('/commands/Status', headers=headers,
+                                      auth=True)
+        return str(context.exception)
+
+    def test_digest_is_not_in_the_error_text(self):
+        message = self._run_failing_request({})
+
+        self.assertNotIn(self.digest, message)
+        self.assertNotIn('0123456789abcdef', message)
+
+    def test_header_stays_recognisable(self):
+        message = self._run_failing_request({})
+
+        self.assertIn(f'Dig...{self.digest[-3:]}', message)
+
+    def test_other_headers_are_still_reported(self):
+        message = self._run_failing_request({'Content-Type': 'application/json'})
+
+        self.assertIn('application/json', message)
+        self.assertIn('Internal Server Error', message)