From 85b132d6a1b5eaef89e71dd42b556391567b05d7 Mon Sep 17 00:00:00 2001
From: Sambhav Kothari <sambhavs.email@gmail.com>
Date: Thu, 28 Apr 2022 10:00:38 +0100
Subject: [PATCH] Fix incorrect JSONPatch paths when special characters are
 used

Signed-off-by: Sambhav Kothari <sambhavs.email@gmail.com>
---
 kopf/_cogs/structs/patches.py     | 14 +++++++++++---
 tests/admission/test_jsonpatch.py | 18 ++++++++++++++++++
 2 files changed, 29 insertions(+), 3 deletions(-)

diff --git a/kopf/_cogs/structs/patches.py b/kopf/_cogs/structs/patches.py
index 2a54ff26..decbdef2 100644
--- a/kopf/_cogs/structs/patches.py
+++ b/kopf/_cogs/structs/patches.py
@@ -18,6 +18,14 @@
 JSONPatchOp = Literal["add", "replace", "remove"]
 
 
+def _escaped_path(keys: List[str]) -> str:
+    """Provides an appropriately escaped path for JSON Patches.
+
+    See https://datatracker.ietf.org/doc/html/rfc6901#section-3 for more details.
+    """
+    return '/'.join(map(lambda key: key.replace('~', '~0').replace('/', '~1'), keys))
+
+
 class JSONPatchItem(TypedDict, total=False):
     op: JSONPatchOp
     path: str
@@ -91,14 +99,14 @@ def as_json_patch(self) -> JSONPatch:
     def _as_json_patch(self, value: object, keys: List[str]) -> JSONPatch:
         result: JSONPatch = []
         if value is None:
-            result.append(JSONPatchItem(op='remove', path='/'.join(keys)))
+            result.append(JSONPatchItem(op='remove', path=_escaped_path(keys)))
         elif len(keys) > 1 and self._original and not self._is_in_original_path(keys):
-            result.append(JSONPatchItem(op='add', path='/'.join(keys), value=value))
+            result.append(JSONPatchItem(op='add', path=_escaped_path(keys), value=value))
         elif isinstance(value, collections.abc.Mapping) and value:
             for key, val in value.items():
                 result.extend(self._as_json_patch(val, keys + [key]))
         else:
-            result.append(JSONPatchItem(op='replace', path='/'.join(keys), value=value))
+            result.append(JSONPatchItem(op='replace', path=_escaped_path(keys), value=value))
         return result
 
     def _is_in_original_path(self, keys: List[str]) -> bool:
diff --git a/tests/admission/test_jsonpatch.py b/tests/admission/test_jsonpatch.py
index 951d27cf..87b5430c 100644
--- a/tests/admission/test_jsonpatch.py
+++ b/tests/admission/test_jsonpatch.py
@@ -66,3 +66,21 @@ def test_removal_of_the_subkey():
     assert jsonpatch == [
         {'op': 'remove', 'path': '/xyz/abc'},
     ]
+
+
+def test_escaping_of_key():
+    patch = Patch()
+    patch['~xyz/test'] = {'abc': None}
+    jsonpatch = patch.as_json_patch()
+    assert jsonpatch == [
+        {'op': 'remove', 'path': '/~0xyz~1test/abc'}
+    ]
+
+
+def test_recursive_escape_of_key():
+    patch = Patch()
+    patch['x/y/~z'] = {'a/b/~0c': None}
+    jsonpatch = patch.as_json_patch()
+    assert jsonpatch == [
+        {'op': 'remove', 'path': '/x~1y~1~0z/a~1b~1~00c'},
+    ]