flexmeasures.utils.secrets_utils

Functions

flexmeasures.utils.secrets_utils.apply_token_refresh_result(secrets: dict[str, Any] | None, namespace: str, result: TokenRefreshResult, *, encryptor: SecretsEncryptor | None = None) → dict[str, Any]

Apply a provider-specific token refresh result to a secrets dictionary.

If a token value is None, only its metadata is updated. This supports providers whose refresh operation extends an existing token instead of returning a replacement token.

Parameters:

  • secrets – Existing secrets dictionary, or None for a new one.

  • namespace – Top-level provider or strategy namespace to update.

  • result – Token values and metadata returned by a provider strategy.

  • encryptor – Optional encryptor; defaults to app configuration.

Returns:

Updated copy of the secrets dictionary.

flexmeasures.utils.secrets_utils.delete_secret(secrets: dict[str, Any] | None, path: str | tuple[str, ...] | list[str]) → dict[str, Any]

Return a copy of secrets without the value at path.

Empty dictionaries containing the deleted value are removed as well.

Parameters:

  • secrets – Existing secrets dictionary.

  • path – Dot-separated path or sequence of keys to remove.

Raises:

KeyError – If the path does not exist.

flexmeasures.utils.secrets_utils.derive_fernet_key(secret: str, *, salt: str = 'flexmeasures-secrets') → bytes

Derive a Fernet-compatible key from an arbitrary non-empty secret.

Parameters:

  • secret – Master secret from configuration.

  • salt – Context-specific salt for key derivation.

Returns:

URL-safe base64-encoded key accepted by Fernet.

flexmeasures.utils.secrets_utils.format_keyring_config_help(setting_name: str, *, purpose: str, filename: str | None = None, key_value_generator_python: str = 'import secrets; print(secrets.token_urlsafe(32))', setting_generator_python: str = 'import json, secrets; print(json.dumps({"1": secrets.token_urlsafe(32)}))') → str

Return instructions for configuring a dictionary-based secret setting.

Parameters:

  • setting_name – Name of the FlexMeasures configuration setting.

  • purpose – Short explanation of what the setting protects.

  • filename – Optional instance-path file where the setting can be stored.

  • key_value_generator_python – Python snippet which prints one secret.

  • setting_generator_python – Python snippet which prints the JSON setting.

flexmeasures.utils.secrets_utils.get_secret(secrets: dict[str, Any] | None, path: str | tuple[str, ...] | list[str], *, encryptor: SecretsEncryptor | None = None) → str

Decrypt and return a secret value from path.

Parameters:

  • secrets – Secrets dictionary containing encrypted envelopes.

  • path – Dot-separated path or sequence of keys to the encrypted value.

  • encryptor – Optional encryptor; defaults to app configuration.

Returns:

Decrypted plaintext value.

flexmeasures.utils.secrets_utils.get_secret_overview(secrets: dict[str, Any] | None) → list[SecretOverview]

Return safe information for listing stored secrets.

Parameters:

secrets – Secrets dictionary containing encrypted envelopes.

Returns:

Secret paths with optional expiry datetimes, without other metadata.

flexmeasures.utils.secrets_utils.get_secret_paths(secrets: dict[str, Any] | None) → list[str]

Return sorted dot-separated paths to encrypted secret values.

Parameters:

secrets – Secrets dictionary containing encrypted envelopes.

Returns:

Secret paths without ciphertext or metadata values.

flexmeasures.utils.secrets_utils.log_keyring_config_error_and_exit(app, setting_name: str, filename: str) → None

Log invalid keyring-file instructions and exit app startup.

Parameters:

  • app – Flask app whose logger should receive the message.

  • setting_name – Name of the FlexMeasures configuration setting.

  • filename – File path containing an invalid value.

flexmeasures.utils.secrets_utils.redact_secrets(secrets: dict[str, Any] | None) → dict[str, Any]

Return secrets metadata without ciphertext or plaintext values.

Parameters:

secrets – Secrets dictionary to redact.

Returns:

Copy with encrypted values replaced by safe metadata.

flexmeasures.utils.secrets_utils.set_secret(secrets: dict[str, Any] | None, path: str | tuple[str, ...] | list[str], value: str, *, encryptor: SecretsEncryptor | None = None, metadata: dict[str, Any] | None = None) → dict[str, Any]

Return a copy of secrets with an encrypted value set at path.

Parameters:

  • secrets – Existing secrets dictionary, or None for a new one.

  • path – Dot-separated path or sequence of keys where the value is stored.

  • value – Plaintext secret value to encrypt.

  • encryptor – Optional encryptor; defaults to app configuration.

  • metadata – Optional non-secret metadata to store with the envelope.

flexmeasures.utils.secrets_utils.set_secret_key(app, filename: str = 'secret_key') → None

Set the SECRET_KEY setting or exit app startup.

Parameters:

  • app – Flask app whose config should receive the setting.

  • filename – File name in the app instance path to check after config and environment values.

flexmeasures.utils.secrets_utils.set_totp_secrets(app, filename: str = 'totp_secrets') → None

Set the SECURITY_TOTP_SECRETS setting or exit app startup.

Parameters:

  • app – Flask app whose config should receive the setting.

  • filename – File name in the app instance path to check after config and environment values.

flexmeasures.utils.secrets_utils.store_account_secret(account: Account, path: str | tuple[str, ...] | list[str], value: str, *, metadata: dict[str, Any] | None = None, encryptor: SecretsEncryptor | None = None) → dict[str, Any]

Store an encrypted secret on an account and return its secrets dict.

Parameters:

  • account – Account whose secrets field should be updated.

  • path – Dot-separated path or sequence of keys where the value is stored.

  • value – Plaintext secret value to encrypt.

  • metadata – Optional non-secret metadata to store with the envelope.

  • encryptor – Optional encryptor; defaults to app configuration.

flexmeasures.utils.secrets_utils.store_asset_secret(asset: GenericAsset, path: str | tuple[str, ...] | list[str], value: str, *, metadata: dict[str, Any] | None = None, encryptor: SecretsEncryptor | None = None) → dict[str, Any]

Store an encrypted secret on an asset and return its secrets dict.

Parameters:

  • asset – Generic asset whose secrets field should be updated.

  • path – Dot-separated path or sequence of keys where the value is stored.

  • value – Plaintext secret value to encrypt.

  • metadata – Optional non-secret metadata to store with the envelope.

  • encryptor – Optional encryptor; defaults to app configuration.

Classes

class flexmeasures.utils.secrets_utils.SecretOverview
class flexmeasures.utils.secrets_utils.SecretsEncryptor(encryption_keys: dict[str, str], key_id: str = '')

Encrypt and decrypt connection secrets with the configured master key.

Parameters:

  • encryption_keys – Mapping from key IDs to master key material.

  • key_id – Identifier of the key used for new encryption. If empty, the latest key ID in encryption_keys is used.

__init__(encryption_keys: dict[str, str], key_id: str = '') → None
decrypt(envelope: dict[str, Any] | str) → str

Decrypt an encrypted envelope or raw Fernet token.

Parameters:

envelope – Dict with a ciphertext field, or a raw token.

Returns:

Decrypted plaintext value.

encrypt(value: str) → dict[str, Any]

Encrypt a string and return a JSON-serializable envelope.

Parameters:

value – Plaintext secret value to encrypt.

Returns:

Dict containing ciphertext, key ID and timestamps.

property fernet: Fernet

Fernet instance derived from the current master key.

classmethod from_current_app() → SecretsEncryptor

Create an encryptor from Flask configuration.

FLEXMEASURES_SECRETS_ENCRYPTION_KEYS must be configured before connection secrets can be stored or decrypted.

class flexmeasures.utils.secrets_utils.TokenRefreshResult(access_token: str | None = None, refresh_token: str | None = None, access_token_expires_at: datetime | None = None, refresh_token_expires_at: datetime | None = None, token_type: str | None = None, metadata: dict[str, ~typing.Any]=<factory>)

Provider-neutral result of refreshing token state.

Parameters:

  • access_token – New access token, if the provider returned one.

  • refresh_token – New refresh token, if it rotated.

  • access_token_expires_at – Expiry timestamp for the access token.

  • refresh_token_expires_at – Expiry timestamp for the refresh token.

  • token_type – Token type metadata, for example Bearer.

  • metadata – Provider-specific non-secret metadata to preserve.

__init__(access_token: str | None = None, refresh_token: str | None = None, access_token_expires_at: datetime | None = None, refresh_token_expires_at: datetime | None = None, token_type: str | None = None, metadata: dict[str, ~typing.Any]=<factory>) → None

Exceptions

exception flexmeasures.utils.secrets_utils.InvalidSecretsEncryptionKey

Raised when a usable encryption key cannot be loaded.

exception flexmeasures.utils.secrets_utils.SecretsDecryptionError

Raised when a value cannot be decrypted.

exception flexmeasures.utils.secrets_utils.SecretsError

Raised when secret handling fails.