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