From b07530c3ec3b757d2daea1e61907617d36084a5d Mon Sep 17 00:00:00 2001 From: Matthias Strubel Date: Wed, 14 Oct 2026 15:42:49 +0200 Subject: [PATCH 1/3] fix(fronius): do not log the digest auth header in clear text Code scanning alert py/clear-text-logging-sensitive-data (high): the auth debug logger received the full digest Authorization header. That header is derived from the inverter password (password -> a1 -> ha1 -> response) and the digest is replayable for as long as the server nonce is valid, so a shared log file hands over inverter access. Credential header values are now shortened to their first and last three characters before logging - enough to tell two credentials apart or to recognise the configured one, nothing more: before: Digest username="customer", ..., response="68d12d78...548b6938" after: {'Authorization': 'Dig...38"', 'Content-Type': 'application/json'} Values of six characters or less are masked completely, since keeping both ends would give them away. Non-credential headers stay readable, and the nonce, cnonce and nc values needed to debug digest auth are still logged in full by the auth logger elsewhere. The same header dict is interpolated into the RuntimeError raised when a request fails without a response, which ends up in the log as well, so it is masked there too. Co-Authored-By: Claude Opus 5 --- src/batcontrol/inverter/fronius.py | 6 +- src/batcontrol/value_utils.py | 56 ++++++++++++++ .../batcontrol/inverter/test_fronius_auth.py | 45 ++++++++++- tests/batcontrol/test_value_utils.py | 77 +++++++++++++++++++ 4 files changed, 181 insertions(+), 3 deletions(-) diff --git a/src/batcontrol/inverter/fronius.py b/src/batcontrol/inverter/fronius.py index ff7e704b..92805730 100644 --- a/src/batcontrol/inverter/fronius.py +++ b/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__) @@ -878,7 +879,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 +905,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}" ) diff --git a/src/batcontrol/value_utils.py b/src/batcontrol/value_utils.py index 93eed6a3..12bd61fa 100644 --- a/src/batcontrol/value_utils.py +++ b/src/batcontrol/value_utils.py @@ -28,6 +28,62 @@ 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 '' + text = str(value) + if not text: + return '' + 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 get_config_value( config: dict, key: str, diff --git a/tests/batcontrol/inverter/test_fronius_auth.py b/tests/batcontrol/inverter/test_fronius_auth.py index 422c3b2a..24301f29 100644 --- a/tests/batcontrol/inverter/test_fronius_auth.py +++ b/tests/batcontrol/inverter/test_fronius_auth.py @@ -4,6 +4,7 @@ from packaging import version from batcontrol.inverter.fronius import hash_utf8, get_api_config, FroniusApiConfig +from batcontrol.value_utils import mask_sensitive_headers class TestFroniusAuthentication(unittest.TestCase): @@ -83,4 +84,46 @@ def test_hash_utf8_with_bytes_input(self): if __name__ == '__main__': - unittest.main() \ No newline at end of file + 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') diff --git a/tests/batcontrol/test_value_utils.py b/tests/batcontrol/test_value_utils.py index 6fbb54fe..a5c7e432 100644 --- a/tests/batcontrol/test_value_utils.py +++ b/tests/batcontrol/test_value_utils.py @@ -3,6 +3,8 @@ from batcontrol.value_utils import ( get_config_value, + mask_secret, + mask_sensitive_headers, parse_bool_flag, parse_optional_ratio, parse_positive_number, @@ -156,3 +158,78 @@ def test_canonical_key_does_not_warn(self, caplog): with caplog.at_level('WARNING'): get_config_value({'api_token': 'x'}, 'api_token', None, 'utility') assert not [r for r in caplog.records if 'deprecated' in r.message] + + +class TestMaskSecret: + """mask_secret() keeps a credential recognisable without logging it. + + Code scanning alert py/clear-text-logging-sensitive-data: the Fronius + digest Authorization header is derived from the inverter password and was + logged in full by the auth debug logger. + """ + + def test_keeps_first_and_last_three_characters(self): + assert mask_secret('eyJhbGciOiJIUzI1NiJ9fQ') == 'eyJ...9fQ' + + def test_recognisable_but_not_reusable(self): + token = 'A' * 40 + 'xyz' + masked = mask_secret(token) + assert masked.startswith('AAA') + assert masked.endswith('xyz') + assert token not in masked + assert len(masked) == 9 + + @pytest.mark.parametrize("secret", ['', None]) + def test_unset_and_empty(self, secret): + assert mask_secret(secret) in ('', '') + + @pytest.mark.parametrize("secret", ['a', 'ab', 'abc', 'abcdef']) + def test_short_secrets_are_masked_completely(self, secret): + """Keeping both ends of a short secret would give it away""" + masked = mask_secret(secret) + assert masked == '*' * len(secret) + assert secret not in masked + + def test_keep_is_configurable(self): + assert mask_secret('0123456789', keep=2) == '01...89' + + def test_non_string_values(self): + assert mask_secret(1234567890) == '123...890' + + +class TestMaskSensitiveHeaders: + """Only credential headers are shortened, the rest stays readable""" + + def test_authorization_is_masked(self): + digest = ('Digest username="customer", realm="Webinterface area", ' + 'nonce="abc", response="0123456789abcdef"') + + masked = mask_sensitive_headers({'Authorization': digest}) + + assert masked['Authorization'] == 'Dig...ef"' + assert 'response="0123456789abcdef"' not in masked['Authorization'] + + def test_other_headers_are_untouched(self): + headers = {'Content-Type': 'application/json', 'Accept': '*/*'} + + assert mask_sensitive_headers(headers) == headers + + @pytest.mark.parametrize("name", [ + 'Authorization', 'authorization', 'AUTHORIZATION', + 'Proxy-Authorization', 'Cookie', 'Set-Cookie', 'X-Api-Key', + 'api-key', 'X-Auth-Token', 'x-access-token', + ]) + def test_sensitive_header_names(self, name): + masked = mask_sensitive_headers({name: 'secretvalue123'}) + assert masked[name] == 'sec...123' + + def test_original_mapping_is_not_modified(self): + headers = {'Authorization': 'secretvalue123'} + + mask_sensitive_headers(headers) + + assert headers['Authorization'] == 'secretvalue123' + + @pytest.mark.parametrize("headers", [None, {}]) + def test_empty_headers(self, headers): + assert mask_sensitive_headers(headers) == {} From fce304c8f53c747fdbf277de6e32427fa9962f7e Mon Sep 17 00:00:00 2001 From: Matthias Strubel Date: Wed, 14 Oct 2026 15:46:49 +0200 Subject: [PATCH 2/3] fix(fcsolar): keep the API key out of logged API responses Code scanning alerts #11-#14 (py/clear-text-logging-sensitive-data, high): forecast.solar takes the API key as the first path segment of the request URL, so every response is derived from the key - and the rate limit branch logs response.text plus a full header dump, the error branch logs the body and puts it into the ProviderError message. A body or header that echoes the request path hands the key to anyone reading the log. The key is now replaced by its masked form (first and last three characters) in everything logged from that response, and credential header values are shortened as well: before: ... "text":"rate limit exceeded for /secret-key-1234567890/estimate" after: ... "text":"rate limit exceeded for /sec...890/estimate" Replacing instead of truncating keeps the API's own message readable, which is the reason it is logged at all. Co-Authored-By: Claude Opus 5 --- src/batcontrol/forecastsolar/fcsolar.py | 17 ++-- src/batcontrol/value_utils.py | 19 ++++ .../batcontrol/forecastsolar/test_fcsolar.py | 91 ++++++++++++++++++- tests/batcontrol/test_value_utils.py | 46 ++++++++++ 4 files changed, 165 insertions(+), 8 deletions(-) diff --git a/src/batcontrol/forecastsolar/fcsolar.py b/src/batcontrol/forecastsolar/fcsolar.py index 85579930..406b5a86 100644 --- a/src/batcontrol/forecastsolar/fcsolar.py +++ b/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,7 +132,7 @@ 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 ) @@ -140,19 +140,22 @@ def get_raw_data_from_provider(self, pvinstallation_name) -> dict: 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__': diff --git a/src/batcontrol/value_utils.py b/src/batcontrol/value_utils.py index 12bd61fa..a8739793 100644 --- a/src/batcontrol/value_utils.py +++ b/src/batcontrol/value_utils.py @@ -84,6 +84,25 @@ def mask_sensitive_headers(headers) -> dict: } +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, diff --git a/tests/batcontrol/forecastsolar/test_fcsolar.py b/tests/batcontrol/forecastsolar/test_fcsolar.py index 7a7f1aef..2c9eb937 100644 --- a/tests/batcontrol/forecastsolar/test_fcsolar.py +++ b/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'} diff --git a/tests/batcontrol/test_value_utils.py b/tests/batcontrol/test_value_utils.py index a5c7e432..9fd41e8d 100644 --- a/tests/batcontrol/test_value_utils.py +++ b/tests/batcontrol/test_value_utils.py @@ -5,6 +5,7 @@ get_config_value, mask_secret, mask_sensitive_headers, + redact_secrets, parse_bool_flag, parse_optional_ratio, parse_positive_number, @@ -233,3 +234,48 @@ def test_original_mapping_is_not_modified(self): @pytest.mark.parametrize("headers", [None, {}]) def test_empty_headers(self, headers): assert mask_sensitive_headers(headers) == {} + + +class TestRedactSecrets: + """redact_secrets() keeps a logged message readable but secret-free. + + forecast.solar takes the API key as the first path segment of the request + URL, so an error body or header echoing that path would leak it + (code scanning py/clear-text-logging-sensitive-data). + """ + + def test_secret_is_replaced_by_its_masked_form(self): + text = 'Forbidden for /secret-key-1234567890/estimate' + + result = redact_secrets(text, 'secret-key-1234567890') + + assert result == 'Forbidden for /sec...890/estimate' + + def test_message_stays_readable(self): + text = '{"message":{"code":429,"text":"rate limit exceeded"}}' + + result = redact_secrets(text, 'secret-key-1234567890') + + assert result == text + + def test_every_occurrence_is_replaced(self): + result = redact_secrets('abcdefghij and abcdefghij', 'abcdefghij') + + assert 'abcdefghij' not in result + assert result.count('abc...hij') == 2 + + def test_several_secrets(self): + result = redact_secrets( + 'key=aaaaaaaaaa token=bbbbbbbbbb', 'aaaaaaaaaa', 'bbbbbbbbbb') + + assert result == 'key=aaa...aaa token=bbb...bbb' + + @pytest.mark.parametrize("secret", [None, '']) + def test_unset_secrets_are_ignored(self, secret): + assert redact_secrets('nothing to hide', secret) == 'nothing to hide' + + def test_no_secret_given(self): + assert redact_secrets('nothing to hide') == 'nothing to hide' + + def test_non_string_input(self): + assert redact_secrets(429, 'aaaaaaaaaa') == '429' From 18d219746138f17e686526995b62c61c91c98eb4 Mon Sep 17 00:00:00 2001 From: Matthias Strubel Date: Wed, 14 Oct 2026 16:01:07 +0200 Subject: [PATCH 3/3] fix(fronius): mask headers in the send_request failure path too Review feedback on PR #452: __send_one_http_request() assigns headers['Authorization'] on the dict it was handed, so the caller's dict carries the password-derived digest afterwards. send_request() interpolates that same dict into the RuntimeError it raises for any response that is neither 200 nor 401/403, which puts the digest back into the log through the other door. CodeQL did not flag this sink because the taint travels through the mutated dict across two methods. Masked the same way, with a regression test that mirrors the mutation. Co-Authored-By: Claude Opus 5 --- src/batcontrol/inverter/fronius.py | 6 +- .../batcontrol/inverter/test_fronius_auth.py | 56 ++++++++++++++++++- 2 files changed, 60 insertions(+), 2 deletions(-) diff --git a/src/batcontrol/inverter/fronius.py b/src/batcontrol/inverter/fronius.py index 92805730..932c020a 100644 --- a/src/batcontrol/inverter/fronius.py +++ b/src/batcontrol/inverter/fronius.py @@ -853,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}" ) diff --git a/tests/batcontrol/inverter/test_fronius_auth.py b/tests/batcontrol/inverter/test_fronius_auth.py index 24301f29..9468d7df 100644 --- a/tests/batcontrol/inverter/test_fronius_auth.py +++ b/tests/batcontrol/inverter/test_fronius_auth.py @@ -1,9 +1,12 @@ """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 @@ -127,3 +130,54 @@ def test_non_sensitive_headers_are_kept(self): 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)