Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def handle_copy(self, dest_path, depth_infinity):
"""Change semantic of COPY to add resource tags."""
# destPath must be '/by_tag//'
if "/by_tag/" not in dest_path:
raise DAVError(HTTP_FORBIDDEN)
catType, tag, _rest = util.save_split(dest_path.strip("/"), "/", 2)
assert catType == "by_tag"
if tag not in self.data["tags"]:
self.data["tags"].append(tag)
return True # OK
projectdb = self.app.config['projectdb']
if projectdb:
self._project = projectdb.get(self.project_name)
if not self._project:
if projectdb.verify_project_name(self.project_name) and text(self.name).endswith('.py'):
self.new_project = True
self._project = {
'name': self.project_name,
'script': '',
'status': 'TODO',
'rate': self.app.config.get('max_rate', 1),
'burst': self.app.config.get('max_burst', 3),
'updatetime': time.time(),
}
else:
raise DAVError(HTTP_FORBIDDEN)
return self._project
"""
path = environ["PATH_INFO"]
res = self._davProvider.get_resource_inst(path, environ)
if util.get_content_length(environ) != 0:
self._fail(
HTTP_MEDIATYPE_NOT_SUPPORTED,
"The server does not handle any body content.",
)
elif environ.setdefault("HTTP_DEPTH", "0") != "0":
self._fail(HTTP_BAD_REQUEST, "Only Depth: 0 supported.")
elif res is None:
self._fail(HTTP_NOT_FOUND, path)
elif res.is_collection:
self._fail(
HTTP_FORBIDDEN,
"Directory browsing is not enabled."
"(to enable it put WsgiDavDirBrowser into middleware_stack"
"option and set dir_browser -> enabled = True option.)",
)
self._evaluate_if_headers(res, environ)
filesize = res.get_content_length()
if filesize is None:
filesize = -1 # flag logic to read until EOF
last_modified = res.get_last_modified()
if last_modified is None:
last_modified = -1
entitytag = res.get_etag()
def set_property_value(self, name, value, dry_run=False):
"""Set or remove property value.
See DAVResource.set_property_value()
"""
raise DAVError(
HTTP_FORBIDDEN, err_condition=PRECONDITION_CODE_ProtectedProperty
)
def deleteCollection(self, path):
raise DAVError(HTTP_FORBIDDEN)
def begin_write(self, content_type=None):
"""Open content as a stream for writing.
See DAVResource.begin_write()
"""
assert not self.is_collection
if self.provider.readonly:
raise DAVError(HTTP_FORBIDDEN)
# _logger.debug("begin_write: {}, {}".format(self._file_path, "wb"))
# GC issue 57: always store as binary
return open(self._file_path, "wb", BUFFER_SIZE)
def moveRecursive(self, destPath):
"""See DAVResource.moveRecursive() """
if self.provider.readonly:
raise DAVError(HTTP_FORBIDDEN)
fpDest = self.provider._locToFilePath(destPath)
assert not util.isEqualOrChildUri(self.path, destPath)
assert not os.path.exists(fpDest)
_logger.debug("moveRecursive(%s, %s)" % (self._filePath, fpDest))
shutil.move(self._filePath, fpDest)
# (Live properties are copied by copy2 or copystat)
# Move dead properties
if self.provider.propManager:
destRes = self.provider.getResourceInst(destPath, self.environ)
self.provider.propManager.moveProperties(self.getRefUrl(),
destRes.getRefUrl(),
withChildren=True)
def delete(self):
"""Remove this resource or collection (recursive).
See DAVResource.delete()
"""
if self.provider.readonly:
raise DAVError(HTTP_FORBIDDEN)
os.unlink(self._file_path)
self.remove_all_properties(True)
self.remove_all_locks(True)
def delete(self):
"""Remove this resource or collection (recursive).
See DAVResource.delete()
"""
if self.provider.readonly:
raise DAVError(HTTP_FORBIDDEN)
shutil.rmtree(self._filePath, ignore_errors=False)
self.removeAllProperties(True)
self.removeAllLocks(True)