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

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