MAX_COMPLETED_TASKS = 50
+class NotAuthorizedError(Exception):
+ pass
+
+
def is_authorized(module, pool, namespace):
return module.is_authorized({"pool": pool or '',
"namespace": namespace or ''})
def authorize_request(module, pool, namespace):
if not is_authorized(module, pool, namespace):
- raise PermissionError("not authorized on pool={}, namespace={}".format(
+ raise NotAuthorizedError("not authorized on pool={}, namespace={}".format(
pool, namespace))
elif prefix.startswith('rbd task '):
return self.task.handle_command(inbuf, prefix[9:], cmd)
+ except NotAuthorizedError:
+ raise
except Exception as ex:
# log the full traceback but don't send it to the CLI user
self.log.fatal("Fatal runtime error: {}\n{}".format(
return -errno.ENOENT, "", str(ex)
except ValueError as ex:
return -errno.EINVAL, "", str(ex)
- except PermissionError as ex:
+ except NotAuthorizedError as ex:
return -errno.EACCES, "", str(ex)
raise NotImplementedError(cmd['prefix'])