add webauthn crud endpoints

added some simple validation to the delete endpoint for sanity, but
generally my assumption is that more validation will happen on the admin
side.

noteably im not checking whether the credentials are duplicated, nor is
there a uniqueness constraint in the database - I'm not sure if the
credential blob will always reliably be equivalent, and I believe the
browser should hopefully take care of dupes.

app/__init__.py

@@ -178,6 +178,7 @@ def register_blueprint(application):
     )
     from app.upload.rest import upload_blueprint
     from app.user.rest import user_blueprint
+    from app.webauthn.rest import webauthn_blueprint
 
     service_blueprint.before_request(requires_admin_auth)
     application.register_blueprint(service_blueprint, url_prefix='/service')
@@ -185,6 +186,9 @@ def register_blueprint(application):
     user_blueprint.before_request(requires_admin_auth)
     application.register_blueprint(user_blueprint, url_prefix='/user')
 
+    webauthn_blueprint.before_request(requires_admin_auth)
+    application.register_blueprint(webauthn_blueprint)
+
     template_blueprint.before_request(requires_admin_auth)
     application.register_blueprint(template_blueprint)
 

app/dao/webauthn_credential_dao.py

@@ -0,0 +1,39 @@
+from app import db
+from app.dao.dao_utils import autocommit
+from app.models import WebauthnCredential
+
+
+def dao_get_webauthn_credential_by_id(webauthn_credential_id):
+    return WebauthnCredential.query.filter_by(
+        id=webauthn_credential_id
+    ).one()
+
+
+@autocommit
+def dao_create_webauthn_credential(
+    *,
+    user_id,
+    name,
+    credential_data,
+    registration_response,
+):
+    webauthn_credential = WebauthnCredential(
+        user_id=user_id,
+        name=name,
+        credential_data=credential_data,
+        registration_response=registration_response
+    )
+    db.session.add(webauthn_credential)
+    return webauthn_credential
+
+
+@autocommit
+def dao_update_webauthn_credential_name(webauthn_credential, new_name):
+    webauthn_credential.name = new_name
+    db.session.add(webauthn_credential)
+    return webauthn_credential
+
+
+@autocommit
+def dao_delete_webauthn_credential(webauthn_credential):
+    db.session.delete(webauthn_credential)

app/models.py

@@ -2615,3 +2615,14 @@ class WebauthnCredential(db.Model):
 
     created_at = db.Column(db.DateTime, nullable=False, default=datetime.datetime.utcnow)
     updated_at = db.Column(db.DateTime, nullable=True, onupdate=datetime.datetime.utcnow)
+
+    def serialize(self):
+        return {
+            'id': str(self.id),
+            'user_id': str(self.user_id),
+            'name': self.name,
+            'credential_data': self.credential_data,
+            'registration_response': self.registration_response,
+            'created_at': self.created_at.strftime(DATETIME_FORMAT),
+            'updated_at': get_dt_string_or_none(self.updated_at),
+        }

app/webauthn/rest.py

@@ -0,0 +1,66 @@
+from flask import Blueprint, jsonify, request
+
+from app.dao.users_dao import get_user_by_id
+from app.dao.webauthn_credential_dao import (
+    dao_create_webauthn_credential,
+    dao_delete_webauthn_credential,
+    dao_get_webauthn_credential_by_id,
+    dao_update_webauthn_credential_name,
+)
+from app.errors import InvalidRequest, register_errors
+from app.schema_validation import validate
+from app.webauthn.webauthn_schema import (
+    post_create_webauthn_credential_schema,
+    post_update_webauthn_credential_schema,
+)
+
+webauthn_blueprint = Blueprint('webauthn', __name__, url_prefix='/user/<uuid:user_id>/webauthn')
+register_errors(webauthn_blueprint)
+
+
+@webauthn_blueprint.route('', methods=['GET'])
+def get_webauthn_credentials(user_id):
+    user = get_user_by_id(user_id)
+    return jsonify(data=[cred.serialize() for cred in user.webauthn_credentials]), 200
+
+
+@webauthn_blueprint.route('', methods=['POST'])
+def create_webauthn_credential(user_id):
+    data = request.get_json()
+    validate(data, post_create_webauthn_credential_schema)
+    webauthn_credential = dao_create_webauthn_credential(
+        user_id=user_id,
+        name=data['name'],
+        credential_data=data['credential_data'],
+        registration_response=data['registration_response']
+    )
+
+    return jsonify(data=webauthn_credential.serialize()), 201
+
+
+@webauthn_blueprint.route('/<uuid:webauthn_credential_id>', methods=['POST'])
+def update_webauthn_credential(user_id, webauthn_credential_id):
+    data = request.get_json()
+    validate(data, post_update_webauthn_credential_schema)
+
+    webauthn_credential = dao_get_webauthn_credential_by_id(webauthn_credential_id)
+
+    dao_update_webauthn_credential_name(webauthn_credential, data['name'])
+
+    return jsonify(data=webauthn_credential.serialize()), 200
+
+
+@webauthn_blueprint.route('/<uuid:webauthn_credential_id>', methods=['DELETE'])
+def delete_webauthn_credential(user_id, webauthn_credential_id):
+    webauthn_credential = dao_get_webauthn_credential_by_id(webauthn_credential_id)
+    user = get_user_by_id(user_id)
+
+    if webauthn_credential.user_id != user.id:
+        raise InvalidRequest('Webauthn credential does not belong to this user', status_code=400)
+
+    if len(user.webauthn_credentials) == 1:
+        raise InvalidRequest('Cannot delete last remaining webauthn credential for user', status_code=400)
+
+    dao_delete_webauthn_credential(webauthn_credential)
+
+    return '', 204

app/webauthn/webauthn_schema.py

@@ -0,0 +1,23 @@
+post_create_webauthn_credential_schema = {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "description": "POST webauthn_credential schema",
+    "type": "object",
+    "properties": {
+        "name": {"type": "string"},
+        "credential_data": {"type": "string"},
+        "registration_response": {"type": "string"},
+    },
+    "required": ["name", "credential_data", "registration_response"],
+    "additionalProperties": False
+}
+
+post_update_webauthn_credential_schema = {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "description": "POST update webauthn_credential schema",
+    "type": "object",
+    "properties": {
+        "name": {"type": "string"},
+    },
+    "required": ["name"],
+    "additionalProperties": False
+}

tests/app/db.py

@@ -74,6 +74,7 @@ from app.models import (
     Template,
     TemplateFolder,
     User,
+    WebauthnCredential,
 )
 
 
@@ -1202,3 +1203,22 @@ def create_broadcast_provider_message(
         db.session.add(provider_message_number)
         db.session.commit()
     return provider_message
+
+
+def create_webauthn_credential(
+    user,
+    name='my key',
+    *,
+    credential_data='ABC123',
+    registration_response='DEF456',
+):
+    webauthn_credential = WebauthnCredential(
+        user=user,
+        name=name,
+        credential_data=credential_data,
+        registration_response=registration_response
+    )
+
+    db.session.add(webauthn_credential)
+    db.session.commit()
+    return webauthn_credential

tests/app/webauthn/test_rest.py

@@ -0,0 +1,173 @@
+import uuid
+from unittest.mock import ANY
+
+from tests.app.db import create_user, create_webauthn_credential
+
+
+def test_get_webauthn_credentials_returns_all_credentials_for_user(admin_request, notify_db_session):
+    me = create_user(email='a')
+    other = create_user(email='b')
+    first = create_webauthn_credential(me, '1')
+    create_webauthn_credential(me, '2')
+    create_webauthn_credential(other, '3')
+
+    response = admin_request.get(
+        'webauthn.get_webauthn_credentials',
+        user_id=me.id,
+    )
+
+    creds = sorted(response['data'], key=lambda x: x['name'])
+    assert len(creds) == 2
+
+    assert creds[0] == {
+        'id': str(first.id),
+        'user_id': str(me.id),
+        'name': '1',
+        'credential_data': 'ABC123',
+        'registration_response': 'DEF456',
+        'created_at': ANY,
+        'updated_at': None
+    }
+
+    assert creds[1]['name'] == '2'
+
+
+def test_get_webauthn_credentials_returns_empty_list_if_no_creds(admin_request, sample_user):
+    response = admin_request.get('webauthn.get_webauthn_credentials', user_id=sample_user.id)
+    assert response == {'data': []}
+
+
+def test_get_webauthn_credentials_errors_if_user_doesnt_exist(admin_request, sample_user):
+    create_webauthn_credential(sample_user, '1')
+
+    admin_request.get(
+        'webauthn.get_webauthn_credentials',
+        user_id=uuid.uuid4(),
+        _expected_status=404
+    )
+
+
+def test_create_webauthn_credential_returns_201(admin_request, sample_user):
+    response = admin_request.post(
+        'webauthn.create_webauthn_credential',
+        user_id=sample_user.id,
+        _data={
+            'name': 'my key',
+            'credential_data': 'ABC123',
+            'registration_response': 'DEF456',
+        },
+        _expected_status=201
+    )
+    assert len(sample_user.webauthn_credentials) == 1
+
+    new_cred = sample_user.webauthn_credentials[0]
+
+    assert new_cred.name == 'my key'
+    assert new_cred.credential_data == 'ABC123'
+    assert new_cred.registration_response == 'DEF456'
+    assert response['data']['id'] == str(new_cred.id)
+
+
+def test_create_webauthn_credential_errors_if_schema_violation(admin_request, sample_user):
+    response = admin_request.post(
+        'webauthn.create_webauthn_credential',
+        user_id=sample_user.id,
+        _data={
+            'name': 'my key',
+            'credential_data': 'ABC123',
+            # missing registration_response
+        },
+        _expected_status=400
+    )
+    assert response['errors'][0] == {
+        'error': 'ValidationError',
+        'message': 'registration_response is a required property'
+    }
+
+
+def test_update_webauthn_credential_returns_200(admin_request, sample_user):
+    cred = create_webauthn_credential(sample_user)
+    assert cred.name != 'new name'
+
+    response = admin_request.post(
+        'webauthn.update_webauthn_credential',
+        user_id=sample_user.id,
+        webauthn_credential_id=cred.id,
+        _data={
+            'name': 'new name',
+        },
+    )
+    assert response['data']['id'] == str(cred.id)
+    assert response['data']['name'] == 'new name'
+
+
+def test_update_webauthn_credential_errors_if_schema_violation(admin_request, sample_user):
+    cred = create_webauthn_credential(sample_user)
+    response = admin_request.post(
+        'webauthn.update_webauthn_credential',
+        user_id=sample_user.id,
+        webauthn_credential_id=cred.id,
+        _data={
+            'name': 'my key',
+            # you can't update credential_data
+            'credential_data': 'NAUGHTY123'
+        },
+        _expected_status=400
+    )
+    assert response['errors'][0] == {
+        'error': 'ValidationError',
+        'message': 'Additional properties are not allowed (credential_data was unexpected)'
+    }
+
+
+def test_update_webauthn_credential_errors_if_webauthn_credential_doesnt_exist(admin_request, sample_user):
+    admin_request.post(
+        'webauthn.update_webauthn_credential',
+        user_id=sample_user.id,
+        webauthn_credential_id=uuid.uuid4(),
+        _data={
+            'name': 'my key',
+        },
+        _expected_status=404
+    )
+
+
+def test_delete_webauthn_credential_returns_204(admin_request, sample_user):
+    cred1 = create_webauthn_credential(sample_user)
+    cred2 = create_webauthn_credential(sample_user)
+    admin_request.delete(
+        'webauthn.update_webauthn_credential',
+        user_id=sample_user.id,
+        webauthn_credential_id=cred1.id,
+        _expected_status=204
+    )
+    assert sample_user.webauthn_credentials == [cred2]
+
+
+def test_delete_webauthn_credential_errors_if_last_key(admin_request, sample_user):
+    cred = create_webauthn_credential(sample_user)
+    response = admin_request.delete(
+        'webauthn.delete_webauthn_credential',
+        user_id=sample_user.id,
+        webauthn_credential_id=cred.id,
+        _expected_status=400
+    )
+    assert response['message'] == 'Cannot delete last remaining webauthn credential for user'
+
+
+def test_delete_webauthn_credential_errors_if_user_id_doesnt_match(admin_request, notify_db_session):
+    user_1 = create_user(email='1')
+    user_2 = create_user(email='2')
+    cred_1a = create_webauthn_credential(user_1)  # noqa
+    cred_1b = create_webauthn_credential(user_1)  # noqa
+    cred_2a = create_webauthn_credential(user_2)
+    cred_2b = create_webauthn_credential(user_2)  # noqa
+
+    response = admin_request.delete(
+        'webauthn.delete_webauthn_credential',
+        user_id=user_1.id,
+        webauthn_credential_id=cred_2a.id,
+        _expected_status=400
+    )
+
+    assert response['message'] == 'Webauthn credential does not belong to this user'