Set euid/egid ourselves. Allow using username/groupnames for the --uid/--gid arguments, also if no --gid is specified, it uses the specified users primary gid.

celery/bin/celerybeat.py

@@ -56,6 +56,7 @@ from celery import platform
 from celery import __version__
 from celery.log import emergency_error
 from celery.beat import ClockService
+from celery.utils import noop
 from celery.loaders import current_loader, settings
 from celery.messaging import get_connection_info
 
@@ -140,16 +141,16 @@ def run_clockservice(detach=False, loglevel=conf.CELERYBEAT_LOG_LEVEL,
     platform.set_process_title("celerybeat",
                                info=" ".join(sys.argv[arg_start:]))
     from celery.log import setup_logger, redirect_stdouts_to_logger
+    on_stop = noop
     if detach:
-        context = platform.create_daemon_context(logfile, pidfile,
+        context, on_stop = platform.create_daemon_context(logfile, pidfile,
                                         chroot_directory=chroot,
                                         working_directory=working_directory,
-                                        umask=umask,
-                                        uid=uid,
-                                        gid=gid)
+                                        umask=umask)
         context.open()
         logger = setup_logger(loglevel, logfile)
         redirect_stdouts_to_logger(logger, loglevel)
+        platform.set_effective_user(uid, gid)
 
     # Run the worker init handler.
     # (Usually imports task modules and such.)
@@ -170,8 +171,7 @@ def run_clockservice(detach=False, loglevel=conf.CELERYBEAT_LOG_LEVEL,
     try:
         _run_clock()
     except:
-        if detach:
-            context.close()
+        on_stop()
         raise
 
 

celery/bin/celeryd.py

@@ -72,6 +72,7 @@ from celery import platform
 from celery import __version__
 from celery.log import emergency_error
 from celery.task import discard_all
+from celery.utils import noop
 from celery.worker import WorkController
 from celery.loaders import current_loader, settings
 from celery.loaders import settings
@@ -117,10 +118,10 @@ OPTION_LIST = (
             action="store_true", dest="detach",
             help="Run in the background as a daemon."),
     optparse.make_option('-u', '--uid', default=None,
-            action="store", dest="uid", type="int",
+            action="store", dest="uid",
             help="User-id to run celeryd as when in daemon mode."),
     optparse.make_option('-g', '--gid', default=None,
-            action="store", dest="gid", type="int",
+            action="store", dest="gid",
             help="Group-id to run celeryd as when in daemon mode."),
     optparse.make_option('--umask', default=0,
             action="store", type="int", dest="umask",
@@ -186,18 +187,17 @@ def run_worker(concurrency=conf.DAEMON_CONCURRENCY, detach=False,
 
     print("Celery has started.")
     set_process_status("Running...")
+    on_stop = noop
     if detach:
         from celery.log import setup_logger, redirect_stdouts_to_logger
-        context = platform.create_daemon_context(logfile, pidfile,
+        context, on_stop = platform.create_daemon_context(logfile, pidfile,
                                         chroot_directory=chroot,
                                         working_directory=working_directory,
-                                        umask=umask,
-                                        uid=uid,
-                                        gid=gid)
+                                        umask=umask)
         context.open()
         logger = setup_logger(loglevel, logfile)
         redirect_stdouts_to_logger(logger, loglevel)
-
+        platform.set_effective_user(uid, gid)
 
     def run_worker():
         worker = WorkController(concurrency=concurrency,
@@ -226,8 +226,7 @@ def run_worker(concurrency=conf.DAEMON_CONCURRENCY, detach=False,
         run_worker()
     except:
         set_process_status("Exiting...")
-        if detach:
-            context.close()
+        on_stop()
         raise
 
 

celery/platform.py

@@ -1,5 +1,7 @@
 import os
 import sys
+import pwd
+import grp
 import signal
 try:
     from setproctitle import setproctitle as _setproctitle
@@ -13,6 +15,8 @@ try:
 except ImportError:
     CAN_DETACH = False
 
+from celery.utils import noop
+
 
 def acquire_pidlock(pidfile):
     """Get the :class:`daemon.pidlockfile.PIDLockFile` handler for
@@ -55,7 +59,8 @@ def create_daemon_context(logfile=None, pidfile=None, **options):
         raise RuntimeError(
                 "This operating system doesn't support detach.")
 
-    from daemon import DaemonContext
+    import daemon
+    daemon.change_process_owner = noop # We handle our own user change.
 
     # set SIGCLD back to the default SIG_DFL (before python-daemon overrode
     # it) lets the parent wait() for the terminated child process and stops
@@ -69,9 +74,10 @@ def create_daemon_context(logfile=None, pidfile=None, **options):
 
     options["pidfile"] = pidfile and acquire_pidlock(pidfile)
 
-    defaults = {"uid": lambda: os.geteuid(),
-                "gid": lambda: os.getegid(),
-                "umask": lambda: 0,
+    #options["uid"] = os.getuid()
+    #options["gid"] = os.getgid()
+
+    defaults = {"umask": lambda: 0,
                 "chroot_directory": lambda: None,
                 "working_directory": lambda: os.getcwd()}
 
@@ -79,7 +85,9 @@ def create_daemon_context(logfile=None, pidfile=None, **options):
         if opt_name not in options or options[opt_name] is None:
             options[opt_name] = opt_default_gen()
 
-    return DaemonContext(**options)
+    context = daemon.DaemonContext(**options)
+
+    return context, context.close
 
 
 def reset_signal(signal_name):
@@ -102,8 +110,47 @@ def set_process_title(progname, info=None):
         proctitle = info and "%s %s" % (proctitle, info) or proctitle
         _setproctitle(proctitle)
 
+
 def set_mp_process_title(progname, info=None):
     from multiprocessing.process import current_process
     return set_process_title("%s.%s" % (progname, current_process().name),
                              info=info)
 
+
+def parse_uid(uid):
+    try:
+        return int(uid)
+    except ValueError:
+        return pwd.getpwnam(uid).pw_uid
+
+
+def parse_gid(gid):
+    try:
+        return int(gid)
+    except ValueError:
+        return grp.getgrnam(gid).gr_gid
+
+
+def setegid(gid):
+    gid = parse_uid(gid)
+    if gid != os.getgid():
+        os.setegid(gid)
+
+
+def seteuid(uid):
+    uid = parse_uid(uid)
+    if uid != os.getuid():
+        os.seteuid(uid)
+
+
+def set_effective_user(uid=None, gid=None):
+    # gid/uid can be int or username/groupname.
+    uid = uid and parse_uid(uid)
+    gid = gid and parse_gid(gid)
+
+    if uid:
+        # If gid isn't defined, get the primary gid of the user.
+        setegid(gid or pwd.getpwuid(uid).pw_gid)
+        seteuid(uid)
+    else:
+        gid and setegid(gid)