Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@public
def compute_wcs(key, challenge):
"""
Compute an WAMP-CRA authentication signature from an authentication
challenge and a (derived) key.
:param key: The key derived (via PBKDF2) from the secret.
:type key: bytes
:param challenge: The authentication challenge to sign.
:type challenge: bytes
:return: The authentication signature.
:rtype: bytes
"""
assert(type(key) in [six.text_type, six.binary_type])
assert(type(challenge) in [six.text_type, six.binary_type])
if type(key) == six.text_type:
@util.public
def public_key(self, binary=False):
"""
Returns the public key part of a signing key or the (public) verification key.
:returns: The public key in Hex encoding.
:rtype: str or None
"""
if isinstance(self._key, signing.SigningKey):
key = self._key.verify_key
else:
key = self._key
if binary:
return key.encode()
else:
return key.encode(encoder=encoding.HexEncoder).decode('ascii')
@public
@abc.abstractmethod
def get_payload_codec(self):
"""
Get the current payload codec (if any) for the session.
@public
def onUserError(self, fail, msg):
"""
Implements :func:`autobahn.wamp.interfaces.ISession.onUserError`
"""
if isinstance(fail.value, exception.ApplicationError):
self.log.warn('{klass}.onUserError(): "{msg}"',
klass=self.__class__.__name__,
msg=fail.value.error_message())
else:
self.log.error(
'{klass}.onUserError(): "{msg}"\n{traceback}',
klass=self.__class__.__name__,
msg=msg,
traceback=txaio.failure_format_traceback(fail),
)
@public
def check_totp(secret, ticket):
"""
Check a TOTP value received from a principal trying to authenticate against
the expected value computed from the secret shared between the principal and
the authenticating entity.
The Internet can be slow, and clocks might not match exactly, so some
leniency is allowed. RFC6238 recommends looking an extra time step in either
direction, which essentially opens the window from 30 seconds to 90 seconds.
:param secret: The secret shared between the principal (eg a client) that
is authenticating, and the authenticating entity (eg a server).
:type secret: unicode
:param ticket: The TOTP value to be checked.
:type ticket: unicode
@public
@abc.abstractmethod
def onConnecting(self, transport_details):
"""
This method is called when we've connected, but before the handshake is done.
@public
def encode(self, is_originating, uri, args=None, kwargs=None):
"""
Encrypt the given WAMP URI, args and kwargs into an EncodedPayload instance, or None
if the URI should not be encrypted.
"""
assert(type(is_originating) == bool)
assert(type(uri) == six.text_type)
assert(args is None or type(args) in (list, tuple))
assert(kwargs is None or type(kwargs) == dict)
box = self._get_box(is_originating, uri)
if not box:
# if we didn't find a crypto box, then return None, which
# signals that the payload travel unencrypted (normal)
return None
@public
def generate_key(self):
"""
Generate a new private key and return a pair with the base64 encodings
of (priv_key, pub_key).
"""
key = PrivateKey.generate()
priv_key = key.encode(encoder=Base64Encoder)
pub_key = key.public_key.encode(encoder=Base64Encoder)
return (u'{}'.format(priv_key), u'{}'.format(pub_key))
@public
def listenWS(factory, contextFactory=None, backlog=50, interface=''):
"""
Listen for incoming WebSocket connections from clients. The connection parameters like
listening port and others are provided via the factory.
:param factory: The WebSocket protocol factory to be used for creating server protocol instances.
:type factory: An :class:`autobahn.websocket.WebSocketServerFactory` instance.
:param contextFactory: SSL context factory, required for secure WebSocket connections ("wss").
:type contextFactory: A twisted.internet.ssl.ContextFactory.
:param backlog: Size of the listen queue.
:type backlog: int
:param interface: The interface (derived from hostname given) to bind to, defaults to '' (all).
:type interface: str
@public
@abc.abstractmethod
def encode(self, is_originating, uri, args=None, kwargs=None):
"""
Encodes application payload.