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/inverter/fronius.py b/src/batcontrol/inverter/fronius.py index ff7e704b..932c020a 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__) @@ -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}" ) diff --git a/src/batcontrol/value_utils.py b/src/batcontrol/value_utils.py index 93eed6a3..a8739793 100644 --- a/src/batcontrol/value_utils.py +++ b/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 '' + 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 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/inverter/test_fronius_auth.py b/tests/batcontrol/inverter/test_fronius_auth.py index 422c3b2a..9468d7df 100644 --- a/tests/batcontrol/inverter/test_fronius_auth.py +++ b/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() \ 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') + + +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) diff --git a/tests/batcontrol/test_value_utils.py b/tests/batcontrol/test_value_utils.py index 6fbb54fe..9fd41e8d 100644 --- a/tests/batcontrol/test_value_utils.py +++ b/tests/batcontrol/test_value_utils.py @@ -3,6 +3,9 @@ from batcontrol.value_utils import ( get_config_value, + mask_secret, + mask_sensitive_headers, + redact_secrets, parse_bool_flag, parse_optional_ratio, parse_positive_number, @@ -156,3 +159,123 @@ 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) == {} + + +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'