fix: validate document keys before they reach request URLs

arangodol/__init__.py

@@ -1,12 +1,50 @@
 """
 arango with a simple (dict-like or list-like) interface
 """
+import re
+
 from dol.base import Persister
 
 from pyArango.connection import Connection
 from pyArango.theExceptions import DocumentNotFoundError
 
 
+# ArangoDB's document-key alphabet, minus ``%``. pyArango places keys in request
+# URLs without encoding them and the server percent-decodes the path, so allowing
+# ``%`` would let encoded ``/`` or ``..`` address other collections or databases.
+# Do not add ``%`` back to match ArangoDB's documented alphabet.
+_VALID_KEY_CHARS = re.compile(r"[A-Za-z0-9_\-:.@()+,=;$!*']+")
+_MAX_KEY_BYTES = 254
+_DOT_SEGMENTS = {".", ".."}
+
+
+def validate_document_key(key):
+    """Return ``key`` if it is a safe ArangoDB document key, else raise ``KeyError``.
+
+    Keys end up in request URLs, so anything outside ArangoDB's key alphabet (such
+    as ``/``) or a dot segment could address a different collection or database.
+
+    >>> validate_document_key('Robot::0a1b')
+    'Robot::0a1b'
+    >>> validate_document_key('a/b')
+    Traceback (most recent call last):
+      ...
+    KeyError: "Invalid ArangoDB document key: 'a/b'"
+    >>> validate_document_key('..')
+    Traceback (most recent call last):
+      ...
+    KeyError: "Invalid ArangoDB document key: '..'"
+    """
+    if (
+        not isinstance(key, str)
+        or key in _DOT_SEGMENTS
+        or len(key.encode("utf-8")) > _MAX_KEY_BYTES
+        or not _VALID_KEY_CHARS.fullmatch(key)
+    ):
+        raise KeyError(f"Invalid ArangoDB document key: {key!r}")
+    return key
+
+
 class ArangoDbPersister(Persister):
     """
     A basic ArangoDB persister.
@@ -111,7 +149,7 @@ def _make_key(self, keys_dict):
         """
         key_values = [keys_dict[key_label] for key_label in self._key_fields]
         key_str = self._key_fields_separator.join(key_values)
-        return key_str
+        return validate_document_key(key_str)
 
     def _split_key(self, joined_key_str):
         """
@@ -143,11 +181,12 @@ def __getitem__(self, keys_dict):
         return data
 
     def __setitem__(self, keys_dict, values_dict):
+        key = self._make_key(keys_dict)  # validate before touching the backend
         try:
             doc = self.__fetchitem__(keys_dict)
         except KeyError:
             doc = self._collection.createDocument()
-            doc._key = self._make_key(keys_dict)
+            doc._key = key
 
         for k, v in values_dict.items():
             doc[k] = v

arangodol/tests/_live_server.py

@@ -0,0 +1,17 @@
+"""Detect whether the local ArangoDB test server is reachable (for skipping live tests)."""
+
+import socket
+from urllib.parse import urlparse
+
+DB_URL = "http://127.0.0.1:8529"
+DB_PASSWORD = "somepassword"
+_CONNECT_TIMEOUT_S = 1
+
+
+def server_is_up(url=DB_URL, *, timeout=_CONNECT_TIMEOUT_S):
+    parsed = urlparse(url)
+    try:
+        with socket.create_connection((parsed.hostname, parsed.port), timeout=timeout):
+            return True
+    except OSError:
+        return False

arangodol/tests/test_arangodb.py

@@ -7,13 +7,16 @@
 pytest tests/test_arangodb.py
 """
 
+import pytest
+
 from arangodol import ArangoDbPersister
 from arangodol import ArangoDbTupleKeyStore
 
+from ._live_server import DB_PASSWORD, DB_URL, server_is_up
 from .base_test import BasePersisterTest, BaseKeyTupleStoreTest
 
-DB_URL = "http://127.0.0.1:8529"
-DB_PASSWORD = "somepassword"
+if not server_is_up(DB_URL):
+    pytest.skip(f"No ArangoDB server at {DB_URL}", allow_module_level=True)
 
 
 class TestArangoDbPersister(BasePersisterTest):

arangodol/tests/test_keys.py

@@ -0,0 +1,64 @@
+"""Document keys are validated before they reach a request URL (no server needed)."""
+
+import pytest
+
+from arangodol import ArangoDbPersister, validate_document_key
+
+
+def _persister(key_fields=("first_name", "last_name"), separator="::"):
+    # Bypass __init__ (which connects to a server); only key handling is tested.
+    p = ArangoDbPersister.__new__(ArangoDbPersister)
+    p._key_fields = key_fields
+    p._key_fields_separator = separator
+    return p
+
+
+@pytest.mark.parametrize(
+    "key", ["abc", "Robot::0a1b", "a-b_c:d.e@f(g)+h,i=j;k$l!m*n'o", "x" * 254]
+)
+def test_valid_keys_pass(key):
+    assert validate_document_key(key) == key
+
+
+@pytest.mark.parametrize(
+    "key",
+    ["", ".", "..", "a/b", "../other", "a%2Fb", "a b", "a?b", "a#b", "x" * 255, 42],
+)
+def test_invalid_keys_raise_key_error(key):
+    with pytest.raises(KeyError):
+        validate_document_key(key)
+
+
+def test_make_key_validates_joined_key():
+    p = _persister()
+    assert p._make_key({"first_name": "a", "last_name": "b"}) == "a::b"
+    with pytest.raises(KeyError):
+        p._make_key({"first_name": "..", "last_name": "/x"})
+
+
+@pytest.mark.parametrize(
+    "key", ["%", "a\\b", "abc\n", "é", "a\x00", "a%2e%2e", "a%2F..%2Fx"]
+)
+def test_more_invalid_keys(key):
+    with pytest.raises(KeyError):
+        validate_document_key(key)
+
+
+class _BackendMustNotBeCalled:
+    def __getitem__(self, key):
+        raise AssertionError(f"backend reached with key {key!r}")
+
+    def createDocument(self, *args, **kwargs):
+        raise AssertionError("backend reached (createDocument)")
+
+
+def test_invalid_key_never_reaches_backend():
+    p = _persister(key_fields=("k",))
+    p._collection = _BackendMustNotBeCalled()
+    bad = {"k": "../x"}
+    with pytest.raises(KeyError):
+        p[bad]
+    with pytest.raises(KeyError):
+        del p[bad]
+    with pytest.raises(KeyError):
+        p[bad] = {"v": 1}

conftest.py

@@ -0,0 +1,20 @@
+"""Skip the doctests that need a live ArangoDB server when none is reachable."""
+
+import pytest
+from _pytest.doctest import DoctestItem
+
+from arangodol.tests._live_server import DB_URL, server_is_up
+
+_LIVE_SERVER_DOCTESTS = {
+    "arangodol.ArangoDbPersister",
+    "arangodol.ArangoDbTupleKeyStore",
+}
+
+
+def pytest_collection_modifyitems(config, items):
+    if server_is_up(DB_URL):
+        return
+    skip = pytest.mark.skip(reason=f"No ArangoDB server at {DB_URL}")
+    for item in items:
+        if isinstance(item, DoctestItem) and item.name in _LIVE_SERVER_DOCTESTS:
+            item.add_marker(skip)

pyproject.toml

@@ -139,6 +139,7 @@ coverage_report_format = [
     "xml",
 ]
 exclude_paths = [
+    "docsrc",
     "examples",
     "scrap",
 ]