Adding more source syncing logic

docs/administration/index.rst

@@ -505,11 +505,19 @@ All commands default to `~/.lemur/lemur.conf.py` if a configuration is not speci
 
 .. data:: sync
 
-    Sync attempts to discover certificates in the environment that were not created by Lemur. There
+    Sync attempts to discover certificates in the environment that were not created by Lemur. If you wish to only sync
+    a few sources you can pass a comma delimited list of sources to sync
 
     ::
 
-        lemur sync --all
+        lemur sync source1,source2
+
+
+    Additionally you can also list the available sources that Lemur can sync
+
+    ::
+
+        lemur sync -list
 
 
 Identity and Access Management

docs/developer/plugins/index.rst

@@ -215,9 +215,9 @@ certificate Lemur does not know about and adding the certificate to it's invento
 The `SourcePlugin` object has one default option of `pollRate`. This controls the number of seconds which to get new certificates.
 
  .. warning::
-    Lemur currently has a very basic polling system of running a cron job every 15min to see which source plugins need to be run.
+    Lemur currently has a very basic polling system of running a cron job every 15min to see which source plugins need to be run. A lock file is generated to guarentee that ]
-    This means special consideration needs to be taken such that running all `SourcePlugins` does not take >15min to run. It also means
+    only one sync is running at a time. It also means that the minimum resolution of a source plugin poll rate is effectively 15min. You can always specify a faster cron
-    that the minimum resolution of a source plugin poll rate is effectively 15min.
+    job if you need a higher resolution sync job.
 
 
 The `SourcePlugin` object requires implementation of one function::

lemur/manage.py

@@ -1,12 +1,15 @@
 import os
 import sys
 import base64
+import time
 from gunicorn.config import make_settings
 
 from cryptography.fernet import Fernet
 
+from lockfile import LockFile, LockTimeout
+
 from flask import current_app
-from flask.ext.script import Manager, Command, Option, Group, prompt_pass
+from flask.ext.script import Manager, Command, Option, prompt_pass
 from flask.ext.migrate import Migrate, MigrateCommand, stamp
 from flask_script.commands import ShowUrls, Clean, Server
 
@@ -15,11 +18,12 @@ from lemur.users import service as user_service
 from lemur.roles import service as role_service
 from lemur.destinations import service as destination_service
 from lemur.certificates import service as cert_service
+from lemur.sources import service as source_service
 
 from lemur.plugins.base import plugins
 
 from lemur.certificates.verify import verify_string
-from lemur.sources import sync
+from lemur.sources.service import sync
 
 from lemur import create_app
 
@@ -176,51 +180,55 @@ def generate_settings():
     return output
 
 
-class Sync(Command):
+@manager.option('-s', '--sources', dest='labels', default='', required=False)
+@manager.option('-l', '--list', dest='view', default=False, required=False)
+def sync_sources(labels, view):
     """
     Attempts to run several methods Certificate discovery. This is
     run on a periodic basis and updates the Lemur datastore with the
     information it discovers.
     """
-
+    if view:
-    # TODO create these commands dynamically
+        for source in source_service.get_all():
-    option_list = [
+            sys.stdout.write(
-        Group(
+                "[{active}]\t{label}\t{description}!\n".format(
-            Option('-a', '--all', action="store_true"),
+                    label=source.label,
-            exclusive=True, required=True
+                    description=source.description,
+                    active=source.active
                 )
-    ]
+            )
+    else:
+        start_time = time.time()
+        lock_file = "/tmp/.lemur_lock"
+        sync_lock = LockFile(lock_file)
 
-    def run(self, all, aws, cloudca, source):
+        while not sync_lock.i_am_locking():
-        sys.stdout.write("[!] Starting to sync with external sources!\n")
-
-        if all or aws:
-            sys.stdout.write("[!] Starting to sync with AWS!\n")
             try:
-                sync.aws()
+                sync_lock.acquire(timeout=10)    # wait up to 10 seconds
-                # sync_all_elbs()
-                sys.stdout.write("[+] Finished syncing with AWS!\n")
-            except Exception as e:
-                sys.stdout.write("[-] Syncing with AWS failed!\n")
 
-        if all or cloudca:
+                if labels:
-            sys.stdout.write("[!] Starting to sync with CloudCA!\n")
+                    sys.stdout.write("[+] Staring to sync sources: {labels}!\n".format(labels))
-            try:
+                    labels = labels.split(",")
-                sync.cloudca()
+                else:
-                sys.stdout.write("[+] Finished syncing with CloudCA!\n")
+                    sys.stdout.write("[+] Starting to sync ALL sources!\n".format(labels))
-            except Exception as e:
-                sys.stdout.write("[-] Syncing with CloudCA failed!\n")
 
-            sys.stdout.write("[!] Starting to sync with Source Code!\n")
+                sync(labels=labels)
+                sys.stdout.write(
+                    "[+] Finished syncing sources. Run Time: {time}\n".format(
+                        time=(time.time() - start_time)
+                    )
+                )
+            except LockTimeout:
+                sys.stderr.write(
+                    "[!] Unable to acquire file lock on {file}, is there another sync running?\n".format(
+                        file=lock_file
+                    )
+                )
+                sync_lock.break_lock()
+                sync_lock.acquire()
+                sync_lock.release()
 
-        if all or source:
+        sync_lock.release()
-            try:
-                sync.source()
-                sys.stdout.write("[+] Finished syncing with Source Code!\n")
-            except Exception as e:
-                sys.stdout.write("[-] Syncing with Source Code failed!\n")
-
-            sys.stdout.write("[+] Finished syncing with external sources!\n")
 
 
 class InitializeApp(Command):
@@ -473,7 +481,6 @@ def main():
     manager.add_command("init", InitializeApp())
     manager.add_command("create_user", CreateUser())
     manager.add_command("create_role", CreateRole())
-    manager.add_command("sync", Sync())
     manager.run()
 
 

lemur/notifications/service.py

@@ -64,7 +64,7 @@ def send_expiration_notifications():
     """
     notifications = 0
 
-    for plugin_name, notifications in database.get_all(Notification, 'active', field='status').group_by(Notification.plugin_name):
+    for plugin_name, notifications in database.get_all(Notification, True, field='active').group_by(Notification.plugin_name):
         notifications += 1
 
         messages = _deduplicate(notifications)

lemur/sources/models.py

@@ -6,7 +6,7 @@
 .. moduleauthor:: Kevin Glisson <kglisson@netflix.com>
 """
 import copy
-from sqlalchemy import Column, Integer, String, Text
+from sqlalchemy import Column, Integer, String, Text, DateTime, Boolean
 from sqlalchemy_utils import JSONType
 from lemur.database import db
 
@@ -20,6 +20,8 @@ class Source(db.Model):
     options = Column(JSONType)
     description = Column(Text())
     plugin_name = Column(String(32))
+    active = Column(Boolean, default=True)
+    last_run = Column(DateTime)
 
     @property
     def plugin(self):

lemur/sources/service.py

@@ -5,9 +5,80 @@
     :license: Apache, see LICENSE for more details.
 .. moduleauthor:: Kevin Glisson <kglisson@netflix.com>
 """
+from flask import current_app
+
 from lemur import database
 from lemur.sources.models import Source
 from lemur.certificates.models import Certificate
+from lemur.certificates import service as cert_service
+
+from lemur.plugins.base import plugins
+
+
+def _disassociate_certs_from_source(current_certificates, found_certificates, source_label):
+    missing = []
+    for cc in current_certificates:
+        for fc in found_certificates:
+            if fc.body == cc.body:
+                break
+        else:
+            missing.append(cc)
+
+    for c in missing:
+        for s in c.sources:
+            if s.label == source_label:
+                current_app.logger.info(
+                    "Certificate {name} is no longer associated with {source}".format(
+                        name=c.name,
+                        source=source_label
+                    )
+                )
+                c.sources.delete(s)
+
+
+def sync(labels=None):
+    new, updated = 0, 0
+    c_certificates = cert_service.get_all_certs()
+
+    for source in database.get_all(Source, True, field='active'):
+        # we should be able to specify, individual sources to sync
+        if labels:
+            if source.label not in labels:
+                continue
+
+        current_app.logger.error("Retrieving certificates from {0}".format(source.title))
+        s = plugins.get(source.plugin_name)
+        certificates = s.get_certificates(source.options)
+
+        for certificate in certificates:
+            exists = cert_service.find_duplicates(certificate)
+
+            if not exists:
+                cert = cert_service.import_certificate(**certificate)
+                cert.sources.append(source)
+                database.update(cert)
+
+                new += 1
+
+            # check to make sure that existing certificates have the current source associated with it
+            if len(exists) == 1:
+                for s in cert.sources:
+                    if s.label == source.label:
+                        break
+                    else:
+                        cert.sources.append(source)
+
+                updated += 1
+
+            else:
+                current_app.logger.warning(
+                    "Multiple certificates found, attempt to deduplicate the following certificates: {0}".format(
+                        ",".join([x.name for x in exists])
+                    )
+                )
+
+        # we need to try and find the absent of certificates so we can properly disassociate them when they are deleted
+        _disassociate_certs_from_source(c_certificates, certificates, source)
 
 
 def create(label, plugin_name, options, description=None):

lemur/sources/sync.py

@@ -1,46 +0,0 @@
-"""
-.. module: lemur.sources.sync
-    :platform: Unix
-    :synopsis: This module contains various certificate syncing operations.
-    Because of the nature of the SSL environment there are multiple ways
-    a certificate could be created without Lemur's knowledge. Lemur attempts
-    to 'sync' with as many different datasources as possible to try and track
-    any certificate that may be in use.
-
-    These operations are typically run on a periodic basis from either the command
-    line or a cron job.
-
-    :copyright: (c) 2015 by Netflix Inc., see AUTHORS for more
-    :license: Apache, see LICENSE for more details.
-.. moduleauthor:: Kevin Glisson <kglisson@netflix.com>
-"""
-from flask import current_app
-
-from lemur.certificates import service as cert_service
-
-from lemur.plugins.base import plugins
-from lemur.plugins.bases.source import SourcePlugin
-
-
-def sync():
-    for plugin in plugins:
-        new = 0
-        updated = 0
-        if isinstance(plugin, SourcePlugin):
-            if plugin.is_enabled():
-                current_app.logger.error("Retrieving certificates from {0}".format(plugin.title))
-                certificates = plugin.get_certificates()
-
-                for certificate in certificates:
-                    exists = cert_service.find_duplicates(certificate)
-
-                    if not exists:
-                        cert_service.import_certificate(**certificate)
-                        new += 1
-
-                    if len(exists) == 1:
-                        updated += 1
-
-                # TODO associated cert with source
-                # TODO update cert if found from different source
-                # TODO disassociate source if missing

lemur/sources/views.py

@@ -23,6 +23,7 @@ FIELDS = {
     'description': fields.String,
     'sourceOptions': fields.Raw(attribute='options'),
     'pluginName': fields.String(attribute='plugin_name'),
+    'lastRun': fields.DateTime(attribute='last_run', dt_format='iso8061'),
     'label': fields.String,
     'id': fields.Integer,
 }
@@ -71,6 +72,7 @@ class SourcesList(AuthenticatedResource):
                             }
                         ],
                         "pluginName": "aws-source",
+                        "lastRun": "2015-08-01T15:40:58",
                         "id": 3,
                         "description": "test",
                         "label": "test"
@@ -120,6 +122,7 @@ class SourcesList(AuthenticatedResource):
                 ],
                 "pluginName": "aws-source",
                 "id": 3,
+                "lastRun": "2015-08-01T15:40:58",
                 "description": "test",
                 "label": "test"
               }
@@ -145,6 +148,7 @@ class SourcesList(AuthenticatedResource):
                 ],
                 "pluginName": "aws-source",
                 "id": 3,
+                "lastRun": "2015-08-01T15:40:58",
                 "description": "test",
                 "label": "test"
               }
@@ -203,6 +207,7 @@ class Sources(AuthenticatedResource):
                 ],
                 "pluginName": "aws-source",
                 "id": 3,
+                "lastRun": "2015-08-01T15:40:58",
                 "description": "test",
                 "label": "test"
               }
@@ -241,6 +246,7 @@ class Sources(AuthenticatedResource):
                 ],
                 "pluginName": "aws-source",
                 "id": 3,
+                "lastRun": "2015-08-01T15:40:58",
                 "description": "test",
                 "label": "test"
               }
@@ -266,6 +272,7 @@ class Sources(AuthenticatedResource):
                 ],
                 "pluginName": "aws-source",
                 "id": 3,
+                "lastRun": "2015-08-01T15:40:58",
                 "description": "test",
                 "label": "test"
               }
@@ -332,6 +339,7 @@ class CertificateSources(AuthenticatedResource):
                         ],
                         "pluginName": "aws-source",
                         "id": 3,
+                        "lastRun": "2015-08-01T15:40:58",
                         "description": "test",
                         "label": "test"
                     }

setup.py

@@ -42,7 +42,8 @@ install_requires = [
     'cryptography>=1.0dev',
     'pyopenssl==0.15.1',
     'pyjwt==1.0.1',
-    'xmltodict==0.9.2'
+    'xmltodict==0.9.2',
+    'lockfile=0.10.2'
 ]
 
 tests_require = [