7ae8c08b-27e7-4df6-bfb9-4892ae974ff7

Committed 17 Mar 2025 11:44PM UTC coverage: 86.954% (+0.02%) from 86.93%

Build # 7ae8c08b-27e7-4df6-bfb9-4892ae974ff7

Build Type

push

circleci

Committed by

web-flow

Commit Message

SNS: fix Filter Policy engine to not evaluate full complex payload (#12395)

Coverage Stats

8 of 8 new or added lines in 1 file covered. (100.0%)

27 existing lines in 8 files now uncovered.

62326 of 71677 relevant lines covered (86.95%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.67

/localstack-core/localstack/services/stepfunctions/asl/jsonata/jsonata.py

from __future__ import annotations

import json
import re
from pathlib import Path
from typing import Any, Callable, Final, Optional

import jpype
import jpype.imports

from localstack.services.stepfunctions.asl.utils.encoding import to_json_str
from localstack.services.stepfunctions.packages import jpype_jsonata_package
from localstack.utils.objects import singleton_factory

JSONataExpression = str
VariableReference = str
VariableDeclarations = str

_PATTERN_VARIABLE_REFERENCE: Final[re.Pattern] = re.compile(
    r"\$\$|\$[a-zA-Z0-9_$]+(?:\.[a-zA-Z0-9_][a-zA-Z0-9_$]*)*|\$"
)
_ILLEGAL_VARIABLE_REFERENCES: Final[set[str]] = {"$", "$$"}
_VARIABLE_REFERENCE_ASSIGNMENT_OPERATOR: Final[str] = ":="
_VARIABLE_REFERENCE_ASSIGNMENT_STOP_SYMBOL: Final[str] = ";"
_EXPRESSION_OPEN_SYMBOL: Final[str] = "("
_EXPRESSION_CLOSE_SYMBOL: Final[str] = ")"


class JSONataException(Exception):
    error: Final[str]
    details: Optional[str]

    def __init__(self, error: str, details: Optional[str]):
        self.error = error
        self.details = details


class _JSONataJVMBridge:
    _java_OBJECT_MAPPER: "com.fasterxml.jackson.databind.ObjectMapper"  # noqa
    _java_JSONATA: "com.dashjoin.jsonata.Jsonata.jsonata"  # noqa

    def __init__(self):
        installer = jpype_jsonata_package.get_installer()
        installer.install()

        from jpype import config as jpype_config

        jpype_config.destroy_jvm = False

        # Limitation: We can only start one JVM instance within LocalStack and using JPype for another purpose
        # (e.g., event-ruler) fails unless we change the way we load/reload the classpath.
        jvm_path = installer.get_java_lib_path()
        jsonata_libs_path = Path(installer.get_installed_dir())
        jsonata_libs_pattern = jsonata_libs_path.joinpath("*")
        jpype.startJVM(jvm_path, classpath=[jsonata_libs_pattern], interrupt=False)

        from com.fasterxml.jackson.databind import ObjectMapper  # noqa
        from com.dashjoin.jsonata.Jsonata import jsonata  # noqa

        self._java_OBJECT_MAPPER = ObjectMapper()
        self._java_JSONATA = jsonata

    @staticmethod
    @singleton_factory
    def get() -> _JSONataJVMBridge:
        return _JSONataJVMBridge()

    def eval_jsonata(self, jsonata_expression: JSONataExpression) -> Any:
        try:
            # Evaluate the JSONata expression with the JVM.
            # TODO: Investigate whether it is worth moving this chain of statements (java_*) to a
            #  Java program to reduce i/o between the JVM and this runtime.
            java_expression = self._java_JSONATA(jsonata_expression)
            java_output = java_expression.evaluate(None)
            java_output_string = self._java_OBJECT_MAPPER.writeValueAsString(java_output)

            # Compute a Python json object from the java string, this is to:
            #  1. Ensure we fully end interactions with the JVM about this value here;
            #  2. The output object may undergo under operations that are not compatible
            #     with jpype objects (such as json.dumps, equality, instanceof, etc.).
            result_str: str = str(java_output_string)
            result_json = json.loads(result_str)

            return result_json
        except Exception as ex:
            raise JSONataException("UNKNOWN", str(ex))


# Lazy initialization of the `eval_jsonata` function pointer.
# This ensures the JVM is only started when JSONata functionality is needed.
_eval_jsonata: Optional[Callable[[JSONataExpression], Any]] = None


def eval_jsonata_expression(jsonata_expression: JSONataExpression) -> Any:
    global _eval_jsonata
    if _eval_jsonata is None:
        # Initialize _eval_jsonata only when invoked for the first time using the Singleton pattern.
        _eval_jsonata = _JSONataJVMBridge.get().eval_jsonata
    return _eval_jsonata(jsonata_expression)


class IllegalJSONataVariableReference(ValueError):
    variable_reference: Final[VariableReference]

    def __init__(self, variable_reference: VariableReference):
        self.variable_reference = variable_reference


def extract_jsonata_variable_references(
    jsonata_expression: JSONataExpression,
) -> set[VariableReference]:
    if not jsonata_expression:
        return set()
    variable_references: list[VariableReference] = _PATTERN_VARIABLE_REFERENCE.findall(
        jsonata_expression
    )
    for variable_reference in variable_references:
        if variable_reference in _ILLEGAL_VARIABLE_REFERENCES:
            raise IllegalJSONataVariableReference(variable_reference=variable_reference)
    return set(variable_references)


def encode_jsonata_variable_declarations(
    bindings: dict[VariableReference, Any],
) -> VariableDeclarations:
    declarations_parts: list[str] = list()
    for variable_reference, value in bindings.items():
        if isinstance(value, str):
            value_str_lit = f'"{value}"'
        else:
            value_str_lit = to_json_str(value, separators=(",", ":"))
        declarations_parts.extend(
            [
                variable_reference,
                _VARIABLE_REFERENCE_ASSIGNMENT_OPERATOR,
                value_str_lit,
                _VARIABLE_REFERENCE_ASSIGNMENT_STOP_SYMBOL,
            ]
        )
    return "".join(declarations_parts)


def compose_jsonata_expression(
    final_jsonata_expression: JSONataExpression,
    variable_declarations_list: list[VariableDeclarations],
) -> JSONataExpression:
    variable_declarations = "".join(variable_declarations_list)
    expression = "".join(
        [
            _EXPRESSION_OPEN_SYMBOL,
            variable_declarations,
            final_jsonata_expression,
            _EXPRESSION_CLOSE_SYMBOL,
        ]
    )
    return expression

1	from __future__ import annotations	1✔
2
3	import json	1✔
4	import re	1✔
5	from pathlib import Path	1✔
6	from typing import Any, Callable, Final, Optional	1✔
7
8	import jpype	1✔
9	import jpype.imports	1✔
10
11	from localstack.services.stepfunctions.asl.utils.encoding import to_json_str	1✔
12	from localstack.services.stepfunctions.packages import jpype_jsonata_package	1✔
13	from localstack.utils.objects import singleton_factory	1✔
14
15	JSONataExpression = str	1✔
16	VariableReference = str	1✔
17	VariableDeclarations = str	1✔
18
19	_PATTERN_VARIABLE_REFERENCE: Final[re.Pattern] = re.compile(	1✔
20	r"\$\$\|\$[a-zA-Z0-9_$]+(?:\.[a-zA-Z0-9_][a-zA-Z0-9_$])\|\$"
21	)
22	_ILLEGAL_VARIABLE_REFERENCES: Final[set[str]] = {"$", "$$"}	1✔
23	_VARIABLE_REFERENCE_ASSIGNMENT_OPERATOR: Final[str] = ":="	1✔
24	_VARIABLE_REFERENCE_ASSIGNMENT_STOP_SYMBOL: Final[str] = ";"	1✔
25	_EXPRESSION_OPEN_SYMBOL: Final[str] = "("	1✔
26	_EXPRESSION_CLOSE_SYMBOL: Final[str] = ")"	1✔
27
28
29	class JSONataException(Exception):	1✔
30	error: Final[str]	1✔
31	details: Optional[str]	1✔
32
33	def __init__(self, error: str, details: Optional[str]):	1✔
34	self.error = error	×
35	self.details = details	×
36
37
38	class _JSONataJVMBridge:	1✔
39	_java_OBJECT_MAPPER: "com.fasterxml.jackson.databind.ObjectMapper" # noqa	1✔
40	_java_JSONATA: "com.dashjoin.jsonata.Jsonata.jsonata" # noqa	1✔
41
42	def __init__(self):	1✔
43	installer = jpype_jsonata_package.get_installer()	1✔
44	installer.install()	1✔
45
46	from jpype import config as jpype_config	1✔
47
48	jpype_config.destroy_jvm = False	1✔
49
50	# Limitation: We can only start one JVM instance within LocalStack and using JPype for another purpose
51	# (e.g., event-ruler) fails unless we change the way we load/reload the classpath.
52	jvm_path = installer.get_java_lib_path()	1✔
53	jsonata_libs_path = Path(installer.get_installed_dir())	1✔
54	jsonata_libs_pattern = jsonata_libs_path.joinpath("*")	1✔
55	jpype.startJVM(jvm_path, classpath=[jsonata_libs_pattern], interrupt=False)	1✔
56
57	from com.fasterxml.jackson.databind import ObjectMapper # noqa	1✔
58	from com.dashjoin.jsonata.Jsonata import jsonata # noqa	1✔
59
60	self._java_OBJECT_MAPPER = ObjectMapper()	1✔
61	self._java_JSONATA = jsonata	1✔
62
63	@staticmethod	1✔
64	@singleton_factory	1✔
65	def get() -> _JSONataJVMBridge:	1✔
66	return _JSONataJVMBridge()	1✔
67
68	def eval_jsonata(self, jsonata_expression: JSONataExpression) -> Any:	1✔
69	try:	1✔
70	# Evaluate the JSONata expression with the JVM.
71	# TODO: Investigate whether it is worth moving this chain of statements (java_*) to a
72	# Java program to reduce i/o between the JVM and this runtime.
73	java_expression = self._java_JSONATA(jsonata_expression)	1✔
74	java_output = java_expression.evaluate(None)	1✔
75	java_output_string = self._java_OBJECT_MAPPER.writeValueAsString(java_output)	1✔
76
77	# Compute a Python json object from the java string, this is to:
78	# 1. Ensure we fully end interactions with the JVM about this value here;
79	# 2. The output object may undergo under operations that are not compatible
80	# with jpype objects (such as json.dumps, equality, instanceof, etc.).
81	result_str: str = str(java_output_string)	1✔
82	result_json = json.loads(result_str)	1✔
83
84	return result_json	1✔
85	except Exception as ex:	×
86	raise JSONataException("UNKNOWN", str(ex))	×
87
88
89	# Lazy initialization of the `eval_jsonata` function pointer.
90	# This ensures the JVM is only started when JSONata functionality is needed.
91	_eval_jsonata: Optional[Callable[[JSONataExpression], Any]] = None	1✔
92
93
94	def eval_jsonata_expression(jsonata_expression: JSONataExpression) -> Any:	1✔
95	global _eval_jsonata
96	if _eval_jsonata is None:	1✔
97	# Initialize _eval_jsonata only when invoked for the first time using the Singleton pattern.
98	_eval_jsonata = _JSONataJVMBridge.get().eval_jsonata	1✔
99	return _eval_jsonata(jsonata_expression)	1✔
100
101
102	class IllegalJSONataVariableReference(ValueError):	1✔
103	variable_reference: Final[VariableReference]	1✔
104
105	def __init__(self, variable_reference: VariableReference):	1✔
UNCOV 106	self.variable_reference = variable_reference	×
107
108
109	def extract_jsonata_variable_references(	1✔
110	jsonata_expression: JSONataExpression,
111	) -> set[VariableReference]:
112	if not jsonata_expression:	1✔
UNCOV 113	return set()	×
114	variable_references: list[VariableReference] = _PATTERN_VARIABLE_REFERENCE.findall(	1✔
115	jsonata_expression
116	)
117	for variable_reference in variable_references:	1✔
118	if variable_reference in _ILLEGAL_VARIABLE_REFERENCES:	1✔
UNCOV 119	raise IllegalJSONataVariableReference(variable_reference=variable_reference)	×
120	return set(variable_references)	1✔
121
122
123	def encode_jsonata_variable_declarations(	1✔
124	bindings: dict[VariableReference, Any],
125	) -> VariableDeclarations:
126	declarations_parts: list[str] = list()	1✔
127	for variable_reference, value in bindings.items():	1✔
128	if isinstance(value, str):	1✔
129	value_str_lit = f'"{value}"'	1✔
130	else:
131	value_str_lit = to_json_str(value, separators=(",", ":"))	1✔
132	declarations_parts.extend(	1✔
133	[
134	variable_reference,
135	_VARIABLE_REFERENCE_ASSIGNMENT_OPERATOR,
136	value_str_lit,
137	_VARIABLE_REFERENCE_ASSIGNMENT_STOP_SYMBOL,
138	]
139	)
140	return "".join(declarations_parts)	1✔
141
142
143	def compose_jsonata_expression(	1✔
144	final_jsonata_expression: JSONataExpression,
145	variable_declarations_list: list[VariableDeclarations],
146	) -> JSONataExpression:
147	variable_declarations = "".join(variable_declarations_list)	1✔
148	expression = "".join(	1✔
149	[
150	_EXPRESSION_OPEN_SYMBOL,
151	variable_declarations,
152	final_jsonata_expression,
153	_EXPRESSION_CLOSE_SYMBOL,
154	]
155	)
156	return expression	1✔

localstack / localstack / 7ae8c08b-27e7-4df6-bfb9-4892ae974ff7

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous