diff --git a/src/palace/manager/service/redis/escape.py b/src/palace/manager/service/redis/escape.py deleted file mode 100644 index 420d81cbb..000000000 --- a/src/palace/manager/service/redis/escape.py +++ /dev/null @@ -1,81 +0,0 @@ -from __future__ import annotations - -import json -from functools import cached_property - -from palace.manager.core.exceptions import PalaceValueError - - -class JsonPathEscapeMixin: - r""" - Mixin to provide methods for escaping and unescaping JsonPaths for use in Redis / ElastiCache. - - This is necessary because some characters in object keys are not handled well by AWS ElastiCache, - and other characters seem problematic in Redis. - - This mixin provides methods to escape and unescape these characters, so that they can be used in - object keys, and the keys can be queried via JSONPath without issue. - - In ElastiCache when ~ is used in a key, the key is never updated, despite returning a success. And - when a / is used in a key, the key is interpreted as a nested path, nesting a new key for every - slash in the path. This is not the behavior we want, so we need to escape these characters. - - In Redis, the \ character is used as an escape character, and the " character is used to denote - the end of a string for the JSONPath. This means that these characters need to be escaped as well. - - Characters are escaped by prefixing them with a backtick character, followed by a single character - from _MAPPING that represents the escaped character. The backtick character itself is escaped by - prefixing it with another backtick character. - """ - - _ESCAPE_CHAR = "`" - - _MAPPING = { - "/": "s", - "\\": "b", - '"': "'", - "~": "t", - } - - @cached_property - def _FORWARD_MAPPING(self) -> dict[str, str]: - mapping = {k: "".join((self._ESCAPE_CHAR, v)) for k, v in self._MAPPING.items()} - mapping[self._ESCAPE_CHAR] = "".join((self._ESCAPE_CHAR, self._ESCAPE_CHAR)) - return mapping - - @cached_property - def _REVERSE_MAPPING(self) -> dict[str, str]: - mapping = {v: k for k, v in self._MAPPING.items()} - mapping[self._ESCAPE_CHAR] = self._ESCAPE_CHAR - return mapping - - def _escape_path(self, path: str, elasticache: bool = False) -> str: - escaped = "".join([self._FORWARD_MAPPING.get(c, c) for c in path]) - if elasticache: - # As well as the simple escaping we have defined here, for ElastiCache we need to fully - # escape the path as if it were a JSON string. So we call json.dumps to do this. We - # strip the leading and trailing quotes from the result, as we only want the escaped - # string, not the quotes. - escaped = json.dumps(escaped)[1:-1] - return escaped - - def _unescape_path(self, path: str) -> str: - in_escape = False - unescaped = [] - for char in path: - if in_escape: - if char not in self._REVERSE_MAPPING: - raise PalaceValueError( - f"Invalid escape sequence '{self._ESCAPE_CHAR}{char}'" - ) - unescaped.append(self._REVERSE_MAPPING[char]) - in_escape = False - elif char == self._ESCAPE_CHAR: - in_escape = True - else: - unescaped.append(char) - - if in_escape: - raise PalaceValueError("Unterminated escape sequence.") - - return "".join(unescaped) diff --git a/tests/manager/service/redis/test_escape.py b/tests/manager/service/redis/test_escape.py deleted file mode 100644 index eae33fc99..000000000 --- a/tests/manager/service/redis/test_escape.py +++ /dev/null @@ -1,72 +0,0 @@ -import json -import re -import string - -import pytest - -from palace.manager.core.exceptions import PalaceValueError -from palace.manager.service.redis.escape import JsonPathEscapeMixin - - -class TestPathEscapeMixin: - @pytest.mark.parametrize( - "path", - [ - "", - "test", - string.printable, - "test/test1/?$xyz.abc", - "`", - "```", - "/~`\\", - "`\\~/``/", - "a", - "/", - "~", - " ", - '"', - "💣ü", - ], - ) - def test_escape_path(self, path: str) -> None: - # Test a round trip - escaper = JsonPathEscapeMixin() - escaped = escaper._escape_path(path) - unescaped = escaper._unescape_path(escaped) - assert unescaped == path - - # Test a round trip with ElastiCache escaping. The json.loads is done implicitly by ElastiCache, - # when using these strings in a JsonPath query. We add a json.loads here to simulate that. - escaped = escaper._escape_path(path, elasticache=True) - unescaped = escaper._unescape_path(json.loads(f'"{escaped}"')) - assert unescaped == path - - # Test that we can handle escaping the escaped path multiple times - escaped = path - for _ in range(10): - escaped = escaper._escape_path(escaped) - - unescaped = escaped - for _ in range(10): - unescaped = escaper._unescape_path(unescaped) - - assert unescaped == path - - def test_unescape(self) -> None: - escaper = JsonPathEscapeMixin() - assert escaper._unescape_path("") == "" - - with pytest.raises( - PalaceValueError, match=re.escape("Invalid escape sequence '`?'") - ): - escaper._unescape_path("test `?") - - with pytest.raises( - PalaceValueError, match=re.escape("Invalid escape sequence '` '") - ): - escaper._unescape_path("``` test") - - with pytest.raises( - PalaceValueError, match=re.escape("Unterminated escape sequence") - ): - escaper._unescape_path("`")