Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import tornado.web as web
from jupyterhub.handlers.base import BaseHandler
import logging
log = logging.getLogger()
class HomeHandler(BaseHandler):
async def get(self):
html = self.render_template("stochss-home.html")
self.finish(html)
class ModelBrowserHandler(BaseHandler):
@web.authenticated
async def get(self):
html = self.render_template("stochss-file-browser.html")
self.finish(html)
class ModelEditorHandler(BaseHandler):
@web.authenticated
async def get(self, model_name):
html = self.render_template("stochss-model-editor.html")
self.finish(html)
class JobEditorHandler(BaseHandler):
@web.authenticated
}
)
# sort oauth clients by last activity, created
def sort_key(client):
return (client['last_activity'] or never, client['created'] or never)
oauth_clients = sorted(oauth_clients, key=sort_key, reverse=True)
html = self.render_template(
'token.html', api_tokens=api_tokens, oauth_clients=oauth_clients
)
self.finish(html)
class ProxyErrorHandler(BaseHandler):
"""Handler for rendering proxy error pages"""
def get(self, status_code_s):
status_code = int(status_code_s)
status_message = responses.get(status_code, 'Unknown HTTP Error')
# build template namespace
hub_home = url_path_join(self.hub.base_url, 'home')
message_html = ''
if status_code == 503:
message_html = ' '.join(
[
"Your server appears to be down.",
"Try restarting it <a href="%s">from the hub</a>" % hub_home,
]
)
logout_handler_self._backend_logout_cleanup(logout_handler_self.current_user.name)
# This is a little janky, but there was a misspelling in a prior version
# where someone could have set the wrong flag because of the documentation.
# We will honor the misspelling until we rev the version, and then we will
# break backward compatibility.
forward_on_logout = True if authenticator_self.slo_forward_on_logout else False
forwad_on_logout = True if authenticator_self.slo_forwad_on_logout else False
if forward_on_logout or forwad_on_logout:
authenticator_self._get_redirect_from_metadata_and_redirect('md:SingleLogoutService',
logout_handler_self)
else:
html = logout_handler_self.render_template('logout.html')
logout_handler_self.finish(html)
class SAMLMetaHandler(BaseHandler):
async def get(meta_handler_self):
xml_content = authenticator_self._make_sp_metadata(meta_handler_self)
meta_handler_self.set_header('Content-Type', 'text/xml')
meta_handler_self.write(xml_content)
return [('/login', SAMLLoginHandler),
('/hub/login', SAMLLoginHandler),
('/logout', SAMLLogoutHandler),
('/hub/logout', SAMLLogoutHandler),
('/metadata', SAMLMetaHandler),
('/hub/metadata', SAMLMetaHandler)]
except:
# In this case, any side effect must be avoided.
ns['no_spawner_check'] = True
html = self.render_template('error.html', **ns)
self.write(html)
class Template404(BaseHandler):
"""Render our 404 template"""
async def prepare(self):
await super().prepare()
raise web.HTTPError(404)
class PrefixRedirectHandler(BaseHandler):
"""Redirect anything outside a prefix inside.
Redirects /foo to /prefix/foo, etc.
"""
def get(self):
uri = self.request.uri
# Since self.base_url will end with trailing slash.
# Ensure uri will end with trailing slash when matching
# with self.base_url.
if not uri.endswith('/'):
uri += '/'
if uri.startswith(self.base_url):
path = self.request.uri[len(self.base_url):]
else:
path = self.request.path
self.redirect(url_path_join(
"""Null Authenticator for JupyterHub
For cases where authentication should be disabled,
e.g. only allowing access via API tokens.
"""
# Copyright (c) Jupyter Development Team.
# Distributed under the terms of the Modified BSD License.
from jupyterhub.auth import Authenticator
from jupyterhub.handlers.base import BaseHandler
__version__ = '1.1.0.dev'
class NullLoginHandler(BaseHandler):
def get(self):
raise web.HTTPError(403, "Login is not supported")
class NullAuthenticator(Authenticator):
# auto_login skips 'Login with...' page on Hub 0.8
auto_login = True
# for Hub 0.7, show 'login with...'
login_service = 'null'
def get_handlers(self, app):
return [('/nologin', NullLoginHandler)]
If the user is not logged in, send to login URL, redirecting back here.
.. versionadded:: 0.7
"""
@web.authenticated
def get(self, path):
user = self.get_current_user()
url = url_path_join(user.url, path)
if self.request.query:
# FIXME: use urlunparse instead?
url += '?' + self.request.query
self.redirect(url)
class CSPReportHandler(BaseHandler):
'''Accepts a content security policy violation report'''
@web.authenticated
def post(self):
'''Log a content security policy violation report'''
self.log.warning(
"Content security violation: %s",
self.request.body.decode('utf8', 'replace')
)
# Report it to statsd as well
self.statsd.incr('csp_report')
class AddSlashHandler(BaseHandler):
"""Handler for adding trailing slash to URLs that need them"""
def get(self, *args):
src = urlparse(self.request.uri)
html = self.render_template(
'admin.html',
current_user=self.current_user,
admin_access=self.settings.get('admin_access', False),
users=users,
running=running,
sort={s: o for s, o in zip(sorts, orders)},
allow_named_servers=self.allow_named_servers,
named_server_limit_per_user=self.named_server_limit_per_user,
server_version='{} {}'.format(__version__, self.version_hash),
)
self.finish(html)
class TokenPageHandler(BaseHandler):
"""Handler for page requesting new API tokens"""
@web.authenticated
def get(self):
never = datetime(1900, 1, 1)
user = self.current_user
def sort_key(token):
return (token.last_activity or never, token.created or never)
now = datetime.utcnow()
api_tokens = []
for token in sorted(user.api_tokens, key=sort_key, reverse=True):
if token.expires_at and token.expires_at < now:
self.db.delete(token)
class ModelBrowserHandler(BaseHandler):
@web.authenticated
async def get(self):
html = self.render_template("stochss-file-browser.html")
self.finish(html)
class ModelEditorHandler(BaseHandler):
@web.authenticated
async def get(self, model_name):
html = self.render_template("stochss-model-editor.html")
self.finish(html)
class JobEditorHandler(BaseHandler):
@web.authenticated
async def get(self, model_name):
html = self.render_template("stochss-job-manager.html")
self.finish(html)
# render the template
try:
html = self.render_template('%s.html' % status_code, **ns)
except TemplateNotFound:
self.log.debug("No template for %d", status_code)
try:
html = self.render_template('error.html', **ns)
except:
# In this case, any side effect must be avoided.
ns['no_spawner_check'] = True
html = self.render_template('error.html', **ns)
self.write(html)
class Template404(BaseHandler):
"""Render our 404 template"""
def prepare(self):
raise web.HTTPError(404)
class PrefixRedirectHandler(BaseHandler):
"""Redirect anything outside a prefix inside.
Redirects /foo to /prefix/foo, etc.
"""
def get(self):
uri = self.request.uri
# Since self.base_url will end with trailing slash.
# Ensure uri will end with trailing slash when matching
# with self.base_url.
if not uri.endswith('/'):
Attributes
----------
model_path : str
Path to target model within user pod container.
'''
checkUserOrRaise(self) # User validation
user = self.current_user.name # Get User Name
model_path = model_path.replace(" ", "\ ")
full_path = '/home/jovyan/{0}'.format(model_path) #full path to model
client, user_pod = stochss_kubernetes.load_kube_client(user) # Load Kube client
stochss_kubernetes.write_to_pod(client,
user_pod, full_path, self.request.body.decode())
class RunModelAPIHandler(BaseHandler):
'''
########################################################################
Handler for running a model from the model editor.
########################################################################
'''
@web.authenticated
async def get(self, run_cmd, outfile, model_path):
'''
Run the model with a 5 second timeout. Only the data from the first
trajectory is transferred to the hub container. Data is transferred
to hub container as JSON string.
Attributes
----------
run_cmd : str