Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def run_cherrypy_server(host="127.0.0.1", port=9008):
"""
Runs a CherryPy server on Python 2.x.
"""
import cherrypy
from ws4py.server.cherrypyserver import WebSocketPlugin, WebSocketTool
from ws4py.websocket import EchoWebSocket
cherrypy.config.update({'server.socket_host': host,
'server.socket_port': port,
'engine.autoreload.on': False,
'log.screen': False})
WebSocketPlugin(cherrypy.engine).subscribe()
cherrypy.tools.websocket = WebSocketTool()
class Root(object):
@cherrypy.expose
def index(self):
pass
config = {
'/': {
'tools.encode.on': False,
'tools.websocket.on': True,
'tools.websocket.handler_cls': EchoWebSocket
}
}
logger = logging.getLogger('autobahn_testsuite')
logger.warning("Serving CherryPy server on %s:%s" % (host, port))
def setup_engine():
# we don't need a HTTP server for this test
cherrypy.server.unsubscribe()
cherrypy.config.update({'log.screen': False})
cherrypy.engine.websocket = WebSocketPlugin(cherrypy.engine)
cherrypy.engine.websocket.subscribe()
cherrypy.engine.websocket.manager.poller = FakePoller()
cherrypy.tools.websocket = WebSocketTool()
config={'/ws': {'tools.websocket.on': True,
'tools.websocket.handler_cls': EchoWebSocket}}
cherrypy.tree.mount(App(), '/', config)
cherrypy.engine.start()
if autosubliminal.USERNAME and autosubliminal.PASSWORD:
users = {s2n(autosubliminal.USERNAME): s2n(autosubliminal.PASSWORD)}
cherrypy.config.update({'tools.auth_digest.on': True,
'tools.auth_digest.realm': 'Auto-Subliminal website',
'tools.auth_digest.get_ha1': auth_digest.get_ha1_dict_plain(users),
'tools.auth_digest.key': 'yek.tsegid_htua.lanimilbuS-otuA' # Can be any random string
})
# Configure our custom json_out_handler (Uncomment if it should be used for any @cherrypy.tools.json_out())
# cherrypy.config.update({'tools.json_out.handler': json_out_handler})
if not restarting:
# Enable websocket plugin
websocket_plugin = WebSocketPlugin(cherrypy.engine)
websocket_plugin.subscribe()
cherrypy.tools.websocket = WebSocketTool()
else:
# When restarting we need to create a new websocket manager thread (you cannot start the same thread twice!)
websocket_plugin.manager = WebSocketManager()
# When restarting we need to clear the httpserver to force the creation of a new one (needed for ip/port change)
cherrypy.server.httpserver = None
tangelo.log_critical("TANGELO", "no such %s '%s' to drop privileges to" % (mode, value))
return 1
# Set the process home directory to be the dropped-down user's.
os.environ["HOME"] = os.path.expanduser("~%s" % (user))
# Perform the actual UID/GID change.
cherrypy.process.plugins.DropPrivileges(cherrypy.engine, uid=uid, gid=gid).subscribe()
else:
tangelo.log_info("TANGELO", "Not performing privilege drop (because not running as superuser)")
# Set up websocket handling. Use the pass-through subclassed version of the
# plugin so we can set a priority on it that doesn't conflict with privilege
# drop.
tangelo.websocket.WebSocketLowPriorityPlugin(cherrypy.engine).subscribe()
cherrypy.tools.websocket = ws4py.server.cherrypyserver.WebSocketTool()
# Replace the stock auth_digest and auth_basic tools with ones that have
# slightly lower priority (so the AuthUpdate tool can run before them).
cherrypy.tools.auth_basic = cherrypy.Tool("before_handler", cherrypy.lib.auth_basic.basic_auth, priority=2)
cherrypy.tools.auth_digest = cherrypy.Tool("before_handler", cherrypy.lib.auth_digest.digest_auth, priority=2)
# Install signal handlers to allow for proper cleanup/shutdown.
for sig in [signal.SIGINT, signal.SIGTERM]:
signal.signal(sig, shutdown)
# Send SIGQUIT to an immediate, ungraceful shutdown instead.
if platform.system() != "Windows":
signal.signal(signal.SIGQUIT, die)
# Start the CherryPy engine.
cherrypy.engine.start()
parser = argparse.ArgumentParser(description='Echo CherryPy Server')
parser.add_argument('--host', default='127.0.0.1')
parser.add_argument('-p', '--port', default=9000, type=int)
parser.add_argument('--ssl', action='store_true')
args = parser.parse_args()
cherrypy.config.update({'server.socket_host': args.host,
'server.socket_port': args.port,
'tools.staticdir.root': os.path.abspath(os.path.join(os.path.dirname(__file__), 'static'))})
if args.ssl:
cherrypy.config.update({'server.ssl_certificate': './server.crt',
'server.ssl_private_key': './server.key'})
WebSocketPlugin(cherrypy.engine).subscribe()
cherrypy.tools.websocket = WebSocketTool()
cherrypy.quickstart(Root(args.host, args.port, args.ssl), '', config={
'/ws': {
'tools.websocket.on': True,
'tools.websocket.handler_cls': ChatWebSocketHandler
},
'/js': {
'tools.staticdir.on': True,
'tools.staticdir.dir': 'js'
}
# encoding: utf-8
from __future__ import absolute_import, print_function, unicode_literals
try:
import cherrypy
from ws4py.server.cherrypyserver import WebSocketPlugin, WebSocketTool
from ws4py.websocket import WebSocket
except ImportError:
raise
cherrypy.tools.websocket = WebSocketTool()
WebSocketPlugin(cherrypy.engine).subscribe()
class SynchronizingWebsocket(WebSocket):
'''
Class to handle requests sent to this websocket connection.
Each instance of this class represents a Salt websocket connection.
Waits to receive a ``ready`` message from the client.
Calls send on it's end of the pipe to signal to the sender on receipt
of ``ready``.
This class also kicks off initial information probing jobs when clients
initially connect. These jobs help gather information about minions, jobs,
and documentation.
'''
def __init__(self, *args, **kwargs): # pylint: disable=E1002
def main():
cherrypy.config.update({
'server.socket_port': 27900,
'request.show_tracebacks': True,
})
WebSocketPlugin(cherrypy.engine).subscribe()
cherrypy.tools.websocket = WebSocketTool()
curr_folder = str(Path(__file__).parent.absolute())
Daemon(cherrypy.engine).subscribe()
config = {
'/': {
'tools.staticdir.root': curr_folder,
},
'/public': {
'tools.staticdir.on': True,
'tools.staticdir.dir': 'public',
},
'/ws': {
'tools.websocket.on': True,
'tools.websocket.handler_cls': WSHandler,
},
}
"""
if __name__ == '__main__':
cherrypy.config.update({
'server.socket_host': '0.0.0.0',
'server.socket_port': 9000,
'tools.staticdir.root': os.path.abspath(os.path.join(os.path.dirname(__file__), 'static'))
}
)
print os.path.abspath(os.path.join(__file__, 'static'))
WebSocketPlugin(cherrypy.engine).subscribe()
cherrypy.tools.websocket = WebSocketTool()
cherrypy.quickstart(Root(), '', config={
'/js': {
'tools.staticdir.on': True,
'tools.staticdir.dir': 'js'
},
'/css': {
'tools.staticdir.on': True,
'tools.staticdir.dir': 'css'
},
'/images': {
'tools.staticdir.on': True,
'tools.staticdir.dir': 'images'
},
'/ws': {
'tools.websocket.on': True,
loggercherry = logging.getLogger('cherrypy.error')
loggercherry.setLevel(app.config.get('LOGLEVEL', logging.ERROR))
cherrypy.server.unsubscribe()
cherrypy.tree.graft(app, '/')
if app.config.get('service', False): # log to file if used as service
cherrypy.config.update({'log.error_file': '{}/service.log'.format(app.config.get('PATH_DATA')), 'log.screen': False})
cherrypy.config.update({'engine.autoreload.on': False, 'engine.SIGHUP': None, 'engine.SIGTERM': None, 'environment': 'embedded'})
cherrypy.server.thread_pool = app.config.get('SERVER_THREADS', 30)
cherrypy.server.thread_pool_max = 20
cherrypy.server.socket_port = int(app.config.get('PORT'))
cherrypy.server.socket_host = "0.0.0.0"
cherrypy.log.screen = False
WebSocketPlugin(cherrypy.engine).subscribe()
cherrypy.tools.websocket = WebSocketTool()
cherrypy.tree.mount(Root(), '/ws', config={'/': {'tools.websocket.on': True, 'tools.trailing_slash.on': False, 'tools.websocket.handler_cls': SocketHandler}})
cherrypy.server.subscribe()
# Start the CherryPy WSGI web server
cherrypy.engine.start()
cherrypy.engine.block()