Celery refactoring, celery beat job in configuration

lemur/certificates/service.py

@@ -5,13 +5,12 @@
     :license: Apache, see LICENSE for more details.
 .. moduleauthor:: Kevin Glisson <kglisson@netflix.com>
 """
-from cryptography import x509
-from sqlalchemy import func, or_, not_, cast, Integer
-
 import arrow
+from cryptography import x509
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import hashes, serialization
 from flask import current_app
+from sqlalchemy import func, or_, not_, cast, Integer
 
 from lemur import database
 from lemur.authorities.models import Authority
@@ -276,10 +275,6 @@ def create(**kwargs):
         certificate_issued.send(certificate=cert, authority=cert.authority)
         metrics.send('certificate_issued', 'counter', 1, metric_tags=dict(owner=cert.owner, issuer=cert.issuer))
 
-    if isinstance(cert, PendingCertificate) and cert.authority.plugin_name == 'acme-issuer':
-        # Call Celery to create acme-issuer (LetsEncrypt) certificates
-        from lemur.common.celery import fetch_acme_cert
-        fetch_acme_cert.delay(cert.id)
     return cert
 
 

lemur/common/celery.py

@@ -8,7 +8,9 @@ command: celery -A lemur.common.celery worker --loglevel=info -l DEBUG -B
 
 """
 import copy
+import datetime
 import sys
+from datetime import timezone
 
 from celery import Celery
 from flask import current_app
@@ -117,3 +119,17 @@ def fetch_acme_cert(id):
             wrong_issuer=wrong_issuer
         )
     )
+
+
+@celery.task()
+def fetch_all_pending_acme_certs():
+    """Instantiate celery workers to resolve all pending Acme certificates"""
+    pending_certs = pending_certificate_service.get_pending_certs('all')
+
+    # We only care about certs using the acme-issuer plugin
+    for cert in pending_certs:
+        cert_authority = get_authority(cert.authority_id)
+        if cert_authority.plugin_name == 'acme-issuer':
+            if cert.last_updated == cert.date_created or datetime.datetime.now(
+                    timezone.utc) - cert.last_updated > datetime.timedelta(minutes=3):
+                fetch_acme_cert.delay(cert.id)

lemur/migrations/versions/9392b9f9a805_.py

@@ -0,0 +1,24 @@
+"""Add last_updated field to Pending Certs
+Revision ID: 9392b9f9a805
+Revises: 5ae0ecefb01f
+Create Date: 2018-09-17 08:33:37.087488
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '9392b9f9a805'
+down_revision = '5ae0ecefb01f'
+
+from alembic import op
+from sqlalchemy_utils import ArrowType
+import sqlalchemy as sa
+
+
+def upgrade():
+    op.add_column('pending_certs', sa.Column('last_updated', ArrowType, server_default=sa.text('now()'), onupdate=sa.text('now()'),
+                                             nullable=False))
+
+
+def downgrade():
+    op.drop_column('pending_certs', 'last_updated')
+

lemur/pending_certificates/models.py

@@ -5,19 +5,18 @@
 """
 from datetime import datetime as dt
 
-from sqlalchemy.orm import relationship
 from sqlalchemy import Integer, ForeignKey, String, PassiveDefault, func, Column, Text, Boolean
-from sqlalchemy_utils.types.arrow import ArrowType
+from sqlalchemy.orm import relationship
 from sqlalchemy_utils import JSONType
+from sqlalchemy_utils.types.arrow import ArrowType
 
 from lemur.certificates.models import get_or_increase_name
 from lemur.common import defaults, utils
 from lemur.database import db
-from lemur.utils import Vault
-
 from lemur.models import pending_cert_source_associations, \
     pending_cert_destination_associations, pending_cert_notification_associations, \
     pending_cert_replacement_associations, pending_cert_role_associations
+from lemur.utils import Vault
 
 
 class PendingCertificate(db.Model):
@@ -40,6 +39,7 @@ class PendingCertificate(db.Model):
     dns_provider_id = Column(Integer, ForeignKey('dns_providers.id', ondelete="CASCADE"))
 
     status = Column(Text(), nullable=True)
+    last_updated = Column(ArrowType, PassiveDefault(func.now()), onupdate=func.now(), nullable=False)
 
     rotation = Column(Boolean, default=False)
     user_id = Column(Integer, ForeignKey('users.id'))
@@ -47,9 +47,12 @@ class PendingCertificate(db.Model):
     root_authority_id = Column(Integer, ForeignKey('authorities.id', ondelete="CASCADE"))
     rotation_policy_id = Column(Integer, ForeignKey('rotation_policies.id'))
 
-    notifications = relationship('Notification', secondary=pending_cert_notification_associations, backref='pending_cert', passive_deletes=True)
+    notifications = relationship('Notification', secondary=pending_cert_notification_associations,
-    destinations = relationship('Destination', secondary=pending_cert_destination_associations, backref='pending_cert', passive_deletes=True)
+                                 backref='pending_cert', passive_deletes=True)
-    sources = relationship('Source', secondary=pending_cert_source_associations, backref='pending_cert', passive_deletes=True)
+    destinations = relationship('Destination', secondary=pending_cert_destination_associations, backref='pending_cert',
+                                passive_deletes=True)
+    sources = relationship('Source', secondary=pending_cert_source_associations, backref='pending_cert',
+                           passive_deletes=True)
     roles = relationship('Role', secondary=pending_cert_role_associations, backref='pending_cert', passive_deletes=True)
     replaces = relationship('Certificate',
                             secondary=pending_cert_replacement_associations,
@@ -77,7 +80,7 @@ class PendingCertificate(db.Model):
             # TODO: Fix auto-generated name, it should be renamed on creation
             self.name = get_or_increase_name(
                 defaults.certificate_name(kwargs['common_name'], kwargs['authority'].name,
-                    dt.now(), dt.now(), False), self.external_id)
+                                          dt.now(), dt.now(), False), self.external_id)
             self.rename = True
 
         self.cn = defaults.common_name(utils.parse_csr(self.csr))

lemur/pending_certificates/schemas.py

@@ -1,5 +1,14 @@
 from marshmallow import fields, post_load
 
+from lemur.authorities.schemas import AuthorityNestedOutputSchema
+from lemur.certificates.schemas import CertificateNestedOutputSchema
+from lemur.common.schema import LemurInputSchema, LemurOutputSchema
+from lemur.destinations.schemas import DestinationNestedOutputSchema
+from lemur.domains.schemas import DomainNestedOutputSchema
+from lemur.notifications import service as notification_service
+from lemur.notifications.schemas import NotificationNestedOutputSchema
+from lemur.policies.schemas import RotationPolicyNestedOutputSchema
+from lemur.roles.schemas import RoleNestedOutputSchema
 from lemur.schemas import (
     AssociatedCertificateSchema,
     AssociatedDestinationSchema,
@@ -8,18 +17,7 @@ from lemur.schemas import (
     EndpointNestedOutputSchema,
     ExtensionSchema
 )
-
-from lemur.common.schema import LemurInputSchema, LemurOutputSchema
 from lemur.users.schemas import UserNestedOutputSchema
-from lemur.authorities.schemas import AuthorityNestedOutputSchema
-from lemur.certificates.schemas import CertificateNestedOutputSchema
-from lemur.destinations.schemas import DestinationNestedOutputSchema
-from lemur.domains.schemas import DomainNestedOutputSchema
-from lemur.notifications.schemas import NotificationNestedOutputSchema
-from lemur.roles.schemas import RoleNestedOutputSchema
-from lemur.policies.schemas import RotationPolicyNestedOutputSchema
-
-from lemur.notifications import service as notification_service
 
 
 class PendingCertificateSchema(LemurInputSchema):
@@ -38,6 +36,7 @@ class PendingCertificateOutputSchema(LemurOutputSchema):
     name = fields.String()
     number_attempts = fields.Integer()
     date_created = fields.Date()
+    last_updated = fields.Date()
 
     rotation = fields.Boolean()
 
@@ -88,7 +87,8 @@ class PendingCertificateEditInputSchema(PendingCertificateSchema):
         """
         if data['owner']:
             notification_name = "DEFAULT_{0}".format(data['owner'].split('@')[0].upper())
-            data['notifications'] += notification_service.create_default_expiration_notifications(notification_name, [data['owner']])
+            data['notifications'] += notification_service.create_default_expiration_notifications(notification_name,
+                                                                                                  [data['owner']])
         return data
 
 

requirements-dev.txt

@@ -12,7 +12,7 @@ chardet==3.0.4            # via requests
 flake8==3.5.0
 identify==1.1.5           # via pre-commit
 idna==2.7                 # via requests
-invoke==1.1.1
+invoke==1.2.0
 mccabe==0.6.1             # via flake8
 nodeenv==1.3.2
 pkginfo==1.4.2            # via twine

requirements-docs.txt

@@ -87,7 +87,7 @@ sphinx-rtd-theme==0.4.1
 sphinx==1.8.0
 sphinxcontrib-httpdomain==1.7.0
 sphinxcontrib-websupport==1.1.0  # via sphinx
-sqlalchemy-utils==0.33.3
+sqlalchemy-utils==0.33.4
 sqlalchemy==1.2.11
 tabulate==0.8.2
 urllib3==1.23

requirements-tests.txt

@@ -8,9 +8,9 @@ asn1crypto==0.24.0        # via cryptography
 atomicwrites==1.2.1       # via pytest
 attrs==18.2.0             # via pytest
 aws-xray-sdk==0.95        # via moto
-boto3==1.9.3              # via moto
+boto3==1.9.4              # via moto
 boto==2.49.0              # via moto
-botocore==1.12.3          # via boto3, moto, s3transfer
+botocore==1.12.4          # via boto3, moto, s3transfer
 certifi==2018.8.24        # via requests
 cffi==1.11.5              # via cryptography
 chardet==3.0.4            # via requests
@@ -23,7 +23,7 @@ docker==3.5.0             # via moto
 docutils==0.14            # via botocore
 ecdsa==0.13               # via python-jose
 factory-boy==2.11.1
-faker==0.9.0
+faker==0.9.1
 flask==1.0.2              # via pytest-flask
 freezegun==0.3.10
 future==0.16.0            # via python-jose