Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def wsgi_server():
"""Set up and tear down a Cheroot WSGI server instance."""
for srv in cheroot_server(cheroot.wsgi.Server):
yield srv
def setup(self):
"""
Generate the flask App and prepare the HTTP server.
:return: server host and port tuple
:rtype: ``Tuple[str, int]``
"""
app, _ = generate_interactive_api(self.cfg.ihandler)
self._server = wsgi.Server((self.cfg.host, self.cfg.port), app)
self._server.prepare()
return self._server.bind_addr
def start(self):
d = PathInfoDispatcher({"/": self.app})
self.server = Server(("0.0.0.0", self.proxyport), d)
self.proxy_thread = threading.Thread(target=self.server.start)
self.proxy_thread.daemon = True
self.proxy_thread.start()
BitcoinD.start(self)
# Now that bitcoind is running on the real rpcport, let's tell all
# future callers to talk to the proxyport. We use the bind_addr as a
# signal that the port is bound and accepting connections.
while self.server.bind_addr[1] == 0:
pass
self.proxiedport = self.rpcport
self.rpcport = self.server.bind_addr[1]
logging.debug(
"bitcoind reverse proxy listening on {}, forwarding to {}".format(
self.rpcport, self.proxiedport
)
print makejson(data)
return makejson(data)
if __name__ == '__main__':
port = DEFAULT_PORT_NO
options, args = getopt.getopt(sys.argv[1:], "", ["help", "port="])
for op, arg1 in options:
if op == "--help":
usageguide()
sys.exit(2)
elif op == "--port":
port = int(arg1)
urls = ("/.*", "app")
apps = web.application(urls, globals())
server = wsgi.Server(("0.0.0.0", port),app,server_name='localhost')
print "The server is hosted on port:",(port)
try:
server.start()
#apps.run(port)
except KeyboardInterrupt:
server.stop()
def run(self, handler): # pragma: no cover
from cheroot import wsgi
from cheroot.ssl import builtin
self.options['bind_addr'] = (self.host, self.port)
self.options['wsgi_app'] = handler
certfile = self.options.pop('certfile', None)
keyfile = self.options.pop('keyfile', None)
chainfile = self.options.pop('chainfile', None)
server = wsgi.Server(**self.options)
if certfile and keyfile:
server.ssl_adapter = builtin.BuiltinSSLAdapter(
certfile, keyfile, chainfile)
try:
server.start()
finally:
server.stop()
#############
def __init__(self, machine_id, port, ui):
self.machine_id = machine_id
# We only use 1 thread, such that subsequent requests don't run into
# SQLite access problems.
from cheroot import wsgi
self.wsgi_server = wsgi.Server\
(("0.0.0.0", port), self.wsgi_app, server_name="localhost",
numthreads=1, timeout=1000)
Partner.__init__(self, ui)
self.text_format = XMLFormat()
self.sessions = {} # {session_token: session}
self.session_token_for_user = {} # {user_name: session_token}
try:
from cheroot import server, wsgi
# from cheroot.ssl.builtin import BuiltinSSLAdapter
# import cheroot.ssl.pyopenssl
except ImportError:
# if config["verbose"] >= 1:
print("*" * 78)
print("ERROR: Could not import Cheroot.")
print("Try `pip install cheroot` or specify another server using the --server option.")
print("*" * 78)
raise
server_name = "WsgiDAV/%s %s Python/%s" % (
__version__,
wsgi.Server.version,
PYTHON_VERSION)
wsgi.Server.version = server_name
# Support SSL
ssl_certificate = _get_checked_path(config.get("ssl_certificate"))
ssl_private_key = _get_checked_path(config.get("ssl_private_key"))
ssl_certificate_chain = _get_checked_path(config.get("ssl_certificate_chain"))
ssl_adapter = config.get("ssl_adapter", "builtin")
protocol = "http"
if ssl_certificate and ssl_private_key:
ssl_adapter = server.get_ssl_adapter_class(ssl_adapter)
wsgi.Server.ssl_adapter = ssl_adapter(
ssl_certificate, ssl_private_key, ssl_certificate_chain)
protocol = "https"
if config["verbose"] >= 1:
print("SSL / HTTPS enabled. Adapter: {}".format(ssl_adapter))
pass
new_version_checker.start(settings_manager.new_version_check_interval)
debug = config.debug
if debug:
secret_key = 'Secret!'
token = 'monitorrent'
else:
secret_key = os.urandom(24)
token = ''.join(random.choice(string.ascii_letters) for _ in range(8))
app = create_app(secret_key, token, tracker_manager, clients_manager, notifier_manager, settings_manager,
engine_runner, log_manager, new_version_checker)
server_start_params = (config.ip, config.port)
server = wsgi.Server(server_start_params, app)
print('Server started on {0}:{1}'.format(*server_start_params))
try:
server.start()
except KeyboardInterrupt:
print('Stopping engine')
engine_runner.stop()
print('Stopping new_version_checker')
new_version_checker.stop()
server.stop()
print('Server stopped')
'ms_mount': False,
'show_user': True,
'ms_sharepoint_plugin': True,
'ms_sharepoint_urls': False,
'response_trailer': False},
'port': self.listenPort, # Specifying listening port
'provider_mapping': {'/': self.WRoot} #Specifying root folder
}
app = WsgiDAVApp(config)
server_args = {
"bind_addr": (config["host"], config["port"]),
"wsgi_app": app,
}
server = wsgi.Server(**server_args)
try:
log = logging.getLogger('wsgidav')
log.raiseExceptions = False # hack to avoid false exceptions
log.propagate = True
log.setLevel(logging.INFO)
server.start()
except KeyboardInterrupt:
logging.info(" [!] Ctrl + C detected, closing WebDAV sever")
server.stop()