⚡️ Speed up function decrypt_api_key by 51% in PR #11376 (fix-encryption-key-logging)

[codeflash-ai]

⚡️ This pull request contains optimizations for PR #11376

If you approve this dependent PR, these changes will be merged into the original PR branch fix-encryption-key-logging.

This PR will be automatically closed if the original PR is merged.

---

📄 51% (0.51x) speedup for decrypt_api_key in src/backend/base/langflow/services/auth/utils.py

⏱️ Runtime : 7.43 milliseconds → 4.92 milliseconds (best of 12 runs)

📝 Explanation and details

The optimized code achieves a 51% speedup by introducing memoization via @lru_cache to avoid redundant Fernet object creation.

Key Optimization

The core change extracts Fernet creation into a cached helper function _fernet_from_secret():

@lru_cache(maxsize=32)
def _fernet_from_secret(secret_key: str) -> Fernet:
    valid_key = ensure_valid_key(secret_key)
    return Fernet(valid_key)

Why This Works

Looking at the line profiler data for get_fernet():

Original version (9.7ms total):

• ensure_valid_key(): 6.77ms (69.6%)
• Fernet() construction: 2.36ms (24.3%)

Optimized version (1.0ms total):

• Cache lookup in _fernet_from_secret(): 0.41ms (40.3%)

The cache eliminates ~90% of get_fernet() execution time by reusing previously created Fernet objects when called with the same secret key, rather than re-executing the expensive ensure_valid_key() (which involves random number generation, base64 encoding) and Fernet initialization on every call.

Impact on decrypt_api_key()

Since decrypt_api_key() calls get_fernet() at the start:

• Original: 10.76ms spent in get_fernet() (35.9% of function time)
• Optimized: 1.95ms spent in get_fernet() (9.3% of function time)

This reduces decrypt_api_key() from 29.9ms to 20.9ms overall.

Test Case Performance

The optimization particularly benefits workloads with:

• Repeated decryption calls with the same settings service (e.g., test_many_sequential_decryptions_stability with 200 iterations)
• Sequential API key operations in request-heavy scenarios where the same secret key is used repeatedly

For single-call scenarios, the speedup is minimal, but the cache has negligible overhead (simple dictionary lookup).

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	✅ 17 Passed
🌀 Generated Regression Tests	✅ 207 Passed
⏪ Replay Tests	🔘 None Found
🔎 Concolic Coverage Tests	🔘 None Found
📊 Tests Coverage	100.0%

⚙️ Click to see Existing Unit Tests

🌀 Click to see Generated Regression Tests

import base64
import random
# imports
import sys
import types
from types import SimpleNamespace

import pytest  # used for our unit tests
from cryptography.fernet import Fernet
from langflow.services.auth.utils import decrypt_api_key
from lfx.log.logger import logger
from lfx.services.settings.service import SettingsService


class _SecretWrapper:
    """Wrap the secret string to emulate get_secret_value() API."""

    def __init__(self, secret: str):
        self._secret = secret

    def get_secret_value(self):
        # Return the underlying secret string when requested.
        return self._secret


class SettingsService:
    """
    Minimal SettingsService compatible object as expected by the function under test.

    It provides an .auth_settings attribute whose .SECRET_KEY has a
    get_secret_value() method.
    """

    def __init__(self, secret: str):
        # The function uses settings_service.auth_settings.SECRET_KEY.get_secret_value()
        # so we emulate that structure here.
        self.auth_settings = SimpleNamespace(SECRET_KEY=_SecretWrapper(secret))

# ---------------------------------------------------------------------------
# Now include the original function and its helper definitions exactly as provided.
# We must preserve the implementation and signature without modification.
# ---------------------------------------------------------------------------


MINIMUM_KEY_LENGTH = 32


def ensure_valid_key(s: str) -> bytes:
    # If the key is too short, we'll use it as a seed to generate a valid key
    if len(s) < MINIMUM_KEY_LENGTH:
        # Use the input as a seed for the random number generator
        random.seed(s)
        # Generate 32 random bytes
        key = bytes(random.getrandbits(8) for _ in range(32))
        key = base64.urlsafe_b64encode(key)
    else:
        key = add_padding(s).encode()
    return key


def get_fernet(settings_service: SettingsService):
    secret_key: str = settings_service.auth_settings.SECRET_KEY.get_secret_value()
    valid_key = ensure_valid_key(secret_key)
    return Fernet(valid_key)
from langflow.services.auth.utils import decrypt_api_key

# ---------------------------------------------------------------------------
# unit tests
# ---------------------------------------------------------------------------

# Helper factory to create SettingsService instances for tests.
def make_settings(secret: str) -> SettingsService:
    # Instantiate the SettingsService class imported into our test environment.
    return SettingsService(secret)


def test_basic_encrypt_decrypt_roundtrip():
    # Basic functionality:
    # Use a short secret so ensure_valid_key() generates a deterministic key via random.seed
    settings = make_settings("short-secret")
    # Create a fernet using the same settings (should be deterministic)
    f = get_fernet(settings)
    plaintext = "my-plain-api-key"
    # Encrypt the plaintext using the fernet obtained from the same settings
    token = f.encrypt(plaintext.encode()).decode()
    # Now call the function under test to decrypt it using the same settings
    codeflash_output = decrypt_api_key(token, settings); result = codeflash_output


def test_plain_text_returned_when_not_fernet_token():
    # If the input is plain text and not a Fernet token, the function should return it unchanged.
    settings = make_settings("another-secret")
    plain_value = "this-is-not-encrypted"
    codeflash_output = decrypt_api_key(plain_value, settings); result = codeflash_output


def test_return_empty_string_for_token_encrypted_with_different_key():
    # Tokens that look like Fernet tokens (start with "gAAAAA") but cannot be decrypted
    # with the provided key should cause the function to return an empty string.
    settings_encrypt = make_settings("encrypt-secret")
    settings_decrypt = make_settings("different-secret")
    f_encrypt = get_fernet(settings_encrypt)
    plaintext = "sensitive-value"
    # Generate a token with one key
    token = f_encrypt.encrypt(plaintext.encode()).decode()
    # Attempt to decrypt with a different settings service => should return empty string
    codeflash_output = decrypt_api_key(token, settings_decrypt); result = codeflash_output


def test_unexpected_type_raises_value_error():
    # Non-string inputs should raise a ValueError per the implementation.
    settings = make_settings("short-secret")
    with pytest.raises(ValueError) as excinfo:
        decrypt_api_key(b"bytes-not-str", settings)  # bytes are invalid type here


def test_empty_string_input_behaves_as_plain_text():
    # An empty string is a string type; decryption attempts will fail and since it does
    # not start with 'gAAAAA' it should be returned as-is (i.e., empty string).
    settings = make_settings("short-secret")
    codeflash_output = decrypt_api_key("", settings); result = codeflash_output


def test_large_plaintext_encrypt_decrypt():
    # Large-scale example (within limits): encrypt and decrypt a large plaintext string.
    settings = make_settings("short-secret")
    f = get_fernet(settings)
    # Create a large plaintext under 1000 characters to test scalability
    large_plain = "A" * 900
    token = f.encrypt(large_plain.encode()).decode()
    codeflash_output = decrypt_api_key(token, settings); decrypted = codeflash_output


def test_many_sequential_decryptions_stability():
    # Stress the function with multiple sequential decryptions (below 1000 iterations).
    settings = make_settings("short-secret")
    f = get_fernet(settings)
    count = 200  # well under the 1000 limit
    tokens = [f.encrypt(f"val-{i}".encode()).decode() for i in range(count)]
    # Decrypt all tokens and verify correctness
    for i, token in enumerate(tokens):
        codeflash_output = decrypt_api_key(token, settings)


def test_strings_starting_with_gAAAAA_but_not_valid_token_return_empty():
    # If a string begins with "gAAAAA" but isn't decryptable with the current key,
    # the function treats it as an encrypted token with a different key and returns "".
    settings = make_settings("short-secret")
    # Construct a string that starts with the Fernet prefix but is not a valid token for our key.
    fake_token_like = "gAAAAA-this-is-just-text-not-a-valid-token"
    codeflash_output = decrypt_api_key(fake_token_like, settings); result = codeflash_output
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.
#------------------------------------------------
import base64
import random
from unittest.mock import MagicMock, Mock

# imports
import pytest
from cryptography.fernet import Fernet
from langflow.services.auth.utils import decrypt_api_key
from lfx.log.logger import logger
from lfx.services.settings.service import SettingsService

MINIMUM_KEY_LENGTH = 32


def ensure_valid_key(s: str) -> bytes:
    # If the key is too short, we'll use it as a seed to generate a valid key
    if len(s) < MINIMUM_KEY_LENGTH:
        # Use the input as a seed for the random number generator
        random.seed(s)
        # Generate 32 random bytes
        key = bytes(random.getrandbits(8) for _ in range(32))
        key = base64.urlsafe_b64encode(key)
    else:
        key = add_padding(s).encode()
    return key


def add_padding(s: str) -> str:
    # Helper function to add padding to make the string valid for base64
    padding_needed = len(s) % 4
    if padding_needed:
        s += '=' * (4 - padding_needed)
    return s


def get_fernet(settings_service: SettingsService):
    secret_key: str = settings_service.auth_settings.SECRET_KEY.get_secret_value()
    valid_key = ensure_valid_key(secret_key)
    return Fernet(valid_key)
from langflow.services.auth.utils import decrypt_api_key

To edit these changes git checkout codeflash/optimize-pr11376-2026-01-20T20.59.12 and push.

[coderabbitai]

Important

Review skipped

Bot user detected.

To trigger a single review, invoke the @coderabbitai review command.

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 34.53%. Comparing base (d90fcdf) to head (6f0775e).

❌ Your project check has failed because the head coverage (41.58%) is below the target coverage (60.00%). You can increase the head coverage or adjust the target coverage.

Additional details and impacted files

@@                     Coverage Diff                     @@
##           fix-encryption-key-logging   #11380   +/-   ##
===========================================================
  Coverage                       34.52%   34.53%           
===========================================================
  Files                            1415     1415           
  Lines                           67343    67347    +4     
  Branches                         9937     9937           
===========================================================
+ Hits                            23250    23257    +7     
+ Misses                          42863    42860    -3     
  Partials                         1230     1230           

Flag	Coverage Δ
backend	53.47% <100.00%> (+0.02%)	⬆️
lfx	41.58% <ø> (ø)

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
src/backend/base/langflow/services/auth/utils.py	65.92% <100.00%> (+0.10%)	⬆️

... and 2 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.