Merge pull request #28 from ksauzz/feature/none-per-process-ccache

Add additional ticket updaters

README.md

@@ -17,6 +17,21 @@ ticket.updater_start()
 
 If `keytab path` is not specifyed, kinit uses `KRB5_KTNAME` env, or `/etc/krb5.keytab` to find a keytab file. see: kerberos(1) and kinit(1).
 
+### Ticket Updater Strategies
+
+To avoid a credential cache (ccache) corruption by concurrent updates from multiple processes, KrbTicketUpdater has a few update strategies:
+
+- SimpleKrbTicketUpdater: for single updater process, or multiple updaters w/ per process ccache. (default)
+- MultiProcessKrbTicketUpdater: for multiple updater processes w/ exclusive file lock
+- SingleProcessKrbTicketUpdater: for multiple updater processes w/ exclusive file lock to restrict the number of updater processes to one against the ccache
+
+```
+from krbticket import KrbTicket, SingleProcessKrbTicketUpdater
+
+ticket = KrbTicket.init("<principal>", "<keytab path>", updater_class=SingleProcessKrbTicketUpdater)
+ticket.updater_start()
+```
+
 ### Retry
 
 krbticket supports retry feature utilizing [retrying](https://github.com/rholder/retrying) which provides various retry strategy. To change the behavior, pass the options using `retry_options` of KrbConfig. The dafault values are:

krbticket/__init__.py

@@ -1,2 +1,3 @@
 from krbticket.ticket import *
+from krbticket.updater import *
 from krbticket.config import *

krbticket/command.py

@@ -5,6 +5,7 @@
 
 logger = logging.getLogger(__name__)
 
+
 class KrbCommand():
     @staticmethod
     def kinit(config):

krbticket/config.py

@@ -7,12 +7,15 @@
 
 
 class KrbConfig():
+    from krbticket.updater import SimpleKrbTicketUpdater
+
     def __init__(self, principal=None, keytab=None, kinit_bin="kinit",
                  klist_bin="klist", kdestroy_bin="kdestroy",
                  renewal_threshold=timedelta(minutes=30),
                  ticket_lifetime=None,
                  ticket_renewable_lifetime=None,
                  ccache_name=None,
+                 updater_class=SimpleKrbTicketUpdater,
                  retry_options={
                      'wait_exponential_multiplier': 1000,
                      'wait_exponential_max': 30000,
@@ -25,26 +28,47 @@ def __init__(self, principal=None, keytab=None, kinit_bin="kinit",
         self.renewal_threshold = renewal_threshold
         self.ticket_lifetime = ticket_lifetime
         self.ticket_renewable_lifetime = ticket_renewable_lifetime
+        self.updater_class = updater_class
         self.retry_options = retry_options
         self.ccache_name = ccache_name if ccache_name else self._ccache_name()
 
-
     def __str__(self):
         super_str = super(KrbConfig, self).__str__()
         return "{}: principal={}, keytab={}, kinit_bin={}," \
                " klist_bin={}, kdestroy_bin={}, " \
                " renewal_threshold={}, ticket_lifetime={}, " \
                " ticket_renewable_lifetime={}, " \
                " retry_options={}, ccache_name={}, " \
-               .format(super_str, self.principal, self.keytab, self.kinit_bin, self.klist_bin, self.kdestroy_bin, self.renewal_threshold, self.ticket_lifetime, self.ticket_renewable_lifetime, self.retry_options, self.ccache_name)
-
+               " updater_class={}" \
+               .format(super_str, self.principal, self.keytab, self.kinit_bin,
+                       self.klist_bin, self.kdestroy_bin,
+                       self.renewal_threshold, self.ticket_lifetime,
+                       self.ticket_renewable_lifetime,
+                       self.retry_options, self.ccache_name,
+                       self.updater_class)
 
     def _ccache_name(self):
-        if multiprocessing.current_process().name == 'MainProcess':
-            return os.environ.get('KRB5CCNAME', '/tmp/krb5cc_{}'.format(os.getuid()));
+        if self.updater_class.use_per_process_ccache():
+            return self._per_process_ccache_name()
+        else:
+            return self._default_ccache_name()
+
+    def _is_main_process(self):
+        return multiprocessing.current_process().name == 'MainProcess'
+
+    def _default_ccache_name(self):
+        return os.environ.get('KRB5CCNAME', '/tmp/krb5cc_{}'.format(os.getuid()))
+
+    @property
+    def ccache_lockfile(self):
+        return '{}.krbticket.lock'.format(self.ccache_name)
+
+    def _per_process_ccache_name(self):
+        if self._is_main_process():
+            return self._default_ccache_name()
 
-        # For multiprocess application. e.g. gunicorn
         new_ccname = "/tmp/krb5cc_{}_{}".format(os.getuid(), os.getpid())
+        # Update KRB5CCNAME for kinit
         os.environ['KRB5CCNAME'] = new_ccname
         logger.info("env KRB5CCNAME is updated to '{}' for multiprocessing".format(new_ccname))
 

krbticket/ticket.py

@@ -1,44 +1,18 @@
 import os
 from datetime import datetime
 import logging
-import threading
-import time
 
 from krbticket.command import KrbCommand
 from krbticket.config import KrbConfig
+from krbticket.updater import KrbTicketUpdater
 
 logger = logging.getLogger(__name__)
 
+
 class NoCredentialFound(Exception):
     pass
 
 
-class KrbTicketUpdater(threading.Thread):
-    DEFAULT_INTERVAL = 60 * 10
-
-    def __init__(self, ticket, interval=DEFAULT_INTERVAL):
-        super(KrbTicketUpdater, self).__init__()
-
-        self.ticket = ticket
-        self.interval = interval
-        self.stop_event = threading.Event()
-        self.daemon = True
-
-    def run(self):
-        logger.info("Ticket updater start...")
-        while True:
-            if self.stop_event.is_set():
-                return
-
-            logger.debug("Trying to update ticket...")
-            self.ticket.maybe_update()
-            time.sleep(self.interval)
-
-    def stop(self):
-        logger.debug("Stopping ticket updater...")
-        self.stop_event.set()
-
-
 class KrbTicket():
     def __init__(self, config=None, file=None, principal=None, starting=None, expires=None,
                  service_principal=None, renew_expires=None):
@@ -55,7 +29,7 @@ def updater_start(self, interval=KrbTicketUpdater.DEFAULT_INTERVAL):
         self.updater(interval=interval).start()
 
     def updater(self, interval=KrbTicketUpdater.DEFAULT_INTERVAL):
-        return KrbTicketUpdater(self, interval=interval)
+        return self.config.updater_class(self, interval=interval)
 
     def maybe_update(self):
         self.reload()
@@ -99,15 +73,13 @@ def need_reinit(self):
         else:
             return self.need_renewal()
 
-
     def __str__(self):
         super_str = super(KrbTicket, self).__str__()
         return "{}: file={}, principal={}, starting={}, expires={}," \
                " service_principal={}, renew_expires={}" \
                .format(super_str, self.file, self.principal, self.starting,
                        self.expires, self.service_principal, self.renew_expires)
 
-
     @staticmethod
     def cache_exists(config):
         return os.path.isfile(config.ccache_name)

krbticket/updater.py

@@ -0,0 +1,99 @@
+import logging
+import threading
+import time
+
+import fasteners
+
+logger = logging.getLogger(__name__)
+
+
+class KrbTicketUpdater(threading.Thread):
+    DEFAULT_INTERVAL = 60 * 10
+
+    def __init__(self, ticket, interval=DEFAULT_INTERVAL):
+        super(KrbTicketUpdater, self).__init__()
+
+        self.ticket = ticket
+        self.interval = interval
+        self.stop_event = threading.Event()
+        self.daemon = True
+
+    def run(self):
+        logger.info("{} start...".format(self.__class__.__name__))
+        while True:
+            if self.stop_event.is_set():
+                return
+
+            logger.debug("Trying to update ticket...")
+            self.update()
+            time.sleep(self.interval)
+
+    def update(self):
+        raise NotImplementedError
+
+    @staticmethod
+    def use_per_process_ccache():
+        raise NotImplementedError
+
+    def stop(self):
+        logger.debug("Stopping ticket updater...")
+        self.stop_event.set()
+
+
+class SimpleKrbTicketUpdater(KrbTicketUpdater):
+    """
+    KrbTicketUpdater w/o exclusion control
+
+    Using this with multiprocessing, child processes uses dedicated ccache file
+    """
+    def update(self):
+        self.ticket.maybe_update()
+
+    @staticmethod
+    def use_per_process_ccache():
+        return True
+
+
+class MultiProcessKrbTicketUpdater(KrbTicketUpdater):
+    """
+    Multiprocess KrbTicket Updater
+
+    KrbTicketUpdater w/ exclusive lock for a ccache
+    """
+    def update(self):
+        with fasteners.InterProcessLock(self.ticket.config.ccache_lockfile):
+            self.ticket.maybe_update()
+
+    @staticmethod
+    def use_per_process_ccache():
+        return False
+
+
+class SingleProcessKrbTicketUpdater(KrbTicketUpdater):
+    """
+    Singleprocess KrbTicket Updater
+
+    Single Process KrbTicketUpdater on the system.
+    Multiple updaters can start, but they immediately stops if a updater is already running on the system.
+    """
+    def run(self):
+        lock = fasteners.InterProcessLock(self.ticket.config.ccache_lockfile)
+        got_lock = lock.acquire(blocking=False)
+        if not got_lock:
+            logger.debug("Another updater is detected. Stopping ticket updater...")
+            return
+
+        logger.debug("Got lock: {}...".format(self.ticket.config.ccache_lockfile))
+        try:
+            super().run()
+        finally:
+            if got_lock:
+                lock.release()
+                logger.debug("Released lock: {}...".format(self.ticket.config.ccache_lockfile))
+
+    def update(self):
+        self.ticket.maybe_update()
+
+    @staticmethod
+    def use_per_process_ccache():
+        return False

pytest.ini

@@ -2,6 +2,6 @@
 addopts = -x -v -s --cov=krbticket --cov-report=html
 log_cli = 1
 log_cli_level = DEBUG
-log_cli_format = %(asctime)s [%(levelname)8s] %(name)s %(message)s (%(filename)s:%(lineno)s)
+log_cli_format = %(asctime)s [%(levelname)8s] %(processName)s - %(name)s %(message)s (%(filename)s:%(lineno)s)
 log_cli_date_format=%Y-%m-%d %H:%M:%S
 

requirements.txt

@@ -1 +1,2 @@
 retrying==1.3.3
+fasteners==0.15

tests/helper.py

@@ -6,8 +6,10 @@
 DEFAULT_PRINCIPAL = '[email protected]'
 DEFAULT_KEYTAB = './tests/conf/krb5.keytab'
 DEFAULT_TICKET_RENEWAL_THRESHOLD_SEC = 1
-DEFAULT_TICKET_LIFETIME = '2s'
-DEFAULT_TICKET_RENEWABLE_LIFETIME = '4s'
+DEFAULT_TICKET_LIFETIME_SEC = 4
+DEFAULT_TICKET_LIFETIME = '{}s'.format(DEFAULT_TICKET_LIFETIME_SEC)
+DEFAULT_TICKET_RENEWABLE_LIFETIME_SEC = 8
+DEFAULT_TICKET_RENEWABLE_LIFETIME = '{}s'.format(DEFAULT_TICKET_RENEWABLE_LIFETIME_SEC)
 DEFAULT_CCACHE_NAME = '/tmp/krb5cc_{}'.format(os.getuid())
 
 
@@ -33,17 +35,20 @@ def assert_config(c1, c2):
     assert c1.ticket_renewable_lifetime == c2.ticket_renewable_lifetime
     assert c1.ccache_name == c2.ccache_name
 
-def default_config():
-    return KrbConfig(
-        DEFAULT_PRINCIPAL,
-        DEFAULT_KEYTAB,
-        renewal_threshold=timedelta(seconds=DEFAULT_TICKET_RENEWAL_THRESHOLD_SEC),
-        ticket_lifetime=DEFAULT_TICKET_LIFETIME,
-        ticket_renewable_lifetime=DEFAULT_TICKET_RENEWABLE_LIFETIME,
-        retry_options={
+
+def default_config(**kwargs):
+    default_options = {
+        'principal': DEFAULT_PRINCIPAL,
+        'keytab': DEFAULT_KEYTAB,
+        'renewal_threshold': timedelta(seconds=DEFAULT_TICKET_RENEWAL_THRESHOLD_SEC),
+        'ticket_lifetime': DEFAULT_TICKET_LIFETIME,
+        'ticket_renewable_lifetime': DEFAULT_TICKET_RENEWABLE_LIFETIME,
+        'retry_options': {
             'wait_exponential_multiplier': 100,
             'wait_exponential_max': 1000,
-            'stop_max_attempt_number': 3})
+            'stop_max_attempt_number': 3}
+        }
+    return KrbConfig(**{**default_options, **kwargs})
 
 
 @pytest.fixture