Repository: lektor/lektor
Pull request #1182 (2024-02-27 18:14:07)

Better API parameter validation (3.3 branch)
<!---
Remember to please check that your code passes tests locally. Test with the Makefile
commands `make test`, or if you only need to test just Python / JS `make test-python`
or `make test-js`.
--->

Backport #1181 to 3.3 branch.



Patch:
diff --git a/lektor/admin/modules/api.py b/lektor/admin/modules/api.py
index 0ee9099..ecdee07 100644
--- a/lektor/admin/modules/api.py
+++ b/lektor/admin/modules/api.py
@@ -1,5 +1,6 @@
 import os
 import posixpath
+from typing import Optional
 
 import click
 from flask import Blueprint
@@ -12,12 +13,39 @@ from lektor.admin.utils import eventstream
 from lektor.constants import PRIMARY_ALT
 from lektor.publisher import publish
 from lektor.publisher import PublishError
+from lektor.utils import cleanup_path
 from lektor.utils import is_valid_id
 
 
 bp = Blueprint("api", __name__, url_prefix="/admin/api")
 
 
+class ValidationError(Exception):
+    """Invalid input."""
+
+
+@bp.errorhandler(ValidationError)
+def validation_error_handler(error):
+    message = str(error)
+    return message, 400
+
+
+def _validate_path(value: str) -> str:
+    path = cleanup_path(value)
+    if path != value:
+        raise ValidationError("Invalid path")
+    return path
+
+
+def _validate_alt(value: Optional[str]) -> str:
+    if value is None:
+        return PRIMARY_ALT
+    lektor_config = g.admin_context.pad.config
+    if not lektor_config.is_valid_alternative(value):
+        raise ValidationError("Invalid alt")
+    return value
+
+
 def get_record_and_parent(path):
     pad = g.admin_context.pad
     record = pad.get(path)
@@ -31,8 +59,9 @@ def get_record_and_parent(path):
 @bp.route("/pathinfo")
 def get_path_info():
     """Returns the path segment information for a record."""
-    alt = request.args.get("alt") or PRIMARY_ALT
-    tree_item = g.admin_context.tree.get(request.args["path"])
+    path = _validate_path(request.args["path"])
+    alt = _validate_alt(request.args.get("alt"))
+    tree_item = g.admin_context.tree.get(path)
     segments = []
 
     while tree_item is not None:
@@ -53,8 +82,8 @@ def get_path_info():
 
 @bp.route("/recordinfo")
 def get_record_info():
-    request_path = request.args["path"]
-    alt = request.args.get("alt") or PRIMARY_ALT
+    request_path = _validate_path(request.args["path"])
+    alt = _validate_alt(request.args.get("alt"))
     tree_item = g.admin_context.tree.get(request_path)
 
     return jsonify(
@@ -99,8 +128,9 @@ def get_record_info():
 
 @bp.route("/previewinfo")
 def get_preview_info():
-    alt = request.args.get("alt") or PRIMARY_ALT
-    record = g.admin_context.pad.get(request.args["path"], alt=alt)
+    path = _validate_path(request.args["path"])
+    alt = _validate_alt(request.args.get("alt"))
+    record = g.admin_context.pad.get(path, alt=alt)
     if record is None:
         return jsonify(exists=False, url=None, is_hidden=True)
     return jsonify(exists=True, url=record.url_path, is_hidden=record.is_hidden)
@@ -108,7 +138,7 @@ def get_preview_info():
 
 @bp.route("/find", methods=["POST"])
 def find():
-    alt = request.values.get("alt") or PRIMARY_ALT
+    alt = _validate_alt(request.args.get("alt"))
     lang = request.values.get("lang") or g.admin_context.info.ui_lang
     q = request.values.get("q")
     builder = current_app.lektor_info.get_builder()
@@ -117,8 +147,9 @@ def find():
 
 @bp.route("/browsefs", methods=["POST"])
 def browsefs():
-    alt = request.values.get("alt") or PRIMARY_ALT
-    record = g.admin_context.pad.get(request.values["path"], alt=alt)
+    path = _validate_path(request.args["path"])
+    alt = _validate_alt(request.args.get("alt"))
+    record = g.admin_context.pad.get(path, alt=alt)
     okay = False
     if record is not None:
         if record.is_attachment:
@@ -143,16 +174,18 @@ def match_url():
 
 @bp.route("/rawrecord")
 def get_raw_record():
-    alt = request.args.get("alt") or PRIMARY_ALT
-    ts = g.admin_context.tree.edit(request.args["path"], alt=alt)
+    path = _validate_path(request.args["path"])
+    alt = _validate_alt(request.args.get("alt"))
+    ts = g.admin_context.tree.edit(path, alt=alt)
     return jsonify(ts.to_json())
 
 
 @bp.route("/newrecord")
 def get_new_record_info():
+    path = _validate_path(request.args["path"])
+    alt = _validate_alt(request.args.get("alt"))
     pad = g.admin_context.pad
-    alt = request.args.get("alt") or PRIMARY_ALT
-    tree_item = g.admin_context.tree.get(request.args["path"])
+    tree_item = g.admin_context.tree.get(path)
 
     def describe_model(model):
         primary_field = None
@@ -186,8 +219,9 @@ def get_new_record_info():
 
 @bp.route("/newattachment")
 def get_new_attachment_info():
-    alt = request.args.get("alt") or PRIMARY_ALT
-    tree_item = g.admin_context.tree.get(request.args["path"])
+    path = _validate_path(request.args["path"])
+    alt = _validate_alt(request.args.get("alt"))
+    tree_item = g.admin_context.tree.get(path)
     label_i18n = tree_item.get_record_label_i18n(alt)
     return jsonify(
         {
@@ -200,8 +234,9 @@ def get_new_attachment_info():
 
 @bp.route("/newattachment", methods=["POST"])
 def upload_new_attachments():
-    alt = request.values.get("alt") or PRIMARY_ALT
-    ts = g.admin_context.tree.edit(request.values["path"], alt=alt)
+    path = _validate_path(request.values["path"])
+    alt = _validate_alt(request.values.get("alt"))
+    ts = g.admin_context.tree.edit(path, alt=alt)
     if not ts.exists or ts.is_attachment:
         return jsonify({"bad_upload": True})
 
@@ -227,13 +262,14 @@ def upload_new_attachments():
 @bp.route("/newrecord", methods=["POST"])
 def add_new_record():
     values = request.get_json()
-    alt = values.get("alt") or PRIMARY_ALT
+    request_path = _validate_path(values["path"])
+    alt = _validate_alt(values.get("alt"))
     exists = False
 
     if not is_valid_id(values["id"]):
         return jsonify(valid_id=False, exists=False, path=None)
 
-    path = posixpath.join(values["path"], values["id"])
+    path = posixpath.join(request_path, values["id"])
 
     ts = g.admin_context.tree.edit(path, datamodel=values.get("model"), alt=alt)
     with ts:
@@ -247,10 +283,11 @@ def add_new_record():
 
 @bp.route("/deleterecord", methods=["POST"])
 def delete_record():
-    alt = request.values.get("alt") or PRIMARY_ALT
+    path = _validate_path(request.values["path"])
+    alt = _validate_alt(request.values.get("alt"))
     delete_master = request.values.get("delete_master") == "1"
-    if request.values["path"] != "/":
-        ts = g.admin_context.tree.edit(request.values["path"], alt=alt)
+    if path != "/":
+        ts = g.admin_context.tree.edit(path, alt=alt)
         with ts:
             ts.delete(delete_master=delete_master)
     return jsonify(okay=True)
@@ -259,9 +296,10 @@ def delete_record():
 @bp.route("/rawrecord", methods=["PUT"])
 def update_raw_record():
     values = request.get_json()
+    path = _validate_path(values["path"])
+    alt = _validate_alt(values.get("alt"))
     data = values["data"]
-    alt = values.get("alt") or PRIMARY_ALT
-    ts = g.admin_context.tree.edit(values["path"], alt=alt)
+    ts = g.admin_context.tree.edit(path, alt=alt)
     with ts:
         ts.data.update(data)
     return jsonify(path=ts.path)
diff --git a/lektor/editor.py b/lektor/editor.py
index 851d724..e29b8b6 100644
--- a/lektor/editor.py
+++ b/lektor/editor.py
@@ -15,8 +15,10 @@ from itertools import chain
 from lektor.constants import PRIMARY_ALT
 from lektor.metaformat import serialize
 from lektor.utils import atomic_open
+from lektor.utils import cleanup_path
 from lektor.utils import increment_filename
 from lektor.utils import is_valid_id
+from lektor.utils import parse_path
 from lektor.utils import secure_filename
 
 
@@ -32,8 +34,19 @@ class BadDelete(BadEdit):
     pass
 
 
+def _is_valid_path(path: str) -> bool:
+    split_path = path.strip("/").split("/")
+    if split_path == [""]:
+        split_path = []
+    return parse_path(path) == split_path
+
+
 def make_editor_session(pad, path, is_attachment=None, alt=PRIMARY_ALT, datamodel=None):
     """Creates an editor session for the given path object."""
+    if not _is_valid_path(path):
+        raise BadEdit("Invalid path")
+    path = cleanup_path(path)
+
     if alt != PRIMARY_ALT and not pad.db.config.is_valid_alternative(alt):
         raise BadEdit("Attempted to edit an invalid alternative (%s)" % alt)
 
diff --git a/lektor/utils.py b/lektor/utils.py
index e343c1f..5aff777 100644
--- a/lektor/utils.py
+++ b/lektor/utils.py
@@ -33,7 +33,7 @@ is_windows = os.name == "nt"
 
 _slash_escape = "\\/" not in json.dumps("/")
 
-_slashes_re = re.compile(r"(/\.{1,2}(/|$))|/")
+_slashes_re = re.compile(r"(?:/(?:\.{1,2}(?=/|$))?)+")
 _last_num_re = re.compile(r"^(.*)(\d+)(.*?)$")
 _list_marker = object()
 _value_marker = object()
@@ -109,10 +109,11 @@ def is_path_child_of(a, b, strict=True):
 
 
 def untrusted_to_os_path(path):
-    path = path.strip("/").replace("/", os.path.sep)
     if not isinstance(path, str):
         path = path.decode(fs_enc, "replace")
-    return path
+    clean_path = cleanup_path(path)
+    assert clean_path.startswith("/")
+    return clean_path[1:].replace("/", os.path.sep)
 
 
 def is_path(path):


Test patch:
diff --git a/tests/test_api.py b/tests/test_api.py
index dae8a29..9bf8f19 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -1,6 +1,7 @@
 import json
 import os
 from operator import itemgetter
+from pathlib import Path
 
 import pytest
 
@@ -86,6 +87,23 @@ def test_recordinfo_children_sort_limited_alts(project, env):
     assert list(sorted(child_data, key=itemgetter("label"))) == child_data
 
 
+def test_newrecord(scratch_project, scratch_env):
+    params = {"path": "/", "id": "new", "data": {}}
+    webadmin = WebAdmin(scratch_env, output_path=scratch_project.tree)
+    resp = webadmin.test_client().post("/admin/api/newrecord", json=params)
+    assert resp.status_code == 200
+    assert resp.get_json() == {"valid_id": True, "exists": False, "path": "/new"}
+    assert Path(scratch_project.tree, "content", "new", "contents.lr").exists()
+
+
+def test_newrecord_bad_path(scratch_project, scratch_env):
+    params = {"path": "/../../templates", "id": "", "data": {}}
+    webadmin = WebAdmin(scratch_env, output_path=scratch_project.tree)
+    resp = webadmin.test_client().post("/admin/api/newrecord", json=params)
+    assert resp.status_code == 400
+    assert resp.get_data(as_text=True) == "Invalid path"
+
+
 def test_eventstream_yield_bytes():
     count = 0
 
diff --git a/tests/test_editor.py b/tests/test_editor.py
index 200b598..3f2f2c4 100644
--- a/tests/test_editor.py
+++ b/tests/test_editor.py
@@ -43,6 +43,7 @@ def test_make_editor_session(pad, path, kwargs, expect):
             "conflicting",
             marks=pytest.mark.xfail(reason="buglet that should be fixed"),
         ),
+        ("/../../templates", {}, "Invalid path"),
     ],
 )
 def test_make_editor_session_raises_bad_edit(pad, path, kwargs, expect):
diff --git a/tests/test_utils.py b/tests/test_utils.py
index bdae109..24ddfce 100644
--- a/tests/test_utils.py
+++ b/tests/test_utils.py
@@ -1,10 +1,12 @@
 # coding: utf-8
+import os
 from dataclasses import dataclass
 from urllib.parse import urlsplit
 
 import pytest
 
 from lektor.utils import build_url
+from lektor.utils import cleanup_path
 from lektor.utils import is_path_child_of
 from lektor.utils import join_path
 from lektor.utils import magic_split_ext
@@ -12,6 +14,7 @@ from lektor.utils import make_relative_url
 from lektor.utils import parse_path
 from lektor.utils import secure_url
 from lektor.utils import slugify
+from lektor.utils import untrusted_to_os_path
 from lektor.utils import Url
 
 
@@ -229,3 +232,30 @@ def test_make_relative_url(source, target, expected):
 def test_make_relative_url_relative_source_absolute_target():
     with pytest.raises(ValueError):
         make_relative_url("rel/a/tive/", "/abs/o/lute")
+
+
+@pytest.mark.parametrize(
+    "db_path, expected",
+    [
+        ("a/b", "/a/b"),
+        ("//a//./b//", "/a/b"),
+        ("//a//../b//", "/a/b"),
+        ("//a//..x/b//", "/a/..x/b"),
+    ],
+)
+def test_cleanup_path(db_path, expected):
+    assert cleanup_path(db_path) == expected
+
+
+@pytest.mark.parametrize(
+    "db_path, expected",
+    [
+        ("a/b", "a/b"),
+        ("/a/b", "a/b"),
+        ("a/b/", "a/b"),
+        ("/../../a", "a"),
+    ],
+)
+def test_untrusted_to_os_path(db_path, expected):
+    os_path = untrusted_to_os_path(db_path)
+    assert os_path.split(os.sep) == expected.split("/")