diff --git a/radicale/__init__.py b/radicale/__init__.py index ac9a3f6..4a74581 100644 --- a/radicale/__init__.py +++ b/radicale/__init__.py @@ -144,32 +144,33 @@ class Application: def collect_allowed_items(self, items, user): """Get items from request that user is allowed to access.""" - read_allowed_items = [] - write_allowed_items = [] for item in items: if isinstance(item, storage.BaseCollection): path = storage.sanitize_path("/%s/" % item.path) - can_read = self.Rights.authorized(user, path, "r") - can_write = self.Rights.authorized(user, path, "w") - target = "collection %r" % item.path + if item.get_meta("tag"): + permissions = self.Rights.authorized(user, path, "rw") + target = "collection with tag %r" % item.path + else: + permissions = self.Rights.authorized(user, path, "RW") + target = "collection %r" % item.path else: - path = storage.sanitize_path("/%s/%s" % (item.collection.path, - item.href)) - can_read = self.Rights.authorized_item(user, path, "r") - can_write = self.Rights.authorized_item(user, path, "w") + path = storage.sanitize_path("/%s/" % item.collection.path) + permissions = self.Rights.authorized(user, path, "rw") target = "item %r from %r" % (item.href, item.collection.path) - text_status = [] - if can_read: - text_status.append("read") - read_allowed_items.append(item) - if can_write: - text_status.append("write") - write_allowed_items.append(item) + if rights.intersect_permissions(permissions, "Ww"): + permission = "w" + status = "write" + elif rights.intersect_permissions(permissions, "Rr"): + permission = "r" + status = "read" + else: + permission = "" + status = "NO" logger.debug( "%s has %s access to %s", - repr(user) if user else "anonymous user", - " and ".join(text_status) if text_status else "NO", target) - return read_allowed_items, write_allowed_items + repr(user) if user else "anonymous user", status, target) + if permission: + yield item, permission def __call__(self, environ, start_response): with log.register_stream(environ["wsgi.errors"]): @@ -315,7 +316,7 @@ class Application: # Create principal collection if user: principal_path = "/%s/" % user - if self.Rights.authorized(user, principal_path, "w"): + if self.Rights.authorized(user, principal_path, "W"): with self.Collection.acquire_lock("r", user): principal = next( self.Collection.discover(principal_path, depth="1"), @@ -365,19 +366,28 @@ class Application: return response(status, headers, answer) def _access(self, user, path, permission, item=None): - """Check if ``user`` can access ``path`` or the parent collection. - - ``permission`` must either be "r" or "w". - - If ``item`` is given, only access to that class of item is checked. - - """ - allowed = False - if not item or isinstance(item, storage.BaseCollection): - allowed |= self.Rights.authorized(user, path, permission) - if not item or not isinstance(item, storage.BaseCollection): - allowed |= self.Rights.authorized_item(user, path, permission) - return allowed + if permission not in "rw": + raise ValueError("Invalid permission argument: %r" % permission) + if not item: + permissions = permission + permission.upper() + parent_permissions = permission + elif isinstance(item, storage.BaseCollection): + if item.get_meta("tag"): + permissions = permission + else: + permissions = permission.upper() + parent_permissions = "" + else: + permissions = "" + parent_permissions = permission + if permissions and self.Rights.authorized(user, path, permissions): + return True + if parent_permissions: + parent_path = storage.sanitize_path( + "/%s/" % posixpath.dirname(path.strip("/"))) + if self.Rights.authorized(user, parent_path, parent_permissions): + return True + return False def _read_raw_content(self, environ): content_length = int(environ.get("CONTENT_LENGTH") or 0) @@ -459,10 +469,10 @@ class Application: return NOT_ALLOWED with self.Collection.acquire_lock("w", user): item = next(self.Collection.discover(path), None) - if not self._access(user, path, "w", item): - return NOT_ALLOWED if not item: return NOT_FOUND + if not self._access(user, path, "w", item): + return NOT_ALLOWED if_match = environ.get("HTTP_IF_MATCH", "*") if if_match not in ("*", item.etag): # ETag precondition not verified, do not delete item @@ -493,10 +503,10 @@ class Application: return NOT_ALLOWED with self.Collection.acquire_lock("r", user): item = next(self.Collection.discover(path), None) - if not self._access(user, path, "r", item): - return NOT_ALLOWED if not item: return NOT_FOUND + if not self._access(user, path, "r", item): + return NOT_ALLOWED if isinstance(item, storage.BaseCollection): tag = item.get_meta("tag") if not tag: @@ -563,7 +573,8 @@ class Application: def do_MKCOL(self, environ, base_prefix, path, user): """Manage MKCOL request.""" - if not self.Rights.authorized(user, path, "w"): + permissions = self.Rights.authorized(user, path, "Ww") + if not permissions: return NOT_ALLOWED try: xml_content = self._read_xml_content(environ) @@ -587,6 +598,9 @@ class Application: parent_item.get_meta("tag")): return FORBIDDEN props = xmlutils.props_from_request(xml_content) + if (props.get("tag") and "w" not in permissions or + not props.get("tag") and "W" not in permissions): + return NOT_ALLOWED try: storage.check_and_sanitize_props(props) self.Collection.create_collection(path, props=props) @@ -617,12 +631,11 @@ class Application: with self.Collection.acquire_lock("w", user): item = next(self.Collection.discover(path), None) - if not self._access(user, path, "w", item): - return NOT_ALLOWED - if not self._access(user, to_path, "w", item): - return NOT_ALLOWED if not item: return NOT_FOUND + if (not self._access(user, path, "w", item) or + not self._access(user, to_path, "w", item)): + return NOT_ALLOWED if isinstance(item, storage.BaseCollection): # TODO: support moving collections return METHOD_NOT_ALLOWED @@ -682,24 +695,24 @@ class Application: path, environ.get("HTTP_DEPTH", "0")) # take root item for rights checking item = next(items, None) - if not self._access(user, path, "r", item): - return NOT_ALLOWED if not item: return NOT_FOUND + if not self._access(user, path, "r", item): + return NOT_ALLOWED # put item back items = itertools.chain([item], items) - read_items, write_items = self.collect_allowed_items(items, user) + allowed_items = self.collect_allowed_items(items, user) headers = {"DAV": DAV_HEADERS, "Content-Type": "text/xml; charset=%s" % self.encoding} status, xml_answer = xmlutils.propfind( - base_prefix, path, xml_content, read_items, write_items, user) + base_prefix, path, xml_content, allowed_items, user) if status == client.FORBIDDEN: return NOT_ALLOWED return status, headers, self._write_xml_content(xml_answer) def do_PROPPATCH(self, environ, base_prefix, path, user): """Manage PROPPATCH request.""" - if not self.Rights.authorized(user, path, "w"): + if not self._access(user, path, "w"): return NOT_ALLOWED try: xml_content = self._read_xml_content(environ) @@ -714,6 +727,8 @@ class Application: item = next(self.Collection.discover(path), None) if not item: return NOT_FOUND + if not self._access(user, path, "w", item): + return NOT_ALLOWED if not isinstance(item, storage.BaseCollection): return FORBIDDEN headers = {"DAV": DAV_HEADERS, @@ -754,7 +769,7 @@ class Application: if write_whole_collection: if not self.Rights.authorized(user, path, "w"): return NOT_ALLOWED - elif not self.Rights.authorized_item(user, path, "w"): + elif not self.Rights.authorized(user, parent_path, "w"): return NOT_ALLOWED etag = environ.get("HTTP_IF_MATCH", "") @@ -850,10 +865,10 @@ class Application: return REQUEST_TIMEOUT with self.Collection.acquire_lock("r", user): item = next(self.Collection.discover(path), None) - if not self._access(user, path, "r", item): - return NOT_ALLOWED if not item: return NOT_FOUND + if not self._access(user, path, "r", item): + return NOT_ALLOWED if isinstance(item, storage.BaseCollection): collection = item else: diff --git a/radicale/rights.py b/radicale/rights.py index b5c3df3..4d3bbbd 100644 --- a/radicale/rights.py +++ b/radicale/rights.py @@ -39,7 +39,6 @@ Leading or ending slashes are trimmed from collection's path. import configparser import os.path -import posixpath import re from importlib import import_module @@ -75,59 +74,61 @@ def load(configuration): return rights_class(configuration) +def intersect_permissions(a, b="RrWw"): + return "".join(set(a).intersection(set(b))) + + class BaseRights: def __init__(self, configuration): self.configuration = configuration - def authorized(self, user, path, permission): + def authorized(self, user, path, permissions): """Check if the user is allowed to read or write the collection. If ``user`` is empty, check for anonymous rights. ``path`` is sanitized. - ``permission`` is "r" or "w". + ``permissions`` can include "R", "r", "W", "w" + + Returns granted rights. """ raise NotImplementedError - def authorized_item(self, user, path, permission): - """Check if the user is allowed to read or write the item.""" - path = storage.sanitize_path(path) - parent_path = storage.sanitize_path( - "/%s/" % posixpath.dirname(path.strip("/"))) - return self.authorized(user, parent_path, permission) - class NoneRights(BaseRights): - def authorized(self, user, path, permission): - return True + def authorized(self, user, path, permissions): + return intersect_permissions(permissions) class AuthenticatedRights(BaseRights): - def authorized(self, user, path, permission): - return bool(user) + def authorized(self, user, path, permissions): + if not user: + return "" + return intersect_permissions(permissions) class OwnerWriteRights(BaseRights): - def authorized(self, user, path, permission): + def authorized(self, user, path, permissions): + if not user: + return "" sane_path = storage.sanitize_path(path).strip("/") - return bool(user) and (permission == "r" or - user == sane_path.split("/", maxsplit=1)[0]) + if user != sane_path.split("/", maxsplit=1)[0]: + return intersect_permissions(permissions, "Rr") + return intersect_permissions(permissions) class OwnerOnlyRights(BaseRights): - def authorized(self, user, path, permission): + def authorized(self, user, path, permissions): + if not user: + return "" sane_path = storage.sanitize_path(path).strip("/") - return bool(user) and ( - permission == "r" and not sane_path or - user == sane_path.split("/", maxsplit=1)[0]) - - def authorized_item(self, user, path, permission): - sane_path = storage.sanitize_path(path).strip("/") - if "/" not in sane_path: - return False - return super().authorized_item(user, path, permission) + if not sane_path: + return intersect_permissions(permissions, "R") + if user != sane_path.split("/", maxsplit=1)[0]: + return "" + return intersect_permissions(permissions) class Rights(BaseRights): @@ -135,41 +136,41 @@ class Rights(BaseRights): super().__init__(configuration) self.filename = os.path.expanduser(configuration.get("rights", "file")) - def authorized(self, user, path, permission): + def authorized(self, user, path, permissions): user = user or "" sane_path = storage.sanitize_path(path).strip("/") # Prevent "regex injection" user_escaped = re.escape(user) sane_path_escaped = re.escape(sane_path) - regex = configparser.ConfigParser( + rights_config = configparser.ConfigParser( {"login": user_escaped, "path": sane_path_escaped}) try: - if not regex.read(self.filename): + if not rights_config.read(self.filename): raise RuntimeError("No such file: %r" % self.filename) except Exception as e: raise RuntimeError("Failed to load rights file %r: %s" % (self.filename, e)) from e - for section in regex.sections(): + for section in rights_config.sections(): try: - re_user_pattern = regex.get(section, "user") - re_collection_pattern = regex.get(section, "collection") - # Emulate fullmatch - user_match = re.match(r"(?:%s)\Z" % re_user_pattern, user) - collection_match = user_match and re.match( - r"(?:%s)\Z" % re_collection_pattern.format( + user_pattern = rights_config.get(section, "user") + collection_pattern = rights_config.get(section, "collection") + user_match = re.fullmatch(user_pattern, user) + collection_match = user_match and re.fullmatch( + collection_pattern.format( *map(re.escape, user_match.groups())), sane_path) except Exception as e: raise RuntimeError("Error in section %r of rights file %r: " "%s" % (section, self.filename, e)) from e if user_match and collection_match: logger.debug("Rule %r:%r matches %r:%r from section %r", - user, sane_path, re_user_pattern, - re_collection_pattern, section) - return permission in regex.get(section, "permission") + user, sane_path, user_pattern, + collection_pattern, section) + return intersect_permissions( + permissions, rights_config.get(section, "permissions")) else: logger.debug("Rule %r:%r doesn't match %r:%r from section %r", - user, sane_path, re_user_pattern, - re_collection_pattern, section) + user, sane_path, user_pattern, + collection_pattern, section) logger.info("Rights: %r:%r doesn't match any section", user, sane_path) - return False + return "" diff --git a/radicale/tests/custom/rights.py b/radicale/tests/custom/rights.py index a8c10ac..9760044 100644 --- a/radicale/tests/custom/rights.py +++ b/radicale/tests/custom/rights.py @@ -23,5 +23,7 @@ from radicale import rights class Rights(rights.BaseRights): - def authorized(self, user, path, permission): - return path.strip("/") in ("tmp", "other") + def authorized(self, user, path, permissions): + if path.strip("/") not in ("tmp", "other"): + return "" + return rights.intersect_permissions(permissions) diff --git a/radicale/tests/test_rights.py b/radicale/tests/test_rights.py index 4d17eb5..5c3f9eb 100644 --- a/radicale/tests/test_rights.py +++ b/radicale/tests/test_rights.py @@ -118,11 +118,11 @@ class TestBaseAuthRequests(BaseTest): [owner] user: .+ collection: %(login)s(/.*)? -permission: rw +permissions: RrWw [custom] user: .* collection: custom(/.*)? -permission: r""") +permissions: Rr""") self.configuration["rights"]["file"] = rights_file_path self._test_rights("from_file", "", "/other", "r", 401) self._test_rights("from_file", "tmp", "/other", "r", 403) diff --git a/radicale/xmlutils.py b/radicale/xmlutils.py index d51a17f..0b7b3f0 100644 --- a/radicale/xmlutils.py +++ b/radicale/xmlutils.py @@ -771,8 +771,7 @@ def delete(base_prefix, path, collection, href=None): return multistatus -def propfind(base_prefix, path, xml_request, read_collections, - write_collections, user): +def propfind(base_prefix, path, xml_request, allowed_items, user): """Read and answer PROPFIND requests. Read rfc4918-9.1 for info. @@ -805,19 +804,10 @@ def propfind(base_prefix, path, xml_request, read_collections, # Writing answer multistatus = ET.Element(_tag("D", "multistatus")) - collections = [] - for collection in write_collections: - collections.append(collection) + for item, permission in allowed_items: + write = permission == "w" response = _propfind_response( - base_prefix, path, collection, props, user, write=True, - allprop=allprop, propname=propname) - if response: - multistatus.append(response) - for collection in read_collections: - if collection in collections: - continue - response = _propfind_response( - base_prefix, path, collection, props, user, write=False, + base_prefix, path, item, props, user, write=write, allprop=allprop, propname=propname) if response: multistatus.append(response) diff --git a/rights b/rights index b93a8fe..f997e00 100644 --- a/rights +++ b/rights @@ -19,30 +19,30 @@ [admin] user: admin.* collection: .* -permission: r +permissions: Rr # This means all users may read and write any collection starting with public. # We do so by just not testing against the user string. [public] user: .* collection: public(/.+)? -permission: rw +permissions: RrWw # A little more complex: give read access to users from a domain for all # collections of all the users (ie. user@domain.tld can read domain/*). [domain-wide-access] user: .+@(.+)\..+ collection: {0}/.+ -permission: r +permissions: Rr # Allow authenticated user to read all collections [allow-everyone-read] user: .+ collection: .* -permission: r +permissions: Rr # Give write access to owners [owner-write] user: .+ collection: %(login)s/.* -permission: w +permissions: Ww