Patched code sample

import re
import operator
import ast

# The vulnerability CVE-2024-26910 (misidentified in the prompt as a future CVE)
# was caused by using `eval()` on a user-provided filter string. This allowed
# for arbitrary code execution.
#
# The code below demonstrates the principle of the fix: replacing the dangerous
# `eval()` call with a safe, purpose-built parser that only understands a
# specific, non-executable filter grammar. This prevents any arbitrary code
# from being executed.

class FixedInMemoryVectorStore:
    """
    A simplified representation of an in-memory vector store demonstrating
    the vulnerability fix. The core of the fix lies in the
    `_parse_and_apply_filter` method, which safely parses filter strings
    instead of evaluating them as executable code.
    """

    def __init__(self):
        """Initializes the store with sample data."""
        self._storage = {
            "doc1": {"id": "doc1", "metadata": {"category": "A", "year": 2022, "published": True}},
            "doc2": {"id": "doc2", "metadata": {"category": "B", "year": 2023, "published": True}},
            "doc3": {"id": "doc3", "metadata": {"category": "A", "year": 2023, "published": False}},
        }
        # A dictionary mapping string operators to safe functions from the operator module.
        self._safe_operators = {
            "==": operator.eq,
            "!=": operator.ne,
            ">": operator.gt,
            "<": operator.lt,
            ">=": operator.ge,
            "<=": operator.le,
        }

    def _parse_and_apply_filter(self, record: dict, filter_str: str) -> bool:
        """
        This method represents the core of the security fix.
        It safely parses a simple 'key op value' filter string and applies it.
        It explicitly AVOIDS using `eval()`.
        """
        # A regular expression to safely parse an expected filter format, e.g.,
        # "year >= 2023" or "category == 'A'". It captures the metadata key,
        # the operator, and the value as separate groups.
        pattern = r"^\s*(?P<key>\w+)\s*(?P<op>==|!=|>=|<=|>|<)\s*(?P<val>.+?)\s*$"
        match = re.match(pattern, filter_str)

        if not match:
            # If the filter does not match the safe pattern, it is rejected.
            # This is where a malicious payload like "__import__('os').system(...)"
            # would be caught and discarded.
            raise ValueError(f"Invalid or unsafe filter format provided: {filter_str}")

        parts = match.groupdict()
        key = parts["key"]
        op_str = parts["op"]
        value_str = parts["val"]

        metadata = record.get("metadata", {})

        if key not in metadata:
            return False  # The record does not have the required metadata key.

        record_value = metadata[key]

        # Safely convert the string value from the filter to a Python literal
        # (e.g., '2023' -> 2023, "'A'" -> "A", 'True' -> True).
        # `ast.literal_eval` is a safe alternative to `eval()` that only
        # processes literals and not variables, functions, or expressions.
        try:
            compare_value = ast.literal_eval(value_str)
        except (ValueError, SyntaxError):
            # If it's not a valid Python literal, it's a syntax error.
            raise ValueError(f"Invalid value in filter: {value_str}")

        # Retrieve the safe comparison function from the predefined dictionary.
        op_func = self._safe_operators[op_str]

        # Perform the comparison using the safe operator.
        return op_func(record_value, compare_value)

    def search(self, filters: list[str]) -> list[dict]:
        """
        Searches for records that match all provided filters.
        """
        results = []
        for record in self._storage.values():
            try:
                # The record is included only if it passes all the safe filter checks.
                if all(self._parse_and_apply_filter(record, f) for f in filters):
                    results.append(record)
            except ValueError:
                # If any filter is malformed or unsafe, we can log it and continue,
                # ensuring the malicious input does not halt the system or execute.
                # For this example, we simply skip the record.
                continue
        return results