fix: strip keys on CRUD operations

folksonomy/api.py

@@ -95,6 +95,12 @@ async def get_current_user(token: str = Depends(oauth2_scheme)):
             return User(user_id=None)
 
 
+def sanitize_data(k, v):
+    """Some sanitization of data"""
+    k = k.strip()
+    return k, v
+
+
 def check_owner_user(user: User, owner, allow_anonymous=False):
     """
     Check authentication depending on current user and 'owner' of the data
@@ -231,6 +237,7 @@ async def product_stats(response: Response,
     The products list can be limited to some tags (k or k=v)
     """
     check_owner_user(user, owner, allow_anonymous=True)
+    k, v = sanitize_data(k, v)
     where, params = property_where(owner, k, v)
     cur, timing = await db.db_exec("""
         SELECT json_agg(j.j)::json FROM(
@@ -267,6 +274,7 @@ async def product_list(response: Response,
     if k == '':
         return JSONResponse(status_code=422, content={"detail": {"msg": "missing value for k"}})
     check_owner_user(user, owner, allow_anonymous=True)
+    k, v = sanitize_data(k, v)
     where, params = property_where(owner, k, v)
     cur, timing = await db.db_exec("""
         SELECT coalesce(json_agg(j.j)::json, '[]'::json) FROM(
@@ -315,7 +323,7 @@ async def product_tag(response: Response,
     - /product/xxx/key returns only the requested key
     - /product/xxx/key* returns the key and subkeys (key:subkey)
     """
-
+    k, v = sanitize_data(k, None)
     key = re.sub(r'[^a-z0-9_\:]', '', k)
     check_owner_user(user, owner, allow_anonymous=True)
     if k[-1:] == '*':
@@ -356,6 +364,7 @@ async def product_tag_list_versions(response: Response,
     """
 
     check_owner_user(user, owner, allow_anonymous=True)
+    k, v = sanitize_data(k, None)
     cur, timing = await db.db_exec(
         """
         SELECT json_agg(j)::json FROM(
@@ -451,6 +460,7 @@ async def product_tag_delete(response: Response,
     Delete a product tag
     """
     check_owner_user(user, owner, allow_anonymous=False)
+    k, v = sanitize_data(k, None)
     try:
         # Setting version to 0, this is seen as a reset, 
         # while maintaining history in folksonomy_versions

folksonomy/models.py

@@ -35,6 +35,8 @@ def product_check(cls, v):
     def key_check(cls, v):
         if not v:
             raise ValueError('k cannot be empty')
+        # strip the key
+        v = v.strip()
         if not re.fullmatch(re_key, v):
             raise ValueError('k must be alpha-numeric [a-z0-9_-:]')
         return v

tests/test_main.py

@@ -43,14 +43,23 @@
 sample_by_keys = {(s["product"], s["k"], s["version"]): s for s in SAMPLES}
 
 
+@pytest.fixture(autouse=True)
+def clean_db():
+    # clean database before running a test
+    asyncio.run(_clean_db())
+
+
+async def _clean_db():
+    async with db.transaction():
+        await clean_data()
+
 @pytest.fixture
-def with_sample():
+def with_sample(auth_tokens):
     asyncio.run(_with_sample())
 
 
 async def _with_sample():
     async with db.transaction():
-        await clean_data()
         await create_data(SAMPLES)
 
 
@@ -86,14 +95,23 @@ async def create_data(samples):
             product_tag.version = version
             req, params = db.update_product_tag_req(product_tag)
             await db.db_exec(req, params)
+
+
+@pytest.fixture
+def auth_tokens():
+    asyncio.run(_add_auth_tokens())
+
+
+async def _add_auth_tokens():
     # add a token to auth foo and bar
-    await db.db_exec(
-        """
-        INSERT INTO auth (user_id, token, last_use) VALUES
-        ('foo','foo__Utest-token',current_timestamp AT TIME ZONE 'GMT'),
-        ('bar','bar__Utest-token',current_timestamp AT TIME ZONE 'GMT')
-        """
-    )
+    async with db.transaction():
+        await db.db_exec(
+            """
+            INSERT INTO auth (user_id, token, last_use) VALUES
+            ('foo','foo__Utest-token',current_timestamp AT TIME ZONE 'GMT'),
+            ('bar','bar__Utest-token',current_timestamp AT TIME ZONE 'GMT')
+            """
+        )
 
 
 class DummyResponse:
@@ -291,6 +309,15 @@ def test_product_key(with_sample):
                 'product': '3701027900001', 'k': 'color', 'v': 'red', 'owner': '', 'version': 1, 'editor': 'foo', 'comment': ''
             }
 
+def test_key_stripped_on_get(with_sample):
+    with TestClient(app) as client:
+        response = client.get(f"/product/{BARCODE_1}/ color  ")
+        assert response.status_code == 200
+        data = response.json()
+        data.pop("last_edit")
+        assert data == {
+            'product': '3701027900001', 'k': 'color', 'v': 'red', 'owner': '', 'version': 1, 'editor': 'foo', 'comment': ''
+        }
 
 def test_product_key_missing(with_sample):
     with TestClient(app) as client:
@@ -473,6 +500,20 @@ async def test_post(with_sample):
         await check_tag(BARCODE_1, "a-1:b_2:c-3:d_4", v="test", version=1)
 
 
+@pytest.mark.asyncio
+async def test_product_key_stripped_on_post(auth_tokens):
+    with TestClient(app) as client:
+        headers = {"Authorization":  "Bearer foo__Utest-token"}
+        response = client.post("/product", headers=headers, json=
+            {"product": BARCODE_1, "version": 1, "k": " test_new2  ", "v": "test"})
+        assert response.status_code == 200, f'valid new entry should return 200, got {response.status_code} {response.text}'
+        # check created stripped
+        await check_tag(BARCODE_1, "test_new2", v="test", version=1)
+        # reachable:
+        response = client.get(f"/product/{BARCODE_1}/test_new2")
+        assert response.status_code == 200, f'getting stripped key should return 200, got {response.status_code} {response.text}'
+
+
 def test_put_invalid(with_sample):
     with TestClient(app) as client:
         headers = {"Authorization":  "Bearer foo__Utest-token"}