Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self, config, templates):
"""
:param config: Crossbar transport configuration.
:type config: dict
"""
options = config.get('options', {})
server = "Crossbar/{}".format(crossbar.__version__)
externalPort = options.get('external_port', None)
WebSocketServerFactory.__init__(self,
url=config.get('url', None),
server=server,
externalPort=externalPort)
# transport configuration
self._config = config
# Jinja2 templates for 404 etc
self._templates = templates
# set WebSocket options
set_websocket_options(self, options)
def sendServerStatus(self, redirectUrl=None, redirectAfter=0):
"""
Used to send out server status/version upon receiving a HTTP/GET without
upgrade to WebSocket header (and option serverStatus is True).
"""
try:
page = self.factory._templates.get_template('cb_ws_testee_status.html')
self.sendHtml(page.render(redirectUrl=redirectUrl,
redirectAfter=redirectAfter,
cbVersion=crossbar.__version__,
wsUri=self.factory.url))
except Exception as e:
self.log.warn("Error rendering WebSocket status page template: {e}", e=e)
def _read_release_key():
release_pubkey_file = 'crossbar-{}.pub'.format('-'.join(crossbar.__version__.split('.')[0:2]))
release_pubkey_path = os.path.join(pkg_resources.resource_filename('crossbar', 'common/keys'), release_pubkey_file)
release_pubkey_hex = binascii.b2a_hex(cryptosign._read_signify_ed25519_pubkey(release_pubkey_path)).decode('ascii')
with open(release_pubkey_path) as f:
release_pubkey_base64 = f.read().splitlines()[1]
release_pubkey_qrcode = cryptosign._qrcode_from_signify_ed25519_pubkey(release_pubkey_path)
release_pubkey = {
'base64': release_pubkey_base64,
'hex': release_pubkey_hex,
'qrcode': release_pubkey_qrcode
}
return release_pubkey
"""
Software personality for Crossbar.io OSS.
This is a policy class that configures various parts of Crossbar.io's
behavior.
"""
log = txaio.make_logger()
NAME = 'standalone'
TITLE = _TITLE
DESC = crossbar.__doc__
BANNER = _BANNER.format(title=_TITLE, version=crossbar.__version__, year=time.strftime('%Y'))
LEGAL = ('crossbar', 'LEGAL')
LICENSE = ('crossbar', 'LICENSE')
LICENSE_FOR_API = ('crossbar', 'LICENSE-FOR-API')
LICENSES_OSS = ('crossbar', 'LICENSES-OSS')
# a list of directories to serach Jinja2 templates for
# rendering various web resources. this must be a list
# of _pairs_ to be used with pkg_resources.resource_filename()!
TEMPLATE_DIRS = [('crossbar', 'webservice/templates')]
WEB_SERVICE_CHECKERS = {
'none': None,
'path': checkconfig.check_web_path_service_path,
'redirect': checkconfig.check_web_path_service_redirect,
def __init__(self, reactor, path_config):
"""
:param path_config: The path configuration of the Web transport resource.
:type path_config: dict
"""
self.reactor = reactor
self.path_config = path_config
url = path_config.get('url', None)
options = path_config.get('options', {})
showServerVersion = options.get('show_server_version', True)
if showServerVersion:
server = "Crossbar/{}".format(crossbar.__version__)
else:
server = "Crossbar"
externalPort = options.get('external_port', None)
protocols = None
headers = None
websocket.WebSocketServerFactory.__init__(self,
reactor=reactor,
url=url,
protocols=protocols,
server=server,
headers=headers,
externalPort=externalPort)
# set WebSocket options
def sendServerStatus(self, redirectUrl=None, redirectAfter=0):
"""
Used to send out server status/version upon receiving a HTTP/GET without
upgrade to WebSocket header (and option serverStatus is True).
"""
try:
page = self.factory._templates.get_template('cb_ws_status.html')
self.sendHtml(page.render(redirectUrl=redirectUrl,
redirectAfter=redirectAfter,
cbVersion=crossbar.__version__,
wsUri=self.factory.url,
peer=self.peer,
workerPid=os.getpid()))
except Exception:
self.log.failure("Error rendering WebSocket status page template: {log_failure.value}")
from autobahn.util import utcstr
from autobahn.wamp import ApplicationError, PublishOptions
import crossbar
from crossbar._util import hltype, hlid, hlval
from crossbar.common.twisted.web import Site
from crossbar.common.twisted.endpoint import create_listening_port_from_config
from crossbar.bridge.mqtt.wamp import WampMQTTServerFactory
from crossbar.router.protocol import WampRawSocketServerFactory, WampWebSocketServerFactory
from crossbar.router.unisocket import UniSocketServerFactory
from crossbar.webservice.flashpolicy import FlashPolicyFactory
from crossbar.worker.testee import WebSocketTesteeServerFactory, StreamTesteeServerFactory
# monkey patch the Twisted Web server identification
twisted.web.server.version = "Crossbar/{}".format(crossbar.__version__)
class RouterTransport(object):
"""
A (listening) transport running on a router worker.
"""
STATE_CREATED = 1
STATE_STARTING = 2
STATE_STARTED = 3
STATE_FAILED = 4
STATE_STOPPING = 5
STATE_STOPPED = 6
STATES = {
STATE_CREATED: "created",
STATE_STARTING: "starting",
supported_serializers.append('Flatbuffers')
except ImportError:
pass
v.supported_serializers = supported_serializers
# LMDB
try:
import lmdb # noqa
lmdb_lib_ver = '.'.join([str(x) for x in lmdb.version()])
v.lmdb_ver = '{}/lmdb-{}'.format(_get_version(lmdb), lmdb_lib_ver)
except ImportError:
pass
# crossbar
v.crossbar_ver = crossbar.__version__
# crossbarfx
try:
import crossbarfx # noqa
v.crossbarfx_ver = _get_version(crossbarfx)
except ImportError:
pass
# zlmdb
try:
import zlmdb # noqa
v.zlmdb_ver = _get_version(zlmdb)
except ImportError:
pass
# numpy
def _delayedRender(self, node_info, request):
try:
peer = request.transport.getPeer()
peer = '{}:{}'.format(peer.host, peer.port)
except:
peer = '?:?'
s = self._page.render(cbVersion=crossbar.__version__,
workerPid=self._pid,
peer=peer,
**node_info)
request.write(s.encode('utf8'))
request.finish()
def render_GET(self, request):
page = templates.get_template('cb_web_404.html')
s = page.render(cbVersion = crossbar.__version__,
wsUri = wsUri,
restUri = restUri)
return s.encode('utf8')