toolshed/backend/authentication/signature_auth.py

from nacl.exceptions import BadSignatureError
from nacl.signing import VerifyKey
from rest_framework import authentication

from authentication.models import KnownIdentity, ToolshedUser


def split_userhandle_or_throw(userhandle):
    if '@' not in userhandle:
        raise ValueError('Userhandle must be in the format username@domain')
    username, domain = userhandle.split('@')
    if not username:
        raise ValueError('Username cannot be empty')
    if not domain:
        raise ValueError('Domain cannot be empty')
    return username, domain


def split_grouphandle_or_throw(grouphandle):
    if not grouphandle.startswith('#'):
        raise ValueError('Group handle must be in the format #name@domain')
    return split_userhandle_or_throw(grouphandle[1:])


def verify_request(request, raw_request_body):
    authentication_header = request.META.get('HTTP_AUTHORIZATION')

    if not authentication_header:
        raise ValueError('No authentication header provided')

    if not authentication_header.startswith('Signature '):
        raise ValueError('Authorization header must be in the format "Signature author@domain:signature_hex[128]"')

    signature = authentication_header.split('Signature ')[1]

    if ':' not in signature:
        raise ValueError('Authorization header must be in the format "Signature author@domain:signature_hex[128]"')

    author = signature.split(':')[0]
    signature_bytes_hex = signature.split(':')[1]

    if not author or not signature_bytes_hex or len(signature_bytes_hex) != 128:
        raise ValueError('Authorization header must be in the format "Signature author@domain:signature_hex[128]"')

    username, domain = split_userhandle_or_throw(author)

    signed_data = request.build_absolute_uri()

    if request.method == 'POST':
        signed_data += raw_request_body
    elif request.method == 'PUT':
        signed_data += raw_request_body
    elif request.method == 'PATCH':
        signed_data += raw_request_body

    return username, domain, signed_data, signature_bytes_hex


def verify_incoming_friend_request(request, raw_request_body):
    try:
        username, domain, signed_data, signature_bytes_hex = verify_request(request, raw_request_body)
    except ValueError:
        return False
    try:
        befriender = request.data['befriender']
        befriender_key = request.data['befriender_key']
    except KeyError:
        return False
    if not befriender or not befriender_key:
        return False
    if username + "@" + domain != befriender:
        return False
    if len(befriender_key) != 64:
        return False
    verify_key = VerifyKey(bytes.fromhex(befriender_key))
    try:
        verify_key.verify(signed_data.encode('utf-8'), bytes.fromhex(signature_bytes_hex))
        return True
    except BadSignatureError:
        return False


def verify_incoming_group_invite(request, raw_request_body, handle_field, key_field):
    """Self-certifying verifier for the two legs of the group invite/accept dance that land on a
    backend which doesn't have the caller cached as a KnownIdentity yet (see
    docs/design-in-progress/groups-mvp.md): the inviter delivering an invite to the invitee's own
    backend (handle_field='inviter', key_field='inviter_key'), and the invitee accepting on the
    group's home backend (handle_field='invitee', key_field='invitee_key'). Mirrors
    verify_incoming_friend_request exactly, just with configurable field names."""
    try:
        username, domain, signed_data, signature_bytes_hex = verify_request(request, raw_request_body)
    except ValueError:
        return False
    try:
        claimed_handle = request.data[handle_field]
        claimed_key = request.data[key_field]
    except KeyError:
        return False
    if not claimed_handle or not claimed_key:
        return False
    if username + "@" + domain != claimed_handle:
        return False
    if len(claimed_key) != 64:
        return False
    verify_key = VerifyKey(bytes.fromhex(claimed_key))
    try:
        verify_key.verify(signed_data.encode('utf-8'), bytes.fromhex(signature_bytes_hex))
        return True
    except BadSignatureError:
        return False


def authenticate_request_against_known_identities(request, raw_request_body):
    try:
        username, domain, signed_data, signature_bytes_hex = verify_request(request, raw_request_body)
    except ValueError:
        return None
    try:
        author_identity = KnownIdentity.objects.get(username=username, domain=domain)
    except KnownIdentity.DoesNotExist:
        return None
    if author_identity.verify(signed_data, signature_bytes_hex):
        return author_identity
    else:
        return None


def authenticate_request_against_local_users(request, raw_request_body):
    try:
        username, domain, signed_data, signature_bytes_hex = verify_request(request, raw_request_body)
    except ValueError:
        return None
    try:
        author_user = ToolshedUser.objects.get(username=username, domain=domain)
    except ToolshedUser.DoesNotExist:
        return None
    if author_user.public_identity.verify(signed_data, signature_bytes_hex):
        return author_user
    else:
        return None


class SignatureAuthentication(authentication.BaseAuthentication):
    def authenticate(self, request):
        identity = authenticate_request_against_known_identities(request, request.body.decode('utf-8'))
        # Returning a bare None (rather than a (None, None) tuple) tells DRF this
        # authenticator doesn't apply, so it moves on to the next authenticator in the
        # authentication_classes list instead of treating the request as authenticated
        # with an empty user.
        if identity is None:
            return None
        return identity, None


class SignatureAuthenticationLocal(authentication.BaseAuthentication):
    def authenticate(self, request):
        user = authenticate_request_against_local_users(request, request.body.decode('utf-8'))
        if user is None:
            return None
        return user, None