From e62e0509631a977b53bb22e56810a26579e9b0d1 Mon Sep 17 00:00:00 2001 From: Leo Hemsted Date: Mon, 10 May 2021 22:09:07 +0100 Subject: [PATCH] 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 | 4 + app/dao/webauthn_credential_dao.py | 39 +++++++ app/models.py | 11 ++ app/webauthn/__init__.py | 0 app/webauthn/rest.py | 66 +++++++++++ app/webauthn/webauthn_schema.py | 23 ++++ tests/app/db.py | 20 ++++ tests/app/webauthn/test_rest.py | 173 +++++++++++++++++++++++++++++ 8 files changed, 336 insertions(+) create mode 100644 app/dao/webauthn_credential_dao.py create mode 100644 app/webauthn/__init__.py create mode 100644 app/webauthn/rest.py create mode 100644 app/webauthn/webauthn_schema.py create mode 100644 tests/app/webauthn/test_rest.py diff --git a/app/__init__.py b/app/__init__.py index 57cd1c2ef..c5a63bcf1 100644 --- a/app/__init__.py +++ b/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) diff --git a/app/dao/webauthn_credential_dao.py b/app/dao/webauthn_credential_dao.py new file mode 100644 index 000000000..36d79315e --- /dev/null +++ b/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) diff --git a/app/models.py b/app/models.py index 9c471b5f2..23928451c 100644 --- a/app/models.py +++ b/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), + } diff --git a/app/webauthn/__init__.py b/app/webauthn/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/app/webauthn/rest.py b/app/webauthn/rest.py new file mode 100644 index 000000000..0342a1523 --- /dev/null +++ b/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//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('/', 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('/', 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 diff --git a/app/webauthn/webauthn_schema.py b/app/webauthn/webauthn_schema.py new file mode 100644 index 000000000..cc2d2f05a --- /dev/null +++ b/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 +} diff --git a/tests/app/db.py b/tests/app/db.py index 2338a62ec..eed02e534 100644 --- a/tests/app/db.py +++ b/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 diff --git a/tests/app/webauthn/test_rest.py b/tests/app/webauthn/test_rest.py new file mode 100644 index 000000000..8256ffb2d --- /dev/null +++ b/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'