diff --git a/lemur/certificates/views.py b/lemur/certificates/views.py index 54c60924..948c44d6 100644 --- a/lemur/certificates/views.py +++ b/lemur/certificates/views.py @@ -6,6 +6,7 @@ .. moduleauthor:: Kevin Glisson """ import base64 +import arrow from builtins import str from flask import Blueprint, make_response, jsonify, g @@ -660,6 +661,51 @@ class Certificates(AuthenticatedResource): log_service.create(g.current_user, 'update_cert', certificate=cert) return cert + def delete(self, certificate_id, data=None): + """ + .. http:delete:: /certificates/1 + + Delete a certificate + + **Example request**: + + .. sourcecode:: http + + DELETE /certificates/1 HTTP/1.1 + Host: example.com + + **Example response**: + + .. sourcecode:: http + + HTTP/1.1 200 OK + + :reqheader Authorization: OAuth token to authenticate + :statuscode 204: no error + :statuscode 403: unauthenticated + :statusoode 404: certificate not found + + """ + cert = service.get(certificate_id) + + if not cert: + return dict(message="Cannot find specified certificate"), 404 + + # allow creators + if g.current_user != cert.user: + owner_role = role_service.get_by_name(cert.owner) + permission = CertificatePermission(owner_role, [x.name for x in cert.roles]) + + if not permission.can(): + return dict(message='You are not authorized to delete this certificate'), 403 + + if arrow.get(cert.not_after) > arrow.utcnow(): + return dict(message='Certificate is still valid, only expired certificates can be deleted'), 412 + + service.update(certificate_id, deleted=True) + log_service.create(g.current_user, 'delete_cert', certificate=cert) + return '', 204 + class NotificationCertificatesList(AuthenticatedResource): """ Defines the 'certificates' endpoint """ diff --git a/lemur/logs/models.py b/lemur/logs/models.py index d4239e59..9f982c24 100644 --- a/lemur/logs/models.py +++ b/lemur/logs/models.py @@ -18,6 +18,6 @@ class Log(db.Model): __tablename__ = 'logs' id = Column(Integer, primary_key=True) certificate_id = Column(Integer, ForeignKey('certificates.id')) - log_type = Column(Enum('key_view', 'create_cert', 'update_cert', 'revoke_cert', name='log_type'), nullable=False) + log_type = Column(Enum('key_view', 'create_cert', 'update_cert', 'revoke_cert', 'delete_cert', name='log_type'), nullable=False) logged_at = Column(ArrowType(), PassiveDefault(func.now()), nullable=False) user_id = Column(Integer, ForeignKey('users.id'), nullable=False) diff --git a/lemur/migrations/versions/9f79024fe67b_.py b/lemur/migrations/versions/9f79024fe67b_.py new file mode 100644 index 00000000..ad22d5f3 --- /dev/null +++ b/lemur/migrations/versions/9f79024fe67b_.py @@ -0,0 +1,22 @@ +""" Add delete_cert to log_type enum + +Revision ID: 9f79024fe67b +Revises: ee827d1e1974 +Create Date: 2019-01-03 15:36:59.181911 + +""" + +# revision identifiers, used by Alembic. +revision = '9f79024fe67b' +down_revision = 'ee827d1e1974' + +from alembic import op +import sqlalchemy as sa + + +def upgrade(): + op.sync_enum_values('public', 'log_type', ['create_cert', 'key_view', 'revoke_cert', 'update_cert'], ['create_cert', 'delete_cert', 'key_view', 'revoke_cert', 'update_cert']) + + +def downgrade(): + op.sync_enum_values('public', 'log_type', ['create_cert', 'delete_cert', 'key_view', 'revoke_cert', 'update_cert'], ['create_cert', 'key_view', 'revoke_cert', 'update_cert']) diff --git a/lemur/tests/conftest.py b/lemur/tests/conftest.py index 3790358e..32733e51 100644 --- a/lemur/tests/conftest.py +++ b/lemur/tests/conftest.py @@ -14,7 +14,8 @@ from lemur.tests.vectors import SAN_CERT_KEY, INTERMEDIATE_KEY from .factories import ApiKeyFactory, AuthorityFactory, NotificationFactory, DestinationFactory, \ CertificateFactory, UserFactory, RoleFactory, SourceFactory, EndpointFactory, \ - RotationPolicyFactory, PendingCertificateFactory, AsyncAuthorityFactory, CryptoAuthorityFactory + RotationPolicyFactory, PendingCertificateFactory, AsyncAuthorityFactory, InvalidCertificateFactory, \ + CryptoAuthorityFactory def pytest_runtest_setup(item): @@ -167,6 +168,15 @@ def pending_certificate(session): return p +@pytest.fixture +def invalid_certificate(session): + u = UserFactory() + a = AsyncAuthorityFactory() + i = InvalidCertificateFactory(user=u, authority=a) + session.commit() + return i + + @pytest.fixture def admin_user(session): u = UserFactory() diff --git a/lemur/tests/factories.py b/lemur/tests/factories.py index 3717c64d..a4af3d43 100644 --- a/lemur/tests/factories.py +++ b/lemur/tests/factories.py @@ -20,7 +20,7 @@ from lemur.policies.models import RotationPolicy from lemur.api_keys.models import ApiKey from .vectors import SAN_CERT_STR, SAN_CERT_KEY, CSR_STR, INTERMEDIATE_CERT_STR, ROOTCA_CERT_STR, INTERMEDIATE_KEY, \ - WILDCARD_CERT_KEY + WILDCARD_CERT_KEY, INVALID_CERT_STR class BaseFactory(SQLAlchemyModelFactory): @@ -137,6 +137,11 @@ class CACertificateFactory(CertificateFactory): private_key = INTERMEDIATE_KEY +class InvalidCertificateFactory(CertificateFactory): + body = INVALID_CERT_STR + private_key = '' + + class AuthorityFactory(BaseFactory): """Authority factory.""" name = Sequence(lambda n: 'authority{0}'.format(n)) diff --git a/lemur/tests/test_certificates.py b/lemur/tests/test_certificates.py index 83bf49e5..8247c36b 100644 --- a/lemur/tests/test_certificates.py +++ b/lemur/tests/test_certificates.py @@ -736,15 +736,26 @@ def test_certificate_put_with_data(client, certificate, issuer_plugin): @pytest.mark.parametrize("token,status", [ - (VALID_USER_HEADER_TOKEN, 405), - (VALID_ADMIN_HEADER_TOKEN, 405), - (VALID_ADMIN_API_TOKEN, 405), - ('', 405) + (VALID_USER_HEADER_TOKEN, 403), + (VALID_ADMIN_HEADER_TOKEN, 412), + (VALID_ADMIN_API_TOKEN, 412), + ('', 401) ]) def test_certificate_delete(client, token, status): assert client.delete(api.url_for(Certificates, certificate_id=1), headers=token).status_code == status +@pytest.mark.parametrize("token,status", [ + (VALID_USER_HEADER_TOKEN, 403), + (VALID_ADMIN_HEADER_TOKEN, 204), + (VALID_ADMIN_API_TOKEN, 204), + ('', 401) +]) +def test_invalid_certificate_delete(client, invalid_certificate, token, status): + assert client.delete( + api.url_for(Certificates, certificate_id=invalid_certificate.id), headers=token).status_code == status + + @pytest.mark.parametrize("token,status", [ (VALID_USER_HEADER_TOKEN, 405), (VALID_ADMIN_HEADER_TOKEN, 405),