Fix the management of paths
This commit is contained in:
parent
2dfde5a7da
commit
12ddd64884
@ -37,7 +37,7 @@ import re
|
||||
from http import client
|
||||
from urllib.parse import unquote, urlparse
|
||||
|
||||
from . import auth, config, ical, log, pathutils, rights, storage, xmlutils
|
||||
from . import auth, config, ical, log, rights, storage, xmlutils
|
||||
|
||||
|
||||
VERSION = "2.0.0-pre"
|
||||
@ -167,7 +167,7 @@ class Application(object):
|
||||
def sanitize_uri(uri):
|
||||
"""Unquote and make absolute to prevent access to other data."""
|
||||
uri = unquote(uri)
|
||||
return pathutils.sanitize_path(uri)
|
||||
return ical.sanitize_path(uri)
|
||||
|
||||
def collect_allowed_items(self, items, user):
|
||||
"""Get items from request that user is allowed to access."""
|
||||
|
@ -23,14 +23,13 @@ Define the main classes of a collection as seen from the server.
|
||||
|
||||
"""
|
||||
|
||||
import os
|
||||
import hashlib
|
||||
import os
|
||||
import posixpath
|
||||
import re
|
||||
from uuid import uuid4
|
||||
from random import randint
|
||||
from contextlib import contextmanager
|
||||
|
||||
from . import pathutils
|
||||
from random import randint
|
||||
from uuid import uuid4
|
||||
|
||||
|
||||
def serialize(tag, headers=(), items=()):
|
||||
@ -52,6 +51,32 @@ def serialize(tag, headers=(), items=()):
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def sanitize_path(path):
|
||||
"""Make path absolute with leading slash to prevent access to other data.
|
||||
|
||||
Preserve a potential trailing slash.
|
||||
|
||||
"""
|
||||
trailing_slash = "/" if path.endswith("/") else ""
|
||||
path = posixpath.normpath(path)
|
||||
new_path = "/"
|
||||
for part in path.split("/"):
|
||||
if not part or part in (".", ".."):
|
||||
continue
|
||||
new_path = posixpath.join(new_path, part)
|
||||
trailing_slash = "" if new_path.endswith("/") else trailing_slash
|
||||
return new_path + trailing_slash
|
||||
|
||||
|
||||
def clean_name(name):
|
||||
"""Clean an item name by removing slashes and leading/ending brackets."""
|
||||
# Remove leading and ending brackets that may have been put by Outlook
|
||||
name = name.strip("{}")
|
||||
# Remove slashes, mostly unwanted when saving on filesystems
|
||||
name = name.replace("/", "_")
|
||||
return name
|
||||
|
||||
|
||||
def unfold(text):
|
||||
"""Unfold multi-lines attributes.
|
||||
|
||||
@ -80,16 +105,15 @@ class Item(object):
|
||||
if line.startswith("X-RADICALE-NAME:"):
|
||||
self._name = line.replace("X-RADICALE-NAME:", "").strip()
|
||||
break
|
||||
elif line.startswith("TZID:"):
|
||||
self._name = line.replace("TZID:", "").strip()
|
||||
break
|
||||
elif line.startswith("UID:"):
|
||||
self._name = line.replace("UID:", "").strip()
|
||||
# Do not break, a ``X-RADICALE-NAME`` can appear next
|
||||
elif line.startswith("TZID:"):
|
||||
self._name = line.replace("TZID:", "").strip()
|
||||
# Do not break, a ``X-RADICALE-NAME`` can appear next
|
||||
|
||||
if self._name:
|
||||
# Remove brackets that may have been put by Outlook
|
||||
self._name = self._name.strip("{}")
|
||||
self._name = clean_name(self._name)
|
||||
if "\nX-RADICALE-NAME:" in text:
|
||||
for line in unfold(self.text):
|
||||
if line.startswith("X-RADICALE-NAME:"):
|
||||
@ -97,12 +121,11 @@ class Item(object):
|
||||
line, "X-RADICALE-NAME:%s" % self._name)
|
||||
else:
|
||||
self.text = self.text.replace(
|
||||
"\nEND:", "\nX-RADICALE-NAME:%s\nEND:" % self._name)
|
||||
"\nEND:V", "\nX-RADICALE-NAME:%s\nEND:V" % self._name)
|
||||
else:
|
||||
# workaround to get unicode on both python2 and 3
|
||||
self._name = uuid4().hex.encode("ascii").decode("ascii")
|
||||
self._name = uuid4().hex
|
||||
self.text = self.text.replace(
|
||||
"\nEND:", "\nX-RADICALE-NAME:%s\nEND:" % self._name)
|
||||
"\nEND:V", "\nX-RADICALE-NAME:%s\nEND:V" % self._name)
|
||||
|
||||
def __hash__(self):
|
||||
return hash(self.text)
|
||||
@ -183,7 +206,7 @@ class Collection(object):
|
||||
"""
|
||||
self.encoding = "utf-8"
|
||||
# path should already be sanitized
|
||||
self.path = pathutils.sanitize_path(path).strip("/")
|
||||
self.path = sanitize_path(path).strip("/")
|
||||
split_path = self.path.split("/")
|
||||
if principal and split_path and self.is_node(self.path):
|
||||
# Already existing principal collection
|
||||
@ -216,7 +239,7 @@ class Collection(object):
|
||||
return []
|
||||
|
||||
# path should already be sanitized
|
||||
sane_path = pathutils.sanitize_path(path).strip("/")
|
||||
sane_path = sanitize_path(path).strip("/")
|
||||
attributes = sane_path.split("/")
|
||||
if not attributes:
|
||||
return []
|
||||
|
@ -24,71 +24,3 @@ import posixpath
|
||||
from . import log
|
||||
|
||||
|
||||
def sanitize_path(path):
|
||||
"""Make path absolute with leading slash to prevent access to other data.
|
||||
|
||||
Preserve a potential trailing slash.
|
||||
|
||||
"""
|
||||
trailing_slash = "/" if path.endswith("/") else ""
|
||||
path = posixpath.normpath(path)
|
||||
new_path = "/"
|
||||
for part in path.split("/"):
|
||||
if not part or part in (".", ".."):
|
||||
continue
|
||||
new_path = posixpath.join(new_path, part)
|
||||
trailing_slash = "" if new_path.endswith("/") else trailing_slash
|
||||
return new_path + trailing_slash
|
||||
|
||||
|
||||
def is_safe_path_component(path):
|
||||
"""Check if path is a single component of a POSIX path.
|
||||
|
||||
Check that the path is safe to join too.
|
||||
|
||||
"""
|
||||
if not path:
|
||||
return False
|
||||
if posixpath.split(path)[0]:
|
||||
return False
|
||||
if path in (".", ".."):
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def is_safe_filesystem_path_component(path):
|
||||
"""Check if path is a single component of a filesystem path.
|
||||
|
||||
Check that the path is safe to join too.
|
||||
|
||||
"""
|
||||
if not path:
|
||||
return False
|
||||
drive, _ = os.path.splitdrive(path)
|
||||
if drive:
|
||||
return False
|
||||
head, _ = os.path.split(path)
|
||||
if head:
|
||||
return False
|
||||
if path in (os.curdir, os.pardir):
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def path_to_filesystem(path, base_folder):
|
||||
"""Convert path to a local filesystem path relative to base_folder.
|
||||
|
||||
Conversion is done in a secure manner, or raises ``ValueError``.
|
||||
|
||||
"""
|
||||
sane_path = sanitize_path(path).strip("/")
|
||||
safe_path = base_folder
|
||||
if not sane_path:
|
||||
return safe_path
|
||||
for part in sane_path.split("/"):
|
||||
if not is_safe_filesystem_path_component(part):
|
||||
log.LOGGER.debug(
|
||||
"Can't translate path safely to filesystem: %s", path)
|
||||
raise ValueError("Unsafe path")
|
||||
safe_path = os.path.join(safe_path, part)
|
||||
return safe_path
|
||||
|
@ -34,7 +34,7 @@ import sys
|
||||
import time
|
||||
from contextlib import contextmanager
|
||||
|
||||
from . import config, ical, log, pathutils
|
||||
from . import config, ical, log
|
||||
|
||||
|
||||
def _load():
|
||||
@ -52,6 +52,59 @@ FOLDER = os.path.expanduser(config.get("storage", "filesystem_folder"))
|
||||
FILESYSTEM_ENCODING = sys.getfilesystemencoding()
|
||||
|
||||
|
||||
def is_safe_path_component(path):
|
||||
"""Check if path is a single component of a POSIX path.
|
||||
|
||||
Check that the path is safe to join too.
|
||||
|
||||
"""
|
||||
if not path:
|
||||
return False
|
||||
if posixpath.split(path)[0]:
|
||||
return False
|
||||
if path in (".", ".."):
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def is_safe_filesystem_path_component(path):
|
||||
"""Check if path is a single component of a filesystem path.
|
||||
|
||||
Check that the path is safe to join too.
|
||||
|
||||
"""
|
||||
if not path:
|
||||
return False
|
||||
drive, _ = os.path.splitdrive(path)
|
||||
if drive:
|
||||
return False
|
||||
head, _ = os.path.split(path)
|
||||
if head:
|
||||
return False
|
||||
if path in (os.curdir, os.pardir):
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def path_to_filesystem(path):
|
||||
"""Convert path to a local filesystem path relative to base_folder.
|
||||
|
||||
Conversion is done in a secure manner, or raises ``ValueError``.
|
||||
|
||||
"""
|
||||
sane_path = ical.sanitize_path(path).strip("/")
|
||||
safe_path = FOLDER
|
||||
if not sane_path:
|
||||
return safe_path
|
||||
for part in sane_path.split("/"):
|
||||
if not is_safe_filesystem_path_component(part):
|
||||
log.LOGGER.debug(
|
||||
"Can't translate path safely to filesystem: %s", path)
|
||||
raise ValueError("Unsafe path")
|
||||
safe_path = os.path.join(safe_path, part)
|
||||
return safe_path
|
||||
|
||||
|
||||
@contextmanager
|
||||
def _open(path, mode="r"):
|
||||
"""Open a file at ``path`` with encoding set in the configuration."""
|
||||
@ -65,7 +118,7 @@ class Collection(ical.Collection):
|
||||
@property
|
||||
def _filesystem_path(self):
|
||||
"""Absolute path of the file at local ``path``."""
|
||||
return pathutils.path_to_filesystem(self.path, FOLDER)
|
||||
return path_to_filesystem(self.path)
|
||||
|
||||
@property
|
||||
def _props_path(self):
|
||||
@ -86,7 +139,7 @@ class Collection(ical.Collection):
|
||||
item_types = (
|
||||
ical.Timezone, ical.Event, ical.Todo, ical.Journal, ical.Card)
|
||||
for name, component in self._parse(text, item_types).items():
|
||||
if not pathutils.is_safe_filesystem_path_component(name):
|
||||
if not is_safe_filesystem_path_component(name):
|
||||
# TODO: Timezones with slashes can't be saved
|
||||
log.LOGGER.debug(
|
||||
"Can't tranlate name safely to filesystem, "
|
||||
@ -107,7 +160,7 @@ class Collection(ical.Collection):
|
||||
os.remove(self._props_path)
|
||||
|
||||
def remove(self, name):
|
||||
if not pathutils.is_safe_filesystem_path_component(name):
|
||||
if not is_safe_filesystem_path_component(name):
|
||||
log.LOGGER.debug(
|
||||
"Can't tranlate name safely to filesystem, "
|
||||
"skipping component: %s", name)
|
||||
@ -145,12 +198,12 @@ class Collection(ical.Collection):
|
||||
|
||||
@classmethod
|
||||
def children(cls, path):
|
||||
filesystem_path = pathutils.path_to_filesystem(path, FOLDER)
|
||||
filesystem_path = path_to_filesystem(path)
|
||||
_, directories, files = next(os.walk(filesystem_path))
|
||||
for filename in directories + files:
|
||||
# make sure that the local filename can be translated
|
||||
# into an internal path
|
||||
if not pathutils.is_safe_path_component(filename):
|
||||
if not is_safe_path_component(filename):
|
||||
log.LOGGER.debug("Skipping unsupported filename: %s", filename)
|
||||
continue
|
||||
rel_filename = posixpath.join(path, filename)
|
||||
@ -159,14 +212,14 @@ class Collection(ical.Collection):
|
||||
|
||||
@classmethod
|
||||
def is_node(cls, path):
|
||||
filesystem_path = pathutils.path_to_filesystem(path, FOLDER)
|
||||
filesystem_path = path_to_filesystem(path)
|
||||
return (
|
||||
os.path.isdir(filesystem_path) and
|
||||
not os.path.exists(filesystem_path + ".props"))
|
||||
|
||||
@classmethod
|
||||
def is_leaf(cls, path):
|
||||
filesystem_path = pathutils.path_to_filesystem(path, FOLDER)
|
||||
filesystem_path = path_to_filesystem(path)
|
||||
return (
|
||||
os.path.isdir(filesystem_path) and
|
||||
os.path.exists(filesystem_path + ".props"))
|
||||
|
Loading…
x
Reference in New Issue
Block a user