tobias/radicale
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Allow finer control in rights plugin

Browse Source 
New permissions:

R: read collections without tag
r: read collections with tag and included objects
W: write and delete collections without tag
w: write and delete collection with tag and included objects
This commit is contained in:
Unrud 2018-08-21 18:43:45 +02:00
parent 72501c6e23
commit 0a492a00b1
6 changed files with 124 additions and 116 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

115
radicale/__init__.py
View File

@ -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:

87
radicale/rights.py
View File

@ -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 ""

6
radicale/tests/custom/rights.py
View File

@ -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)

4
radicale/tests/test_rights.py
View File

@ -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)

18
radicale/xmlutils.py
View File

@ -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)

10
rights
View File

@ -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