From e67e59017ec098983ba8022ab3bd68c6524083ea Mon Sep 17 00:00:00 2001 From: David Berenstein Date: Wed, 12 Aug 2026 17:51:05 +0200 Subject: [PATCH 1/5] feat: cache and back off Electricity Maps Carbon intensity was fetched from the Electricity Maps API on every emissions computation, so a long run with a short measure_power_secs issued thousands of requests for a value the grid publishes hourly. A failing token produced one doomed request per measurement tick for the whole run. Extract get_carbon_intensity() from get_emissions(), cache its result for 5 minutes per location, and put the API in an exponential cooldown (30s to 1h) after a failure. get_emissions() is unchanged for callers. Refs #1354 Co-Authored-By: Claude Opus 5 (1M context) --- codecarbon/core/electricitymaps_api.py | 146 +++++++++++++++++++++---- tests/test_electricitymaps_api.py | 1 + tests/test_electricitymaps_cache.py | 130 ++++++++++++++++++++++ 3 files changed, 254 insertions(+), 23 deletions(-) create mode 100644 tests/test_electricitymaps_cache.py diff --git a/codecarbon/core/electricitymaps_api.py b/codecarbon/core/electricitymaps_api.py index cb22c0f79..a01bac575 100644 --- a/codecarbon/core/electricitymaps_api.py +++ b/codecarbon/core/electricitymaps_api.py @@ -1,13 +1,134 @@ -from typing import Any, Dict +import time +from typing import Any, Dict, Optional, Tuple import requests from codecarbon.core.units import EmissionsPerKWh, Energy from codecarbon.external.geography import GeoMetadata +from codecarbon.external.logger import logger URL: str = "https://api.electricitymaps.com/v3/carbon-intensity/latest" ELECTRICITYMAPS_API_TIMEOUT: int = 30 +# Grid carbon intensity is published hourly at best, while emissions are computed +# on every measurement tick, so the value is cached instead of refetched. +ELECTRICITYMAPS_CACHE_TTL: int = 300 +# After a failure (bad token, network down), retry with an exponential cooldown +# instead of issuing one doomed request per measurement tick. +ELECTRICITYMAPS_COOLDOWN_MIN: int = 30 +ELECTRICITYMAPS_COOLDOWN_MAX: int = 3600 + +# {cache key: (monotonic fetch time, carbon intensity in gCO2e/kWh)} +_cache: Dict[str, Tuple[float, float]] = {} +_cooldown_until: float = 0.0 +_cooldown_duration: float = 0.0 + + +def reset_cache() -> None: + """Drop the cached carbon intensities and any pending failure cooldown.""" + global _cooldown_until, _cooldown_duration + _cache.clear() + _cooldown_until = 0.0 + _cooldown_duration = 0.0 + + +def _cache_key(params: Dict[str, Any]) -> str: + return ",".join(f"{key}={params[key]}" for key in sorted(params)) + + +def _get_cached_carbon_intensity(key: str) -> Optional[float]: + cached = _cache.get(key) + if cached is None: + return None + fetched_at, carbon_intensity_g_per_kWh = cached + if time.monotonic() - fetched_at > ELECTRICITYMAPS_CACHE_TTL: + return None + return carbon_intensity_g_per_kWh + + +def _start_cooldown() -> None: + global _cooldown_until, _cooldown_duration + _cooldown_duration = min( + ELECTRICITYMAPS_COOLDOWN_MAX, + max(ELECTRICITYMAPS_COOLDOWN_MIN, _cooldown_duration * 2), + ) + _cooldown_until = time.monotonic() + _cooldown_duration + + +def get_carbon_intensity( + geo: GeoMetadata, electricitymaps_api_token: str = "" +) -> float: + """ + Retrieve the carbon intensity of the grid, in gCO2e/kWh, from the Electricity + Maps API (formerly CO2 Signal) for the given geographic location. + + Values are cached for ``ELECTRICITYMAPS_CACHE_TTL`` seconds, and failures put + the API in an exponential cooldown during which no request is issued. + + Args: + geo (GeoMetadata): + Geographic metadata, including either latitude/longitude + or a country code. + electricitymaps_api_token (str, optional): + The API token for authenticating with the Electricity Maps API + (default is an empty string). + + Returns: + float: + The carbon intensity of the grid, in grams of CO2eq per kWh. + + Raises: + ElectricityMapsAPIError: + If the Electricity Maps API request fails, returns an error, or is + currently in a failure cooldown. + """ + global _cooldown_duration + params: Dict[str, Any] + if geo.latitude: + params = {"lat": geo.latitude, "lon": geo.longitude} + else: + params = {"countryCode": geo.country_2letter_iso_code} + + key = _cache_key(params) + cached_carbon_intensity = _get_cached_carbon_intensity(key) + if cached_carbon_intensity is not None: + logger.debug( + "electricitymaps_api: using cached carbon intensity " + f"{cached_carbon_intensity} gCO2e/kWh for {key}" + ) + return cached_carbon_intensity + + if time.monotonic() < _cooldown_until: + raise ElectricityMapsAPIError( + "Electricity Maps API is in cooldown after a previous failure, " + f"retrying in {_cooldown_until - time.monotonic():.0f} seconds" + ) + + try: + resp = requests.get( + URL, + params=params, + headers={"auth-token": electricitymaps_api_token}, + timeout=ELECTRICITYMAPS_API_TIMEOUT, + ) + if resp.status_code != 200: + message = resp.json().get("error") or resp.json().get("message") + raise ElectricityMapsAPIError(message) + + # API v3 response structure: carbonIntensity is at the root level + response_data = resp.json() + carbon_intensity_g_per_kWh = response_data.get("carbonIntensity") + + if carbon_intensity_g_per_kWh is None: + raise ElectricityMapsAPIError("No carbonIntensity data in response") + except Exception: + _start_cooldown() + raise + + _cooldown_duration = 0.0 + _cache[key] = (time.monotonic(), carbon_intensity_g_per_kWh) + return carbon_intensity_g_per_kWh + def get_emissions( energy: Energy, geo: GeoMetadata, electricitymaps_api_token: str = "" @@ -37,28 +158,7 @@ def get_emissions( ElectricityMapsAPIError: If the Electricity Maps API request fails or returns an error. """ - params: Dict[str, Any] - if geo.latitude: - params = {"lat": geo.latitude, "lon": geo.longitude} - else: - params = {"countryCode": geo.country_2letter_iso_code} - resp = requests.get( - URL, - params=params, - headers={"auth-token": electricitymaps_api_token}, - timeout=ELECTRICITYMAPS_API_TIMEOUT, - ) - if resp.status_code != 200: - message = resp.json().get("error") or resp.json().get("message") - raise ElectricityMapsAPIError(message) - - # API v3 response structure: carbonIntensity is at the root level - response_data = resp.json() - carbon_intensity_g_per_kWh = response_data.get("carbonIntensity") - - if carbon_intensity_g_per_kWh is None: - raise ElectricityMapsAPIError("No carbonIntensity data in response") - + carbon_intensity_g_per_kWh = get_carbon_intensity(geo, electricitymaps_api_token) emissions_per_kWh: EmissionsPerKWh = EmissionsPerKWh.from_g_per_kWh( carbon_intensity_g_per_kWh ) diff --git a/tests/test_electricitymaps_api.py b/tests/test_electricitymaps_api.py index ce81c0e85..1200d8e27 100644 --- a/tests/test_electricitymaps_api.py +++ b/tests/test_electricitymaps_api.py @@ -11,6 +11,7 @@ class TestElectricityMapsAPI(unittest.TestCase): def setUp(self) -> None: # GIVEN + electricitymaps_api.reset_cache() self._energy = Energy.from_energy(kWh=10) self._geo = GeoMetadata( country_iso_code="FRA", diff --git a/tests/test_electricitymaps_cache.py b/tests/test_electricitymaps_cache.py new file mode 100644 index 000000000..d022491a7 --- /dev/null +++ b/tests/test_electricitymaps_cache.py @@ -0,0 +1,130 @@ +import unittest +from unittest import mock + +import responses + +from codecarbon.core import electricitymaps_api +from codecarbon.external.geography import GeoMetadata + + +class TestElectricityMapsCache(unittest.TestCase): + def setUp(self) -> None: + # GIVEN + electricitymaps_api.reset_cache() + self._geo = GeoMetadata( + country_iso_code="FRA", + country_name="France", + region=None, + country_2letter_iso_code="FR", + ) + self._other_geo = GeoMetadata( + country_iso_code="DEU", + country_name="Germany", + region=None, + country_2letter_iso_code="DE", + ) + + def tearDown(self) -> None: + electricitymaps_api.reset_cache() + + def _add_success_response(self, carbon_intensity: float = 58.7) -> None: + responses.add( + responses.GET, + electricitymaps_api.URL, + json={"zone": "FR", "carbonIntensity": carbon_intensity}, + status=200, + ) + + @responses.activate + def test_second_call_within_ttl_does_not_hit_the_api(self): + self._add_success_response() + + first = electricitymaps_api.get_carbon_intensity(self._geo) + second = electricitymaps_api.get_carbon_intensity(self._geo) + + assert first == second == 58.7 + assert len(responses.calls) == 1 + + @responses.activate + def test_a_long_run_issues_a_bounded_number_of_requests(self): + self._add_success_response() + + for _ in range(1000): + electricitymaps_api.get_carbon_intensity(self._geo) + + assert len(responses.calls) == 1 + + @responses.activate + def test_expired_cache_entry_is_refetched(self): + self._add_success_response() + + with mock.patch.object(electricitymaps_api, "ELECTRICITYMAPS_CACHE_TTL", 0): + electricitymaps_api.get_carbon_intensity(self._geo) + electricitymaps_api.get_carbon_intensity(self._geo) + + assert len(responses.calls) == 2 + + @responses.activate + def test_cache_is_keyed_by_location(self): + self._add_success_response() + + electricitymaps_api.get_carbon_intensity(self._geo) + electricitymaps_api.get_carbon_intensity(self._other_geo) + + assert len(responses.calls) == 2 + + @responses.activate + def test_failure_puts_the_api_in_cooldown(self): + responses.add( + responses.GET, + electricitymaps_api.URL, + json={"error": "invalid token"}, + status=401, + ) + + with self.assertRaises(electricitymaps_api.ElectricityMapsAPIError): + electricitymaps_api.get_carbon_intensity(self._geo) + for _ in range(100): + with self.assertRaises(electricitymaps_api.ElectricityMapsAPIError): + electricitymaps_api.get_carbon_intensity(self._geo) + + assert len(responses.calls) == 1 + + @responses.activate + def test_cooldown_doubles_up_to_the_ceiling(self): + responses.add( + responses.GET, + electricitymaps_api.URL, + json={"error": "invalid token"}, + status=401, + ) + + durations = [] + for _ in range(10): + with self.assertRaises(electricitymaps_api.ElectricityMapsAPIError): + electricitymaps_api.get_carbon_intensity(self._geo) + durations.append(electricitymaps_api._cooldown_duration) + # Let the cooldown expire so the next call reaches the API again. + electricitymaps_api._cooldown_until = 0.0 + + assert durations[0] == electricitymaps_api.ELECTRICITYMAPS_COOLDOWN_MIN + assert durations[1] == electricitymaps_api.ELECTRICITYMAPS_COOLDOWN_MIN * 2 + assert durations[-1] == electricitymaps_api.ELECTRICITYMAPS_COOLDOWN_MAX + + @responses.activate + def test_cooldown_is_reset_after_a_successful_call(self): + responses.add( + responses.GET, + electricitymaps_api.URL, + json={"error": "invalid token"}, + status=401, + ) + with self.assertRaises(electricitymaps_api.ElectricityMapsAPIError): + electricitymaps_api.get_carbon_intensity(self._geo) + + responses.reset() + self._add_success_response() + electricitymaps_api._cooldown_until = 0.0 + electricitymaps_api.get_carbon_intensity(self._geo) + + assert electricitymaps_api._cooldown_duration == 0.0 From c737bccb4d12f9db06b2c02cc7217b706370fa46 Mon Sep 17 00:00:00 2001 From: David Berenstein Date: Wed, 12 Aug 2026 19:08:59 +0200 Subject: [PATCH 2/5] test: bypass intensity cache in cumulative test The test varies carbon intensity per measurement to prove emissions are accumulated as deltas rather than recomputed from the latest intensity. The new 5-minute TTL cache served the first value for all three ticks. Disable the TTL for this test so it still exercises the cumulation contract, and reset the module-level cache to avoid cross-test leakage. Co-Authored-By: Claude Opus 5 (1M context) --- tests/test_emissions_tracker.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/tests/test_emissions_tracker.py b/tests/test_emissions_tracker.py index 8ab12e5d8..e3042e4c1 100644 --- a/tests/test_emissions_tracker.py +++ b/tests/test_emissions_tracker.py @@ -11,6 +11,7 @@ import requests import responses +from codecarbon.core import electricitymaps_api from codecarbon.core.units import Energy, Power from codecarbon.emissions_tracker import ( EmissionsTracker, @@ -1021,6 +1022,9 @@ def test_get_detected_hardware( "codecarbon.emissions_tracker.BaseEmissionsTracker.get_detected_hardware" ) @mock.patch("codecarbon.emissions_tracker.PeriodicScheduler") + # A negative TTL expires every entry immediately, so each measurement sees a + # fresh intensity: this test is about cumulating deltas, not about caching. + @mock.patch("codecarbon.core.electricitymaps_api.ELECTRICITYMAPS_CACHE_TTL", -1) def test_cumulative_emissions_with_varying_intensity( self, mock_scheduler, @@ -1038,6 +1042,8 @@ def test_cumulative_emissions_with_varying_intensity( mocked_is_nvidia_system, ): # Setup mocks + electricitymaps_api.reset_cache() + self.addCleanup(electricitymaps_api.reset_cache) mock_geo.return_value = mock.MagicMock( latitude=1.0, longitude=1.0, From a8ad641125fd873e96e828ab69a0eaf0cb74b9fc Mon Sep 17 00:00:00 2001 From: David Berenstein Date: Wed, 12 Aug 2026 19:43:22 +0200 Subject: [PATCH 3/5] fix: address review nits on Electricity Maps caching - key the cache by token so trackers with different tokens do not share a value - raise a dedicated cooldown error and log it at debug, so a bad token no longer produces one error line per measurement tick - document the 5 minute cache TTL as a behaviour change Co-Authored-By: Claude Opus 5 (1M context) --- codecarbon/core/electricitymaps_api.py | 15 ++++++++---- codecarbon/core/emissions.py | 8 +++++++ docs/how-to/configuration.md | 10 ++++++++ tests/test_electricitymaps_cache.py | 33 ++++++++++++++++++++++++++ 4 files changed, 62 insertions(+), 4 deletions(-) diff --git a/codecarbon/core/electricitymaps_api.py b/codecarbon/core/electricitymaps_api.py index a01bac575..a95068a7a 100644 --- a/codecarbon/core/electricitymaps_api.py +++ b/codecarbon/core/electricitymaps_api.py @@ -32,8 +32,11 @@ def reset_cache() -> None: _cooldown_duration = 0.0 -def _cache_key(params: Dict[str, Any]) -> str: - return ",".join(f"{key}={params[key]}" for key in sorted(params)) +def _cache_key(params: Dict[str, Any], electricitymaps_api_token: str) -> str: + # The token is part of the key: two trackers in one process may use + # different tokens, and must not share a cached value. + joined = ",".join(f"{key}={params[key]}" for key in sorted(params)) + return f"{joined},token={electricitymaps_api_token}" def _get_cached_carbon_intensity(key: str) -> Optional[float]: @@ -89,7 +92,7 @@ def get_carbon_intensity( else: params = {"countryCode": geo.country_2letter_iso_code} - key = _cache_key(params) + key = _cache_key(params, electricitymaps_api_token) cached_carbon_intensity = _get_cached_carbon_intensity(key) if cached_carbon_intensity is not None: logger.debug( @@ -99,7 +102,7 @@ def get_carbon_intensity( return cached_carbon_intensity if time.monotonic() < _cooldown_until: - raise ElectricityMapsAPIError( + raise ElectricityMapsAPICooldownError( "Electricity Maps API is in cooldown after a previous failure, " f"retrying in {_cooldown_until - time.monotonic():.0f} seconds" ) @@ -167,3 +170,7 @@ def get_emissions( class ElectricityMapsAPIError(Exception): pass + + +class ElectricityMapsAPICooldownError(ElectricityMapsAPIError): + """Raised when a request is skipped because a previous one failed.""" diff --git a/codecarbon/core/emissions.py b/codecarbon/core/emissions.py index 3b2f10fad..e3fbda581 100644 --- a/codecarbon/core/emissions.py +++ b/codecarbon/core/emissions.py @@ -169,6 +169,14 @@ def get_private_infra_emissions(self, energy: Energy, geo: GeoMetadata) -> float + f"Retrieved emissions for {geo.country_name} using Electricity Maps API :{emissions * 1000} g CO2eq" ) return emissions + except electricitymaps_api.ElectricityMapsAPICooldownError as e: + # The failure that started the cooldown was already logged as an + # error: skipped requests must not log one line per tick. + logger.debug( + "electricitymaps_api.get_emissions: " + + str(e) + + " >>> Using CodeCarbon's data." + ) except Exception as e: logger.error( "electricitymaps_api.get_emissions: " diff --git a/docs/how-to/configuration.md b/docs/how-to/configuration.md index 9f6766aa1..a199b745e 100644 --- a/docs/how-to/configuration.md +++ b/docs/how-to/configuration.md @@ -100,6 +100,16 @@ carbon intensity of your grid. The query runs at the end of each tracking run, and also periodically during long runs (every `api_call_interval × measure_power_secs` seconds; default: every ~2 minutes). +!!! warning "Carbon intensity is cached for 5 minutes" + + A fetched carbon intensity is reused for 5 minutes before the API is + queried again, so measurements taken within that window share the same + intensity value. Electricity Maps publishes hourly at best, but this does + mean a run shorter than 5 minutes converts all of its energy with a single + intensity reading rather than one per tick. After a failure (invalid token, + network down), requests are skipped for an exponentially growing cooldown + (30 s up to 1 hour) and CodeCarbon falls back to its own country data. + The Electricity Maps API offers a free tier. You can sign up and get a token at [electricitymaps.com](https://app.electricitymaps.com/sign-up). diff --git a/tests/test_electricitymaps_cache.py b/tests/test_electricitymaps_cache.py index d022491a7..537746fa8 100644 --- a/tests/test_electricitymaps_cache.py +++ b/tests/test_electricitymaps_cache.py @@ -4,7 +4,10 @@ import responses from codecarbon.core import electricitymaps_api +from codecarbon.core.emissions import Emissions +from codecarbon.core.units import Energy from codecarbon.external.geography import GeoMetadata +from codecarbon.input import DataSource class TestElectricityMapsCache(unittest.TestCase): @@ -128,3 +131,33 @@ def test_cooldown_is_reset_after_a_successful_call(self): electricitymaps_api.get_carbon_intensity(self._geo) assert electricitymaps_api._cooldown_duration == 0.0 + + @responses.activate + def test_cache_is_not_shared_between_tokens(self): + self._add_success_response(carbon_intensity=58.7) + assert electricitymaps_api.get_carbon_intensity(self._geo, "token-a") == 58.7 + + responses.reset() + self._add_success_response(carbon_intensity=412.0) + # WHEN another tracker in the same process uses a different token, it + # must not be served the value cached for the first one. + assert electricitymaps_api.get_carbon_intensity(self._geo, "token-b") == 412.0 + + @responses.activate + def test_cooldown_does_not_log_one_error_per_call(self): + responses.add( + responses.GET, + electricitymaps_api.URL, + json={"error": "invalid token"}, + status=401, + ) + emissions = Emissions(DataSource(), electricitymaps_api_token="bad-token") + energy = Energy.from_energy(kWh=1.0) + + with mock.patch("codecarbon.core.emissions.logger") as mock_logger: + for _ in range(3): + emissions.get_private_infra_emissions(energy, self._geo) + + # THEN only the first, real failure is an error; the calls skipped + # during the cooldown stay at debug level. + assert mock_logger.error.call_count == 1 From 13d568443118ea3f2744c19c4613491011e3baca Mon Sep 17 00:00:00 2001 From: David Berenstein Date: Thu, 13 Aug 2026 07:11:33 +0200 Subject: [PATCH 4/5] fix: hash the Electricity Maps token in the cache key and lock the cache The raw API token was part of the in-memory cache key, so it was stored in the module-level cache and rendered by the debug log that reports a cache hit. Key on a sha256 prefix instead: tokens still get distinct cache entries, but the secret is never held nor logged. The cache and cooldown state are read-modify-written from the background measurement thread. Guard them with one module-level lock, never held across the HTTP request. Co-Authored-By: Claude Opus 5 (1M context) --- codecarbon/core/electricitymaps_api.py | 44 +++++++++++++++++--------- tests/test_electricitymaps_cache.py | 11 +++++++ 2 files changed, 40 insertions(+), 15 deletions(-) diff --git a/codecarbon/core/electricitymaps_api.py b/codecarbon/core/electricitymaps_api.py index a95068a7a..90d969153 100644 --- a/codecarbon/core/electricitymaps_api.py +++ b/codecarbon/core/electricitymaps_api.py @@ -1,3 +1,5 @@ +import hashlib +import threading import time from typing import Any, Dict, Optional, Tuple @@ -22,25 +24,33 @@ _cache: Dict[str, Tuple[float, float]] = {} _cooldown_until: float = 0.0 _cooldown_duration: float = 0.0 +# Emissions are computed from a background measurement thread, so every +# read-modify-write of the state above is serialised. The lock is never held +# across the HTTP request. +_lock = threading.Lock() def reset_cache() -> None: """Drop the cached carbon intensities and any pending failure cooldown.""" global _cooldown_until, _cooldown_duration - _cache.clear() - _cooldown_until = 0.0 - _cooldown_duration = 0.0 + with _lock: + _cache.clear() + _cooldown_until = 0.0 + _cooldown_duration = 0.0 def _cache_key(params: Dict[str, Any], electricitymaps_api_token: str) -> str: # The token is part of the key: two trackers in one process may use - # different tokens, and must not share a cached value. + # different tokens, and must not share a cached value. It is hashed so the + # raw secret is never held in the cache nor rendered in logs. joined = ",".join(f"{key}={params[key]}" for key in sorted(params)) - return f"{joined},token={electricitymaps_api_token}" + token_digest = hashlib.sha256(electricitymaps_api_token.encode()).hexdigest()[:16] + return f"{joined},token={token_digest}" def _get_cached_carbon_intensity(key: str) -> Optional[float]: - cached = _cache.get(key) + with _lock: + cached = _cache.get(key) if cached is None: return None fetched_at, carbon_intensity_g_per_kWh = cached @@ -51,11 +61,12 @@ def _get_cached_carbon_intensity(key: str) -> Optional[float]: def _start_cooldown() -> None: global _cooldown_until, _cooldown_duration - _cooldown_duration = min( - ELECTRICITYMAPS_COOLDOWN_MAX, - max(ELECTRICITYMAPS_COOLDOWN_MIN, _cooldown_duration * 2), - ) - _cooldown_until = time.monotonic() + _cooldown_duration + with _lock: + _cooldown_duration = min( + ELECTRICITYMAPS_COOLDOWN_MAX, + max(ELECTRICITYMAPS_COOLDOWN_MIN, _cooldown_duration * 2), + ) + _cooldown_until = time.monotonic() + _cooldown_duration def get_carbon_intensity( @@ -101,10 +112,12 @@ def get_carbon_intensity( ) return cached_carbon_intensity - if time.monotonic() < _cooldown_until: + with _lock: + cooldown_until = _cooldown_until + if time.monotonic() < cooldown_until: raise ElectricityMapsAPICooldownError( "Electricity Maps API is in cooldown after a previous failure, " - f"retrying in {_cooldown_until - time.monotonic():.0f} seconds" + f"retrying in {cooldown_until - time.monotonic():.0f} seconds" ) try: @@ -128,8 +141,9 @@ def get_carbon_intensity( _start_cooldown() raise - _cooldown_duration = 0.0 - _cache[key] = (time.monotonic(), carbon_intensity_g_per_kWh) + with _lock: + _cooldown_duration = 0.0 + _cache[key] = (time.monotonic(), carbon_intensity_g_per_kWh) return carbon_intensity_g_per_kWh diff --git a/tests/test_electricitymaps_cache.py b/tests/test_electricitymaps_cache.py index 537746fa8..6c433fb37 100644 --- a/tests/test_electricitymaps_cache.py +++ b/tests/test_electricitymaps_cache.py @@ -143,6 +143,17 @@ def test_cache_is_not_shared_between_tokens(self): # must not be served the value cached for the first one. assert electricitymaps_api.get_carbon_intensity(self._geo, "token-b") == 412.0 + @responses.activate + def test_cache_key_never_holds_the_raw_token(self): + self._add_success_response() + electricitymaps_api.get_carbon_intensity(self._geo, "super-secret-token") + + # THEN the secret is only present as a hash, so it cannot leak through + # the cache nor the debug log that renders the key. + keys = list(electricitymaps_api._cache) + assert len(keys) == 1 + assert "super-secret-token" not in keys[0] + @responses.activate def test_cooldown_does_not_log_one_error_per_call(self): responses.add( From 3319a9424370a78fe780d02f32f7268ff1a2a2d0 Mon Sep 17 00:00:00 2001 From: David Berenstein Date: Thu, 13 Aug 2026 07:21:05 +0200 Subject: [PATCH 5/5] fix: stop hashing the Electricity Maps token for the cache key CodeQL flags the SHA256 digest as py/weak-sensitive-data-hashing: the parameter name marks it as a credential, and any hashlib digest of a credential reads as an insecure password hash. The key only has to tell two tokens apart inside one process, so use the builtin randomly-seeded hash() instead. No caching, backoff or public API change. Co-Authored-By: Claude Opus 5 (1M context) --- codecarbon/core/electricitymaps_api.py | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/codecarbon/core/electricitymaps_api.py b/codecarbon/core/electricitymaps_api.py index 90d969153..57ff3668e 100644 --- a/codecarbon/core/electricitymaps_api.py +++ b/codecarbon/core/electricitymaps_api.py @@ -1,4 +1,3 @@ -import hashlib import threading import time from typing import Any, Dict, Optional, Tuple @@ -41,11 +40,13 @@ def reset_cache() -> None: def _cache_key(params: Dict[str, Any], electricitymaps_api_token: str) -> str: # The token is part of the key: two trackers in one process may use - # different tokens, and must not share a cached value. It is hashed so the - # raw secret is never held in the cache nor rendered in logs. + # different tokens, and must not share a cached value. Only an opaque, + # process-local marker is kept, so the raw secret is never held in the + # cache nor rendered in logs. builtin hash() is randomly seeded per + # process and is not a password digest: it is used to tell tokens apart, + # never to protect one. joined = ",".join(f"{key}={params[key]}" for key in sorted(params)) - token_digest = hashlib.sha256(electricitymaps_api_token.encode()).hexdigest()[:16] - return f"{joined},token={token_digest}" + return f"{joined},token={hash(electricitymaps_api_token):x}" def _get_cached_carbon_intensity(key: str) -> Optional[float]: