[FSSDK-11901] Fix concurrency bug in cmab service #462
Conversation
![prisma-cloud-devsecops[bot]](https://avatars.githubusercontent.com/in/135857?s=60&v=4){kind=small}
optimizely/cmab/cmab_service.py
Outdated
| """Calculate the lock index for a given user and rule combination.""" | ||
| # Create a hash of user_id + rule_id for consistent lock selection | ||
| hash_input = f"{user_id}{rule_id}" | ||
| hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16) % NUM_LOCK_STRIPES |
There was a problem hiding this comment.
Weak cryptographic algorithm used
File: cmab_service.py | Checkov ID: CKV3_SAST_55
How To Fix
Using Cryptodome library with AES (recommended)
from Cryptodome.Cipher import AES
cipher = AES.new(YOUR_KEY)
Description
CWE: CWE-327: Use of a Broken or Risky Cryptographic Algorithm
OWASP: A02:2021-Cryptographic Failures
CWE-327: Use of a Broken or Risky Cryptographic AlgorithmOWASP:
A02:2021-Cryptographic FailuresUtilizing weak or deprecated cryptographic algorithms poses a threat by exposing sensitive information to unnecessary vulnerabilities. This policy pinpoints the uses of outdated or insecure cryptographic algorithms that are considered unsafe and thus not recommended.
When weak algorithms are in play, it can result in:
- Exposed sensitive information.
- Greater susceptibility to brute-force attacks.
- The potential for revealing system configurations or secrets.
In the observed codebase, cryptography is performed with algorithms that are deprecated or considered as risky. Leveraging these algorithms is a crucial security misconfiguration that can expose applications to threats.
Avoid practices like:
python
# Using Cryptodome library with DES (not recommended)
import Cryptodome.Cipher.DES as DES
cipher = DES.new(YOUR_KEY)
There was a problem hiding this comment.
| hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16) % NUM_LOCK_STRIPES | |
| from optimizely.lib import pymmh3 as mmh3 | |
| def _get_lock_index(self, user_id: str, rule_id: str) -> int: | |
| """Calculate the lock index for a given user and rule combination.""" | |
| hash_input = f"{user_id}{rule_id}" | |
| hash_value = mmh3.hash(hash_input, seed=0) & 0xFFFFFFFF # Convert to unsigned | |
| return hash_value % NUM_LOCK_STRIPES |
bucketer already uses mm3, so we can use it here. And mm3 should avoid prisma scanning...
| hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16) % NUM_LOCK_STRIPES | ||
| return hash_value | ||
|
|
||
| def get_decision(self, project_config: ProjectConfig, user_context: OptimizelyUserContext, |
There was a problem hiding this comment.
instead of indenting all code in this function, we can make this internal _get_decision(), and create another function that calls this internal with the lock
get_decision():
with lock:
return _get_decision()
optimizely/cmab/cmab_service.py
Outdated
| """Calculate the lock index for a given user and rule combination.""" | ||
| # Create a hash of user_id + rule_id for consistent lock selection | ||
| hash_input = f"{user_id}{rule_id}" | ||
| hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16) % NUM_LOCK_STRIPES |
There was a problem hiding this comment.
| hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16) % NUM_LOCK_STRIPES | |
| from optimizely.lib import pymmh3 as mmh3 | |
| def _get_lock_index(self, user_id: str, rule_id: str) -> int: | |
| """Calculate the lock index for a given user and rule combination.""" | |
| hash_input = f"{user_id}{rule_id}" | |
| hash_value = mmh3.hash(hash_input, seed=0) & 0xFFFFFFFF # Convert to unsigned | |
| return hash_value % NUM_LOCK_STRIPES |
bucketer already uses mm3, so we can use it here. And mm3 should avoid prisma scanning...
3 participants
Summary
Fixes concurrency bug in cmab service by introducing locking in order to ensure consistent result.
Test plan
Introduced unit tests
Issues
FSSDK-11901