2018-07-15 15:27:35 +00:00
|
|
|
from functools import wraps
|
2018-07-15 15:29:16 +00:00
|
|
|
from base64 import b64encode, b64decode
|
|
|
|
from copy import deepcopy
|
2018-07-15 15:27:35 +00:00
|
|
|
|
|
|
|
from cryptography.hazmat.backends import default_backend
|
|
|
|
from cryptography.hazmat.primitives.hashes import SHA256
|
|
|
|
from cryptography.hazmat.primitives.hmac import HMAC as _HMAC
|
|
|
|
from cryptography.exceptions import InvalidSignature
|
|
|
|
|
2019-07-24 13:17:16 +00:00
|
|
|
from tfw.internals.networking import message_bytes
|
2018-07-15 15:30:19 +00:00
|
|
|
|
|
|
|
|
2018-07-15 15:29:16 +00:00
|
|
|
def sign_message(key, message):
|
2019-07-04 13:51:55 +00:00
|
|
|
message.pop('scope', None)
|
2019-07-04 13:19:48 +00:00
|
|
|
message.pop('signature', None)
|
2018-07-16 08:29:06 +00:00
|
|
|
signature = message_signature(key, message)
|
2018-07-15 15:29:16 +00:00
|
|
|
message['signature'] = b64encode(signature).decode()
|
|
|
|
|
|
|
|
|
2018-07-16 08:29:06 +00:00
|
|
|
def message_signature(key, message):
|
|
|
|
return HMAC(key, message_bytes(message)).signature
|
|
|
|
|
|
|
|
|
2018-07-15 15:29:16 +00:00
|
|
|
def verify_message(key, message):
|
2019-07-04 13:51:55 +00:00
|
|
|
message.pop('scope', None)
|
2018-07-15 15:29:16 +00:00
|
|
|
message = deepcopy(message)
|
|
|
|
try:
|
|
|
|
signature_b64 = message.pop('signature')
|
|
|
|
signature = b64decode(signature_b64)
|
2018-07-16 08:29:06 +00:00
|
|
|
actual_signature = message_signature(key, message)
|
2018-07-15 15:29:16 +00:00
|
|
|
return signature == actual_signature
|
|
|
|
except KeyError:
|
|
|
|
return False
|
|
|
|
|
2018-07-15 15:27:35 +00:00
|
|
|
|
|
|
|
class HMAC:
|
|
|
|
def __init__(self, key, message):
|
|
|
|
self.key = key
|
|
|
|
self.message = message
|
|
|
|
self._hmac = _HMAC(
|
|
|
|
key=key,
|
|
|
|
algorithm=SHA256(),
|
|
|
|
backend=default_backend()
|
|
|
|
)
|
|
|
|
|
|
|
|
def _reload_if_finalized(f):
|
|
|
|
# pylint: disable=no-self-argument,not-callable
|
|
|
|
@wraps(f)
|
|
|
|
def wrapped(instance, *args, **kwargs):
|
|
|
|
if getattr(instance, '_finalized', False):
|
|
|
|
instance.__init__(instance.key, instance.message)
|
|
|
|
ret_val = f(instance, *args, **kwargs)
|
|
|
|
setattr(instance, '_finalized', True)
|
|
|
|
return ret_val
|
|
|
|
return wrapped
|
|
|
|
|
|
|
|
@property
|
|
|
|
@_reload_if_finalized
|
|
|
|
def signature(self):
|
|
|
|
self._hmac.update(self.message)
|
|
|
|
signature = self._hmac.finalize()
|
|
|
|
return signature
|
|
|
|
|
|
|
|
@_reload_if_finalized
|
|
|
|
def verify(self, signature):
|
|
|
|
self._hmac.update(self.message)
|
|
|
|
try:
|
|
|
|
self._hmac.verify(signature)
|
|
|
|
return True
|
|
|
|
except InvalidSignature:
|
|
|
|
return False
|