ceph/src/pybind/mgr/dashboard/controllers/docs.py

   1 # -*- coding: utf-8 -*-
   2 from __future__ import absolute_import
   3
   4 import logging
   5 import cherrypy
   6
   7 from . import Controller, BaseController, Endpoint, ENDPOINT_MAP
   8 from .. import mgr
   9
  10 from ..tools import str_to_bool
  11
  12
  13 logger = logging.getLogger('controllers.docs')
  14
  15
  16 @Controller('/docs', secure=False)
  17 class Docs(BaseController):
  18
  19     @classmethod
  20     def _gen_tags(cls, all_endpoints):
  21         """ Generates a list of all tags and corresponding descriptions. """
  22         # Scenarios to consider:
  23         #     * Intentionally make up a new tag name at controller => New tag name displayed.
  24         #     * Misspell or make up a new tag name at endpoint => Neither tag or endpoint displayed.
  25         #     * Misspell tag name at controller (when referring to another controller) =>
  26         #       Tag displayed but no endpoints assigned
  27         #     * Description for a tag added at multiple locations => Only one description displayed.
  28         list_of_ctrl = set()
  29         for endpoints in ENDPOINT_MAP.values():
  30             for endpoint in endpoints:
  31                 if endpoint.is_api or all_endpoints:
  32                     list_of_ctrl.add(endpoint.ctrl)
  33
  34         tag_map = {}
  35         for ctrl in list_of_ctrl:
  36             tag_name = ctrl.__name__
  37             tag_descr = ""
  38             if hasattr(ctrl, 'doc_info'):
  39                 if ctrl.doc_info['tag']:
  40                     tag_name = ctrl.doc_info['tag']
  41                 tag_descr = ctrl.doc_info['tag_descr']
  42             if tag_name not in tag_map or not tag_map[tag_name]:
  43                 tag_map[tag_name] = tag_descr
  44
  45         tags = [{'name': k, 'description': v if v else "*No description available*"}
  46                 for k, v in tag_map.items()]
  47         tags.sort(key=lambda e: e['name'])
  48         return tags
  49
  50     @classmethod
  51     def _get_tag(cls, endpoint):
  52         """ Returns the name of a tag to assign to a path. """
  53         ctrl = endpoint.ctrl
  54         func = endpoint.func
  55         tag = ctrl.__name__
  56         if hasattr(func, 'doc_info') and func.doc_info['tag']:
  57             tag = func.doc_info['tag']
  58         elif hasattr(ctrl, 'doc_info') and ctrl.doc_info['tag']:
  59             tag = ctrl.doc_info['tag']
  60         return tag
  61
  62     @classmethod
  63     def _gen_type(cls, param):
  64         # pylint: disable=too-many-return-statements
  65         """
  66         Generates the type of parameter based on its name and default value,
  67         using very simple heuristics.
  68         Used if type is not explicitly defined.
  69         """
  70         param_name = param['name']
  71         def_value = param['default'] if 'default' in param else None
  72         if param_name.startswith("is_"):
  73             return "boolean"
  74         if "size" in param_name:
  75             return "integer"
  76         if "count" in param_name:
  77             return "integer"
  78         if "num" in param_name:
  79             return "integer"
  80         if isinstance(def_value, bool):
  81             return "boolean"
  82         if isinstance(def_value, int):
  83             return "integer"
  84         return "string"
  85
  86     @classmethod
  87     # isinstance doesn't work: input is always <type 'type'>.
  88     def _type_to_str(cls, type_as_type):
  89         """ Used if type is explicitly defined. """
  90         if type_as_type is str:
  91             type_as_str = 'string'
  92         elif type_as_type is int:
  93             type_as_str = 'integer'
  94         elif type_as_type is bool:
  95             type_as_str = 'boolean'
  96         elif type_as_type is list or type_as_type is tuple:
  97             type_as_str = 'array'
  98         elif type_as_type is float:
  99             type_as_str = 'number'
 100         else:
 101             type_as_str = 'object'
 102         return type_as_str
 103
 104     @classmethod
 105     def _add_param_info(cls, parameters, p_info):
 106         # Cases to consider:
 107         #     * Parameter name (if not nested) misspelt in decorator => parameter not displayed
 108         #     * Sometimes a parameter is used for several endpoints (e.g. fs_id in CephFS).
 109         #       Currently, there is no possibility of reuse. Should there be?
 110         #       But what if there are two parameters with same name but different functionality?
 111         """
 112         Adds explicitly described information for parameters of an endpoint.
 113
 114         There are two cases:
 115         * Either the parameter in p_info corresponds to an endpoint parameter. Implicit information
 116         has higher priority, so only information that doesn't already exist is added.
 117         * Or the parameter in p_info describes a nested parameter inside an endpoint parameter.
 118         In that case there is no implicit information at all so all explicitly described info needs
 119         to be added.
 120         """
 121         for p in p_info:
 122             if not p['nested']:
 123                 for parameter in parameters:
 124                     if p['name'] == parameter['name']:
 125                         parameter['type'] = p['type']
 126                         parameter['description'] = p['description']
 127                         if 'nested_params' in p:
 128                             parameter['nested_params'] = cls._add_param_info([], p['nested_params'])
 129             else:
 130                 nested_p = {
 131                     'name': p['name'],
 132                     'type': p['type'],
 133                     'description': p['description'],
 134                     'required': p['required'],
 135                 }
 136                 if 'default' in p:
 137                     nested_p['default'] = p['default']
 138                 if 'nested_params' in p:
 139                     nested_p['nested_params'] = cls._add_param_info([], p['nested_params'])
 140                 parameters.append(nested_p)
 141
 142         return parameters
 143
 144     @classmethod
 145     def _gen_schema_for_content(cls, params):
 146         """
 147         Generates information to the content-object in OpenAPI Spec.
 148         Used to for request body and responses.
 149         """
 150         required_params = []
 151         properties = {}
 152
 153         for param in params:
 154             if param['required']:
 155                 required_params.append(param['name'])
 156
 157             props = {}
 158             if 'type' in param:
 159                 props['type'] = cls._type_to_str(param['type'])
 160                 if 'nested_params' in param:
 161                     if props['type'] == 'array':  # dict in array
 162                         props['items'] = cls._gen_schema_for_content(param['nested_params'])
 163                     else:  # dict in dict
 164                         props = cls._gen_schema_for_content(param['nested_params'])
 165                 elif props['type'] == 'object':  # e.g. [int]
 166                     props['type'] = 'array'
 167                     props['items'] = {'type': cls._type_to_str(param['type'][0])}
 168             else:
 169                 props['type'] = cls._gen_type(param)
 170             if 'description' in param:
 171                 props['description'] = param['description']
 172             if 'default' in param:
 173                 props['default'] = param['default']
 174             properties[param['name']] = props
 175
 176         schema = {
 177             'type': 'object',
 178             'properties': properties,
 179         }
 180         if required_params:
 181             schema['required'] = required_params
 182         return schema
 183
 184     @classmethod
 185     def _gen_responses(cls, method, resp_object=None):
 186         resp = {
 187             '400': {
 188                 "description": "Operation exception. Please check the "
 189                                "response body for details."
 190             },
 191             '401': {
 192                 "description": "Unauthenticated access. Please login first."
 193             },
 194             '403': {
 195                 "description": "Unauthorized access. Please check your "
 196                                "permissions."
 197             },
 198             '500': {
 199                 "description": "Unexpected error. Please check the "
 200                                "response body for the stack trace."
 201             }
 202         }
 203         if method.lower() == 'get':
 204             resp['200'] = {'description': "OK"}
 205         if method.lower() == 'post':
 206             resp['201'] = {'description': "Resource created."}
 207         if method.lower() == 'put':
 208             resp['200'] = {'description': "Resource updated."}
 209         if method.lower() == 'delete':
 210             resp['204'] = {'description': "Resource deleted."}
 211         if method.lower() in ['post', 'put', 'delete']:
 212             resp['202'] = {'description': "Operation is still executing."
 213                                           " Please check the task queue."}
 214
 215         if resp_object:
 216             for status_code, response_body in resp_object.items():
 217                 resp[status_code].update({
 218                     'content': {
 219                         'application/json': {
 220                             'schema': cls._gen_schema_for_content(response_body)}}})
 221
 222         return resp
 223
 224     @classmethod
 225     def _gen_params(cls, params, location):
 226         parameters = []
 227         for param in params:
 228             if 'type' in param:
 229                 _type = cls._type_to_str(param['type'])
 230             else:
 231                 _type = cls._gen_type(param)
 232             if 'description' in param:
 233                 descr = param['description']
 234             else:
 235                 descr = "*No description available*"
 236             res = {
 237                 'name': param['name'],
 238                 'in': location,
 239                 'schema': {
 240                     'type': _type
 241                 },
 242                 'description': descr
 243             }
 244             if param['required']:
 245                 res['required'] = True
 246             elif param['default'] is None:
 247                 res['allowEmptyValue'] = True
 248             else:
 249                 res['default'] = param['default']
 250             parameters.append(res)
 251
 252         return parameters
 253
 254     @classmethod
 255     def _gen_paths(cls, all_endpoints, base_url):
 256         method_order = ['get', 'post', 'put', 'delete']
 257         paths = {}
 258         for path, endpoints in sorted(list(ENDPOINT_MAP.items()),
 259                                       key=lambda p: p[0]):
 260             methods = {}
 261             skip = False
 262
 263             endpoint_list = sorted(endpoints, key=lambda e:
 264                                    method_order.index(e.method.lower()))
 265             for endpoint in endpoint_list:
 266                 if not endpoint.is_api and not all_endpoints:
 267                     skip = True
 268                     break
 269
 270                 method = endpoint.method
 271                 func = endpoint.func
 272
 273                 summary = "No description available"
 274                 resp = {}
 275                 p_info = []
 276                 if hasattr(func, 'doc_info'):
 277                     if func.doc_info['summary']:
 278                         summary = func.doc_info['summary']
 279                     resp = func.doc_info['response']
 280                     p_info = func.doc_info['parameters']
 281                 params = []
 282                 if endpoint.path_params:
 283                     params.extend(
 284                         cls._gen_params(
 285                             cls._add_param_info(endpoint.path_params, p_info), 'path'))
 286                 if endpoint.query_params:
 287                     params.extend(
 288                         cls._gen_params(
 289                             cls._add_param_info(endpoint.query_params, p_info), 'query'))
 290
 291                 methods[method.lower()] = {
 292                     'tags': [cls._get_tag(endpoint)],
 293                     'summary': summary,
 294                     'description': func.__doc__,
 295                     'parameters': params,
 296                     'responses': cls._gen_responses(method, resp)
 297                 }
 298
 299                 if method.lower() in ['post', 'put']:
 300                     if endpoint.body_params:
 301                         body_params = cls._add_param_info(endpoint.body_params, p_info)
 302                         methods[method.lower()]['requestBody'] = {
 303                             'content': {
 304                                 'application/json': {
 305                                     'schema': cls._gen_schema_for_content(body_params)}}}
 306
 307                 if endpoint.is_secure:
 308                     methods[method.lower()]['security'] = [{'jwt': []}]
 309
 310             if not skip:
 311                 paths[path[len(base_url):]] = methods
 312
 313         return paths
 314
 315     def _gen_spec(self, all_endpoints=False, base_url=""):
 316         if all_endpoints:
 317             base_url = ""
 318
 319         host = cherrypy.request.base
 320         host = host[host.index(':')+3:]
 321         logger.debug("Host: %s", host)
 322
 323         paths = self._gen_paths(all_endpoints, base_url)
 324
 325         if not base_url:
 326             base_url = "/"
 327
 328         scheme = 'https'
 329         ssl = str_to_bool(mgr.get_localized_module_option('ssl', True))
 330         if not ssl:
 331             scheme = 'http'
 332
 333         spec = {
 334             'openapi': "3.0.0",
 335             'info': {
 336                 'description': "Please note that this API is not an official "
 337                                "Ceph REST API to be used by third-party "
 338                                "applications. It's primary purpose is to serve"
 339                                " the requirements of the Ceph Dashboard and is"
 340                                " subject to change at any time. Use at your "
 341                                "own risk.",
 342                 'version': "v1",
 343                 'title': "Ceph-Dashboard REST API"
 344             },
 345             'host': host,
 346             'basePath': base_url,
 347             'servers': [{'url': "{}{}".format(cherrypy.request.base, base_url)}],
 348             'tags': self._gen_tags(all_endpoints),
 349             'schemes': [scheme],
 350             'paths': paths,
 351             'components': {
 352                 'securitySchemes': {
 353                     'jwt': {
 354                         'type': 'http',
 355                         'scheme': 'bearer',
 356                         'bearerFormat': 'JWT'
 357                     }
 358                 }
 359             }
 360         }
 361
 362         return spec
 363
 364     @Endpoint(path="api.json")
 365     def api_json(self):
 366         return self._gen_spec(False, "/api")
 367
 368     @Endpoint(path="api-all.json")
 369     def api_all_json(self):
 370         return self._gen_spec(True, "/api")
 371
 372     def _swagger_ui_page(self, all_endpoints=False, token=None):
 373         base = cherrypy.request.base
 374         if all_endpoints:
 375             spec_url = "{}/docs/api-all.json".format(base)
 376         else:
 377             spec_url = "{}/docs/api.json".format(base)
 378
 379         auth_header = cherrypy.request.headers.get('authorization')
 380         jwt_token = ""
 381         if auth_header is not None:
 382             scheme, params = auth_header.split(' ', 1)
 383             if scheme.lower() == 'bearer':
 384                 jwt_token = params
 385         else:
 386             if token is not None:
 387                 jwt_token = token
 388
 389         api_key_callback = """, onComplete: () => {{
 390                         ui.preauthorizeApiKey('jwt', '{}');
 391                     }}
 392         """.format(jwt_token)
 393
 394         page = """
 395         <!DOCTYPE html>
 396         <html>
 397         <head>
 398             <meta charset="UTF-8">
 399             <meta name="referrer" content="no-referrer" />
 400             <link rel="stylesheet" type="text/css"
 401                   href="/swagger-ui.css" >
 402             <style>
 403                 html
 404                 {{
 405                     box-sizing: border-box;
 406                     overflow: -moz-scrollbars-vertical;
 407                     overflow-y: scroll;
 408                 }}
 409                 *,
 410                 *:before,
 411                 *:after
 412                 {{
 413                     box-sizing: inherit;
 414                 }}
 415                 body {{
 416                     margin:0;
 417                     background: #fafafa;
 418                 }}
 419             </style>
 420         </head>
 421         <body>
 422         <div id="swagger-ui"></div>
 423         <script src="/swagger-ui-bundle.js">
 424         </script>
 425         <script>
 426             window.onload = function() {{
 427                 const ui = SwaggerUIBundle({{
 428                     url: '{}',
 429                     dom_id: '#swagger-ui',
 430                     presets: [
 431                         SwaggerUIBundle.presets.apis
 432                     ],
 433                     layout: "BaseLayout"
 434                     {}
 435                 }})
 436                 window.ui = ui
 437             }}
 438         </script>
 439         </body>
 440         </html>
 441         """.format(spec_url, api_key_callback)
 442
 443         return page
 444
 445     @Endpoint(json_response=False)
 446     def __call__(self, all_endpoints=False):
 447         return self._swagger_ui_page(all_endpoints)
 448
 449     @Endpoint('POST', path="/", json_response=False,
 450               query_params="{all_endpoints}")
 451     def _with_token(self, token, all_endpoints=False):
 452         return self._swagger_ui_page(all_endpoints, token)