-
Notifications
You must be signed in to change notification settings - Fork 272
/
handlers.py
137 lines (109 loc) · 4.45 KB
/
handlers.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
"""Tornado handlers for api specifications."""
# Copyright (c) Jupyter Development Team.
# Distributed under the terms of the Modified BSD License.
import json
import os
from typing import Dict, List
from jupyter_core.utils import ensure_async
from tornado import web
from jupyter_server._tz import isoformat, utcfromtimestamp
from jupyter_server.auth import authorized
from ...base.handlers import APIHandler, JupyterHandler
AUTH_RESOURCE = "api"
class APISpecHandler(web.StaticFileHandler, JupyterHandler):
"""A spec handler for the REST API."""
auth_resource = AUTH_RESOURCE
def initialize(self):
"""Initialize the API spec handler."""
web.StaticFileHandler.initialize(self, path=os.path.dirname(__file__))
@web.authenticated
@authorized
def get(self):
"""Get the API spec."""
self.log.warning("Serving api spec (experimental, incomplete)")
return web.StaticFileHandler.get(self, "api.yaml")
def get_content_type(self):
"""Get the content type."""
return "text/x-yaml"
class APIStatusHandler(APIHandler):
"""An API status handler."""
auth_resource = AUTH_RESOURCE
_track_activity = False
@web.authenticated
@authorized
async def get(self):
"""Get the API status."""
# if started was missing, use unix epoch
started = self.settings.get("started", utcfromtimestamp(0))
started = isoformat(started)
kernels = await ensure_async(self.kernel_manager.list_kernels())
total_connections = sum(k["connections"] for k in kernels)
last_activity = isoformat(self.application.last_activity()) # type:ignore[attr-defined]
model = {
"started": started,
"last_activity": last_activity,
"kernels": len(kernels),
"connections": total_connections,
}
self.finish(json.dumps(model, sort_keys=True))
class IdentityHandler(APIHandler):
"""Get the current user's identity model"""
@web.authenticated
def get(self):
"""Get the identity model."""
permissions_json: str = self.get_argument("permissions", "")
bad_permissions_msg = f'permissions should be a JSON dict of {{"resource": ["action",]}}, got {permissions_json!r}'
if permissions_json:
try:
permissions_to_check = json.loads(permissions_json)
except ValueError as e:
raise web.HTTPError(400, bad_permissions_msg) from e
if not isinstance(permissions_to_check, dict):
raise web.HTTPError(400, bad_permissions_msg)
else:
permissions_to_check = {}
permissions: Dict[str, List[str]] = {}
user = self.current_user
for resource, actions in permissions_to_check.items():
if (
not isinstance(resource, str)
or not isinstance(actions, list)
or not all(isinstance(action, str) for action in actions)
):
raise web.HTTPError(400, bad_permissions_msg)
allowed = permissions[resource] = []
for action in actions:
if self.authorizer.is_authorized(self, user=user, resource=resource, action=action):
allowed.append(action)
identity: Dict = self.identity_provider.identity_model(user)
model = {
"identity": identity,
"permissions": permissions,
}
self.write(json.dumps(model))
class PathResolverHandler(APIHandler):
"""Path resolver handler."""
auth_resource = AUTH_RESOURCE
_track_activity = False
@web.authenticated
@authorized
async def get(self):
"""Resolve the path."""
path = self.get_query_argument("path")
kernel_uuid = self.get_query_argument("kernel", default=None)
scopes = {"server": self.contents_manager}
if kernel_uuid:
scopes["kernel"] = self.kernel_manager.get_kernel(kernel_uuid)
resolved = [
{"scope": name, "path": await ensure_async(scope.resolve_path(path))}
for name, scope in scopes.items()
if hasattr(scope, "resolve_path")
]
response = {"resolved": [entry for entry in resolved if entry["path"] is not None]}
self.finish(json.dumps(response))
default_handlers = [
(r"/api/spec.yaml", APISpecHandler),
(r"/api/status", APIStatusHandler),
(r"/api/me", IdentityHandler),
(r"/api/resolvePath", PathResolverHandler),
]