base.py 5.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131
  1. # This file is part of Radicale - CalDAV and CardDAV server
  2. # Copyright © 2020 Unrud <unrud@outlook.com>
  3. #
  4. # This library is free software: you can redistribute it and/or modify
  5. # it under the terms of the GNU General Public License as published by
  6. # the Free Software Foundation, either version 3 of the License, or
  7. # (at your option) any later version.
  8. #
  9. # This library is distributed in the hope that it will be useful,
  10. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  12. # GNU General Public License for more details.
  13. #
  14. # You should have received a copy of the GNU General Public License
  15. # along with Radicale. If not, see <http://www.gnu.org/licenses/>.
  16. import io
  17. import logging
  18. import posixpath
  19. import sys
  20. import xml.etree.ElementTree as ET
  21. from typing import Optional
  22. from radicale import (auth, config, httputils, pathutils, rights, storage,
  23. types, web, xmlutils)
  24. from radicale.log import logger
  25. # HACK: https://github.com/tiran/defusedxml/issues/54
  26. import defusedxml.ElementTree as DefusedET # isort:skip
  27. sys.modules["xml.etree"].ElementTree = ET # type:ignore[attr-defined]
  28. class ApplicationBase:
  29. configuration: config.Configuration
  30. _auth: auth.BaseAuth
  31. _storage: storage.BaseStorage
  32. _rights: rights.BaseRights
  33. _web: web.BaseWeb
  34. _encoding: str
  35. def __init__(self, configuration: config.Configuration) -> None:
  36. self.configuration = configuration
  37. self._auth = auth.load(configuration)
  38. self._storage = storage.load(configuration)
  39. self._rights = rights.load(configuration)
  40. self._web = web.load(configuration)
  41. self._encoding = configuration.get("encoding", "request")
  42. def _read_xml_request_body(self, environ: types.WSGIEnviron
  43. ) -> Optional[ET.Element]:
  44. content = httputils.decode_request(
  45. self.configuration, environ,
  46. httputils.read_raw_request_body(self.configuration, environ))
  47. if not content:
  48. return None
  49. try:
  50. xml_content = DefusedET.fromstring(content)
  51. except ET.ParseError as e:
  52. logger.debug("Request content (Invalid XML):\n%s", content)
  53. raise RuntimeError("Failed to parse XML: %s" % e) from e
  54. if logger.isEnabledFor(logging.DEBUG):
  55. logger.debug("Request content:\n%s",
  56. xmlutils.pretty_xml(xml_content))
  57. return xml_content
  58. def _xml_response(self, xml_content: ET.Element) -> bytes:
  59. if logger.isEnabledFor(logging.DEBUG):
  60. logger.debug("Response content:\n%s",
  61. xmlutils.pretty_xml(xml_content))
  62. f = io.BytesIO()
  63. ET.ElementTree(xml_content).write(f, encoding=self._encoding,
  64. xml_declaration=True)
  65. return f.getvalue()
  66. def _webdav_error_response(self, status: int, human_tag: str
  67. ) -> types.WSGIResponse:
  68. """Generate XML error response."""
  69. headers = {"Content-Type": "text/xml; charset=%s" % self._encoding}
  70. content = self._xml_response(xmlutils.webdav_error(human_tag))
  71. return status, headers, content
  72. class Access:
  73. """Helper class to check access rights of an item"""
  74. user: str
  75. path: str
  76. parent_path: str
  77. permissions: str
  78. _rights: rights.BaseRights
  79. _parent_permissions: Optional[str]
  80. def __init__(self, rights: rights.BaseRights, user: str, path: str
  81. ) -> None:
  82. self._rights = rights
  83. self.user = user
  84. self.path = path
  85. self.parent_path = pathutils.unstrip_path(
  86. posixpath.dirname(pathutils.strip_path(path)), True)
  87. self.permissions = self._rights.authorization(self.user, self.path)
  88. self._parent_permissions = None
  89. @property
  90. def parent_permissions(self) -> str:
  91. if self.path == self.parent_path:
  92. return self.permissions
  93. if self._parent_permissions is None:
  94. self._parent_permissions = self._rights.authorization(
  95. self.user, self.parent_path)
  96. return self._parent_permissions
  97. def check(self, permission: str,
  98. item: Optional[types.CollectionOrItem] = None) -> bool:
  99. if permission not in "rw":
  100. raise ValueError("Invalid permission argument: %r" % permission)
  101. if not item:
  102. permissions = permission + permission.upper()
  103. parent_permissions = permission
  104. elif isinstance(item, storage.BaseCollection):
  105. if item.tag:
  106. permissions = permission
  107. else:
  108. permissions = permission.upper()
  109. parent_permissions = ""
  110. else:
  111. permissions = ""
  112. parent_permissions = permission
  113. return bool(rights.intersect(self.permissions, permissions) or (
  114. self.path != self.parent_path and
  115. rights.intersect(self.parent_permissions, parent_permissions)))