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=None, reactor=None, personality=None):
# base ctor
NativeProcess.__init__(self, config=config, reactor=reactor, personality=personality)
# Release (public) key
self._release_pubkey = _read_release_key()
# Node (private) key (as a string, in hex)
node_key_hex = _read_node_key(self.config.extra.cbdir, private=True)['hex']
privkey = nacl.signing.SigningKey(node_key_hex, encoder=nacl.encoding.HexEncoder)
# WAMP-cryptosign signing key
self._node_key = cryptosign.SigningKey(privkey)
log = make_logger()
try:
if not reactor:
from twisted.internet import reactor
extra = {
'key': binascii.a2b_hex(_read_node_key(cbdir, private=True)['hex']),
}
comp = Component(
transports=[backend_config['transport']],
realm=realm,
extra=extra,
authentication={
"cryptosign": {
"privkey": _read_node_key(cbdir, private=True)['hex'],
}
},
)
ready = Deferred()
@comp.on_join
def joined(session, details):
ready.callback(session)
@comp.on_disconnect
def disconnect(session, was_clean=False):
if not ready.called:
ready.errback(Exception("Disconnected unexpectedly"))
comp.start(reactor)
return ready
def _backend_connected(backend_session):
try:
# wait for the WAMP-level transport to connect
yield backend_session._on_connect
# node private key
key = _read_node_key(self._controller._cbdir, private=False)
# FIXME
authmethods = [
'{}-proxy'.format(x)
for x in backend_session._authenticators.keys()
]
# authmethods = ['cryptosign-proxy']
self.log.info('Proxy backend session authenticating using authmethods={authmethods} ..',
authmethods=authmethods)
backend_session.join(
accept.realm,
authmethods=authmethods,
authid=None,
authrole=None,
authextra={
},
'serializer': 'cbor',
'url': 'rs://localhost'
}
}
:param frontend_session: The frontend proxy session for which to create a mapped backend connection.
:param cbdir: The node directory.
"""
from twisted.internet import reactor
connected_d = Deferred()
backend = _create_transport(0, backend_config['transport'])
key = _read_node_key(cbdir, private=True)
def create_session():
session = ProxyBackendSession()
# we will do cryptosign authentication to any backend
if 'auth' in backend_config and 'cryptosign-proxy' in backend_config['auth']:
session.add_authenticator(create_authenticator("cryptosign", privkey=key['hex']))
# we allow anonymous authentication to just unix-sockets
# currently. I don't think it's a good idea to allow any
# anonymous auth to "real" backends over TCP due to
# cross-protocol hijinks (and if a Web browser is running on
# that machine, any website can try to access the "real"
# backend)
if 'auth' not in backend_config or 'anonymous-proxy' in backend_config['auth']:
if backend_config['transport']['endpoint']['type'] == 'unix':
def _run_command_keys(options, reactor, personality):
"""
Subcommand "crossbar keys".
"""
log = make_logger()
# Generate a new node key pair (2 files), load and check
_maybe_generate_key(options.cbdir)
# Print keys
# Release (public) key
release_pubkey = _read_release_key()
# Node key
node_key = _read_node_key(options.cbdir, private=options.private)
if options.private:
key_title = 'Crossbar.io Node PRIVATE Key'
else:
key_title = 'Crossbar.io Node PUBLIC Key'
log.info('')
log.info('{key_title}', key_title=hl('Crossbar Software Release Key', color='yellow', bold=True))
log.info('base64: {release_pubkey}', release_pubkey=release_pubkey['base64'])
log.info(release_pubkey['qrcode'].strip())
log.info('')
log.info('{key_title}', key_title=hl(key_title, color='yellow', bold=True))
log.info('hex: {node_key}', node_key=node_key['hex'])
log.info(node_key['qrcode'].strip())
log.info('')