⚡️ Speed up function decrypt_api_key by 54% in PR #11397 (docs-mcp-header)

[codeflash-ai]

⚡️ This pull request contains optimizations for PR #11397

If you approve this dependent PR, these changes will be merged into the original PR branch docs-mcp-header.

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

---

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

⏱️ Runtime : 17.9 milliseconds → 11.6 milliseconds (best of 26 runs)

📝 Explanation and details

The optimization achieves a 54% speedup by introducing memoization to avoid redundant cryptographic operations.

Key Optimization: Caching Fernet Instance Creation

The optimization introduces @lru_cache(maxsize=128) on a new helper function _get_fernet_cached() that caches Fernet instances by secret key. This addresses the most expensive operations in the original code:

Original Performance Bottlenecks:

• ensure_valid_key(): 66.6% of get_fernet() runtime (16.8ms out of 25.3ms)
• Fernet() initialization: 21.3% of get_fernet() runtime (5.4ms)

How Caching Works:
When get_fernet() is called with the same secret key multiple times, _get_fernet_cached() returns the pre-computed Fernet instance from cache instead of:

1. Re-validating the key (which involves random number generation and base64 encoding for short keys)
2. Re-initializing the Fernet cipher (which involves cryptographic setup)

Performance Impact:
The line profiler shows get_fernet() dropped from 25.3ms to 4.5ms (82% faster). In decrypt_api_key(), the get_fernet() call dropped from 28.1ms (37.9% of function time) to 7.2ms (13.9% of function time).

Why This Works

In authentication systems, the same secret key is typically used repeatedly across many API key decryption operations. The test test_large_scale_batch_decryption_performance_and_correctness demonstrates this pattern: 500 decryptions using the same settings service benefit significantly from caching, as only the first call incurs the full initialization cost.

The lru_cache(maxsize=128) is appropriately sized - authentication systems rarely use more than a handful of different secret keys in a single process, so a cache of 128 entries provides excellent hit rates without excessive memory overhead.

Trade-offs

• Memory: Cached Fernet instances persist in memory, but 128 entries is negligible
• Thread-safety: lru_cache is thread-safe by default
• Correctness: All test cases pass, confirming the optimization doesn't change behavior

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	✅ 17 Passed
🌀 Generated Regression Tests	✅ 510 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

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:
    def __init__(self, value: str):
        self._value = value

    def get_secret_value(self):
        # returns the underlying secret string
        return self._value


class _AuthSettings:
    def __init__(self, secret_value: str):
        # attribute name SECRET_KEY is used by the function
        self.SECRET_KEY = _SecretWrapper(secret_value)


class SettingsService:
    """
    Minimal SettingsService compatible with the function's expectations.
    It provides .auth_settings.SECRET_KEY.get_secret_value() -> str
    """

    def __init__(self, secret_value: str):
        # store the provided secret in the expected nested structure
        self.auth_settings = _AuthSettings(secret_value)


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)


def test_roundtrip_decrypt_with_short_secret():
    # Basic functionality: encryption and decryption roundtrip using a short secret.
    # Use secret shorter than MINIMUM_KEY_LENGTH to exercise ensure_valid_key's seed generation.
    secret = "short_seed_31"  # length < 32 ensures ensure_valid_key does not call add_padding
    svc = SettingsService(secret)
    # get a Fernet using the same settings_service; encrypt a value
    f = get_fernet(svc)
    plaintext = "my_api_key_123"
    # encrypt returns bytes; decode to str because decrypt_api_key expects str input
    token = f.encrypt(plaintext.encode()).decode()
    # decrypt_api_key should return the original plaintext
    codeflash_output = decrypt_api_key(token, svc); result = codeflash_output


def test_plain_text_returned_when_not_fernet_and_not_token_like():
    # If input is plain text (not decryptable) and doesn't start with 'gAAAAA',
    # decrypt_api_key should return the string unchanged.
    secret = "another_short_secret"
    svc = SettingsService(secret)
    plain = "I am a plain string, not encrypted"
    # call the function; both decryption attempts will fail and startswith check is False
    codeflash_output = decrypt_api_key(plain, svc); result = codeflash_output



def test_non_string_input_raises_value_error():
    # Non-string inputs should raise a ValueError with the expected message.
    svc = SettingsService("short_secret")
    with pytest.raises(ValueError) as excinfo:
        # Passing bytes instead of str should trigger ValueError
        decrypt_api_key(b"bytes-not-str", svc)


def test_empty_string_input_returns_empty_string():
    # An empty string is valid input; decryption attempts fail and should result in returning ""
    # (plain text path) because startswith('gAAAAA') is False for empty string.
    svc = SettingsService("tiny_seed")
    codeflash_output = decrypt_api_key("", svc); result = codeflash_output


def test_same_short_seed_produces_compatible_fernet_keys():
    # ensure_valid_key with the same short seed produces same Fernet key deterministically.
    seed = "deterministic_seed"
    svc1 = SettingsService(seed)
    svc2 = SettingsService(seed)
    f1 = get_fernet(svc1)
    f2 = get_fernet(svc2)

    # Encrypt with f1 and decrypt with the function using svc2; should succeed.
    plaintext = "deterministic_test"
    token = f1.encrypt(plaintext.encode()).decode()
    codeflash_output = decrypt_api_key(token, svc2); result = codeflash_output


def test_large_scale_batch_decryption_performance_and_correctness():
    # Large Scale Test: create a moderate sized batch (500 elements) of encrypted tokens
    # encrypted with the same key, then decrypt each using decrypt_api_key and verify correctness.
    # This stays under the 1000-loop limit and reasonable memory usage.
    svc = SettingsService("batch_seed")
    f = get_fernet(svc)

    # prepare 500 plaintexts
    count = 500
    plaintexts = [f"batch_key_{i}" for i in range(count)]
    # encrypt all plaintexts using the same Fernet instance
    tokens = [f.encrypt(p.encode()).decode() for p in plaintexts]

    # decrypt them via decrypt_api_key and assert all values match
    for original, token in zip(plaintexts, tokens):
        codeflash_output = decrypt_api_key(token, svc); decrypted = codeflash_output


def test_decrypt_non_token_but_bad_data_returns_original():
    # If the input is not a token-like string (does not start with 'gAAAAA'),
    # even if decryption fails, the original value should be returned.
    svc = SettingsService("short_seed_for_bad_data")
    bad_string = "this_is_not_a_token_but_may_fail_decryption"
    codeflash_output = decrypt_api_key(bad_string, svc); 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
from unittest.mock import MagicMock, Mock

import pytest
from cryptography.fernet import Fernet
from langflow.services.auth.utils import (decrypt_api_key, ensure_valid_key,
                                          get_fernet)




def test_decrypt_non_string_input_raises_error():
    """Test that non-string inputs raise ValueError."""
    # Setup: Create a valid settings service mock
    settings_service = Mock()
    settings_service.auth_settings.SECRET_KEY.get_secret_value.return_value = "test_key"
    
    # Test: Pass a non-string input
    with pytest.raises(ValueError) as excinfo:
        decrypt_api_key(123, settings_service)






def test_decrypt_with_bytes_input_raises_error():
    """Test that bytes input raises ValueError."""
    # Setup: Create a valid settings service mock
    settings_service = Mock()
    settings_service.auth_settings.SECRET_KEY.get_secret_value.return_value = "test_key"
    
    # Test: Pass bytes input
    with pytest.raises(ValueError) as excinfo:
        decrypt_api_key(b"encrypted_key", settings_service)


def test_decrypt_with_none_input_raises_error():
    """Test that None input raises ValueError."""
    # Setup: Create a valid settings service mock
    settings_service = Mock()
    settings_service.auth_settings.SECRET_KEY.get_secret_value.return_value = "test_key"
    
    # Test: Pass None input
    with pytest.raises(ValueError) as excinfo:
        decrypt_api_key(None, settings_service)





def test_decrypt_with_short_secret_key():
    """Test decryption with a short secret key that triggers key generation."""
    # Setup: Create a valid settings service mock with a short secret key
    settings_service = Mock()
    # Use a key shorter than MINIMUM_KEY_LENGTH (32)
    short_secret_key = "short_key"
    settings_service.auth_settings.SECRET_KEY.get_secret_value.return_value = short_secret_key
    
    # Create a Fernet instance using the short key (which will be expanded)
    valid_key = ensure_valid_key(short_secret_key)
    fernet = Fernet(valid_key)
    
    # Encrypt a test API key
    test_api_key = "test-api-key-with-short-secret"
    encrypted_key = fernet.encrypt(test_api_key.encode()).decode()
    
    # Test: Decrypt the encrypted key
    codeflash_output = decrypt_api_key(encrypted_key, settings_service); result = codeflash_output

To edit these changes git checkout codeflash/optimize-pr11397-2026-01-21T19.49.29 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.
⚠️ Please upload report for BASE (docs-1.8-release@771a552). Learn more about missing BASE report.

Additional details and impacted files

@@                 Coverage Diff                 @@
##             docs-1.8-release   #11398   +/-   ##
===================================================
  Coverage                    ?   34.56%           
===================================================
  Files                       ?     1416           
  Lines                       ?    67428           
  Branches                    ?     9931           
===================================================
  Hits                        ?    23309           
  Misses                      ?    42895           
  Partials                    ?     1224           

Flag	Coverage Δ
backend	53.50% <100.00%> (?)
lfx	41.63% <ø> (?)

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	66.66% <100.00%> (ø)

🚀 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.

[ogabrielluiz]

Closing automated codeflash PR.