"""Platforms.Utilities dealing with platform specifics: signals, daemonization,users, groups, and so on."""importatexitimporterrnoimportmathimportnumbersimportosimportplatformas_platformimportsignalas_signalimportsysimportwarningsfromcontextlibimportcontextmanagerfrombilliard.compatimportclose_open_fds,get_fdmaxfrombilliard.utilimportset_pdeathsigas_set_pdeathsig# fileno used to be in this modulefromkombu.utils.compatimportmaybe_filenofromkombu.utils.encodingimportsafe_strfrom.exceptionsimportSecurityError,SecurityWarning,reraisefrom.localimporttry_importtry:frombilliard.processimportcurrent_processexceptImportError:current_process=None_setproctitle=try_import('setproctitle')resource=try_import('resource')pwd=try_import('pwd')grp=try_import('grp')mputil=try_import('multiprocessing.util')__all__=('EX_OK','EX_FAILURE','EX_UNAVAILABLE','EX_USAGE','SYSTEM','IS_macOS','IS_WINDOWS','SIGMAP','pyimplementation','LockFailed','get_fdmax','Pidfile','create_pidlock','close_open_fds','DaemonContext','detached','parse_uid','parse_gid','setgroups','initgroups','setgid','setuid','maybe_drop_privileges','signals','signal_name','set_process_title','set_mp_process_title','get_errno_name','ignore_errno','fd_by_path','isatty',)# exitcodesEX_OK=getattr(os,'EX_OK',0)EX_FAILURE=1EX_UNAVAILABLE=getattr(os,'EX_UNAVAILABLE',69)EX_USAGE=getattr(os,'EX_USAGE',64)EX_CANTCREAT=getattr(os,'EX_CANTCREAT',73)SYSTEM=_platform.system()IS_macOS=SYSTEM=='Darwin'IS_WINDOWS=SYSTEM=='Windows'DAEMON_WORKDIR='/'PIDFILE_FLAGS=os.O_CREAT|os.O_EXCL|os.O_WRONLYPIDFILE_MODE=((os.R_OK|os.W_OK)<<6)|((os.R_OK)<<3)|(os.R_OK)PIDLOCKED="""ERROR: Pidfile ({0}) already exists.Seems we're already running? (pid: {1})"""ROOT_DISALLOWED="""\Running a worker with superuser privileges when theworker accepts messages serialized with pickle is a very bad idea!If you really want to continue then you have to set the C_FORCE_ROOTenvironment variable (but please think about this before you do).User information: uid={uid} euid={euid} gid={gid} egid={egid}"""ROOT_DISCOURAGED="""\You're running the worker with superuser privileges: this isabsolutely not recommended!Please specify a different user using the --uid option.User information: uid={uid} euid={euid} gid={gid} egid={egid}"""ASSUMING_ROOT="""\An entry for the specified gid or egid was not found.We're assuming this is a potential security issue."""SIGNAMES={sigforsigindir(_signal)ifsig.startswith('SIG')and'_'notinsig}SIGMAP={getattr(_signal,name):namefornameinSIGNAMES}
[文档]defisatty(fh):"""Return true if the process has a controlling terminal."""try:returnfh.isatty()exceptAttributeError:pass
[文档]defpyimplementation():"""Return string identifying the current Python implementation."""ifhasattr(_platform,'python_implementation'):return_platform.python_implementation()elifsys.platform.startswith('java'):return'Jython '+sys.platformelifhasattr(sys,'pypy_version_info'):v='.'.join(str(p)forpinsys.pypy_version_info[:3])ifsys.pypy_version_info[3:]:v+='-'+''.join(str(p)forpinsys.pypy_version_info[3:])return'PyPy '+velse:return'CPython'
[文档]classLockFailed(Exception):"""Raised if a PID lock can't be acquired."""
[文档]classPidfile:"""Pidfile. This is the type returned by :func:`create_pidlock`. See Also: Best practice is to not use this directly but rather use the :func:`create_pidlock` function instead: more convenient and also removes stale pidfiles (when the process holding the lock is no longer running). """#: Path to the pid lock file.path=Nonedef__init__(self,path):self.path=os.path.abspath(path)
[文档]defread_pid(self):"""Read and return the current pid."""withignore_errno('ENOENT'):withopen(self.path)asfh:line=fh.readline()ifline.strip()==line:# must contain '\n'raiseValueError(f'Partial or invalid pidfile {self.path}')try:returnint(line.strip())exceptValueError:raiseValueError(f'pidfile {self.path} contents invalid.')
[文档]defremove(self):"""Remove the lock."""withignore_errno(errno.ENOENT,errno.EACCES):os.unlink(self.path)
[文档]defremove_if_stale(self):"""Remove the lock if the process isn't running. I.e. process does not respond to signal. """try:pid=self.read_pid()exceptValueError:print('Broken pidfile found - Removing it.',file=sys.stderr)self.remove()returnTrueifnotpid:self.remove()returnTrueifpid==os.getpid():# this can be common in k8s pod with PID of 1 - don't killself.remove()returnTruetry:os.kill(pid,0)exceptOSErrorasexc:ifexc.errno==errno.ESRCHorexc.errno==errno.EPERM:print('Stale pidfile exists - Removing it.',file=sys.stderr)self.remove()returnTrueexceptSystemError:print('Stale pidfile exists - Removing it.',file=sys.stderr)self.remove()returnTruereturnFalse
[文档]defwrite_pid(self):pid=os.getpid()content=f'{pid}\n'pidfile_fd=os.open(self.path,PIDFILE_FLAGS,PIDFILE_MODE)pidfile=os.fdopen(pidfile_fd,'w')try:pidfile.write(content)# flush and sync so that the re-read below works.pidfile.flush()try:os.fsync(pidfile_fd)exceptAttributeError:# pragma: no coverpassfinally:pidfile.close()rfh=open(self.path)try:ifrfh.read()!=content:raiseLockFailed("Inconsistency: Pidfile content doesn't match at re-read")finally:rfh.close()
PIDFile=Pidfile# XXX compat alias
[文档]defcreate_pidlock(pidfile):"""Create and verify pidfile. If the pidfile already exists the program exits with an error message, however if the process it refers to isn't running anymore, the pidfile is deleted and the program continues. This function will automatically install an :mod:`atexit` handler to release the lock at exit, you can skip this by calling :func:`_create_pidlock` instead. Returns: Pidfile: used to manage the lock. Example: >>> pidlock = create_pidlock('/var/run/app.pid') """pidlock=_create_pidlock(pidfile)atexit.register(pidlock.release)returnpidlock
[文档]deffd_by_path(paths):"""Return a list of file descriptors. This method returns list of file descriptors corresponding to file paths passed in paths variable. Arguments: paths: List[str]: List of file paths. Returns: List[int]: List of file descriptors. Example: >>> keep = fd_by_path(['/dev/urandom', '/my/precious/']) """stats=set()forpathinpaths:try:fd=os.open(path,os.O_RDONLY)exceptOSError:continuetry:stats.add(os.fstat(fd)[1:3])finally:os.close(fd)deffd_in_stats(fd):try:returnos.fstat(fd)[1:3]instatsexceptOSError:returnFalsereturn[_fdfor_fdinrange(get_fdmax(2048))iffd_in_stats(_fd)]
[文档]classDaemonContext:"""Context manager daemonizing the process."""_is_open=Falsedef__init__(self,pidfile=None,workdir=None,umask=None,fake=False,after_chdir=None,after_forkers=True,**kwargs):ifisinstance(umask,str):# octal or decimal, depending on initial zero.umask=int(umask,8ifumask.startswith('0')else10)self.workdir=workdirorDAEMON_WORKDIRself.umask=umaskself.fake=fakeself.after_chdir=after_chdirself.after_forkers=after_forkersself.stdfds=(sys.stdin,sys.stdout,sys.stderr)
[文档]defopen(self):ifnotself._is_open:ifnotself.fake:self._detach()os.chdir(self.workdir)ifself.umaskisnotNone:os.umask(self.umask)ifself.after_chdir:self.after_chdir()ifnotself.fake:# We need to keep /dev/urandom from closing because# shelve needs it, and Beat needs shelve to start.keep=list(self.stdfds)+fd_by_path(['/dev/urandom'])close_open_fds(keep)forfdinself.stdfds:self.redirect_to_null(maybe_fileno(fd))ifself.after_forkersandmputilisnotNone:mputil._run_after_forkers()self._is_open=True
__exit__=closedef_detach(self):ifos.fork()==0:# first childos.setsid()# create new sessionifos.fork()>0:# pragma: no cover# second childos._exit(0)else:os._exit(0)returnself
[文档]defdetached(logfile=None,pidfile=None,uid=None,gid=None,umask=0,workdir=None,fake=False,**opts):"""Detach the current process in the background (daemonize). Arguments: logfile (str): Optional log file. The ability to write to this file will be verified before the process is detached. pidfile (str): Optional pid file. The pidfile won't be created, as this is the responsibility of the child. But the process will exit if the pid lock exists and the pid written is still running. uid (int, str): Optional user id or user name to change effective privileges to. gid (int, str): Optional group id or group name to change effective privileges to. umask (str, int): Optional umask that'll be effective in the child process. workdir (str): Optional new working directory. fake (bool): Don't actually detach, intended for debugging purposes. **opts (Any): Ignored. Example: >>> from celery.platforms import detached, create_pidlock >>> with detached( ... logfile='/var/log/app.log', ... pidfile='/var/run/app.pid', ... uid='nobody'): ... # Now in detached child process with effective user set to nobody, ... # and we know that our logfile can be written to, and that ... # the pidfile isn't locked. ... pidlock = create_pidlock('/var/run/app.pid') ... ... # Run the program ... program.run(logfile='/var/log/app.log') """ifnotresource:raiseRuntimeError('This platform does not support detach.')workdir=os.getcwd()ifworkdirisNoneelseworkdirsignals.reset('SIGCLD')# Make sure SIGCLD is using the default handler.maybe_drop_privileges(uid=uid,gid=gid)defafter_chdir_do():# Since without stderr any errors will be silently suppressed,# we need to know that we have access to the logfile.logfileandopen(logfile,'a').close()# Doesn't actually create the pidfile, but makes sure it's not stale.ifpidfile:_create_pidlock(pidfile).release()returnDaemonContext(umask=umask,workdir=workdir,fake=fake,after_chdir=after_chdir_do,)
[文档]defparse_uid(uid):"""Parse user id. Arguments: uid (str, int): Actual uid, or the username of a user. Returns: int: The actual uid. """try:returnint(uid)exceptValueError:try:returnpwd.getpwnam(uid).pw_uidexcept(AttributeError,KeyError):raiseKeyError(f'User does not exist: {uid}')
[文档]defparse_gid(gid):"""Parse group id. Arguments: gid (str, int): Actual gid, or the name of a group. Returns: int: The actual gid of the group. """try:returnint(gid)exceptValueError:try:returngrp.getgrnam(gid).gr_gidexcept(AttributeError,KeyError):raiseKeyError(f'Group does not exist: {gid}')
def_setgroups_hack(groups):# :fun:`setgroups` may have a platform-dependent limit,# and it's not always possible to know in advance what this limit# is, so we use this ugly hack stolen from glibc.groups=groups[:]while1:try:returnos.setgroups(groups)exceptValueError:# error from Python's check.iflen(groups)<=1:raisegroups[:]=groups[:-1]exceptOSErrorasexc:# error from the OS.ifexc.errno!=errno.EINVALorlen(groups)<=1:raisegroups[:]=groups[:-1]
[文档]defsetgroups(groups):"""Set active groups from a list of group ids."""max_groups=Nonetry:max_groups=os.sysconf('SC_NGROUPS_MAX')exceptException:# pylint: disable=broad-exceptpasstry:return_setgroups_hack(groups[:max_groups])exceptOSErrorasexc:ifexc.errno!=errno.EPERM:raiseifany(groupnotingroupsforgroupinos.getgroups()):# we shouldn't be allowed to change to this group.raise
[文档]definitgroups(uid,gid):"""Init process group permissions. Compat version of :func:`os.initgroups` that was first added to Python 2.7. """ifnotpwd:# pragma: no coverreturnusername=pwd.getpwuid(uid)[0]ifhasattr(os,'initgroups'):# Python 2.7+returnos.initgroups(username,gid)groups=[gr.gr_gidforgringrp.getgrall()ifusernameingr.gr_mem]setgroups(groups)
[文档]defsetgid(gid):"""Version of :func:`os.setgid` supporting group names."""os.setgid(parse_gid(gid))
[文档]defsetuid(uid):"""Version of :func:`os.setuid` supporting usernames."""os.setuid(parse_uid(uid))
[文档]defmaybe_drop_privileges(uid=None,gid=None):"""Change process privileges to new user/group. If UID and GID is specified, the real user/group is changed. If only UID is specified, the real user is changed, and the group is changed to the users primary group. If only GID is specified, only the group is changed. """ifsys.platform=='win32':returnifos.geteuid():# no point trying to setuid unless we're root.ifnotos.getuid():raiseSecurityError('contact support')uid=uidandparse_uid(uid)gid=gidandparse_gid(gid)ifuid:_setuid(uid,gid)else:gidandsetgid(gid)ifuidandnotos.getuid()andnotos.geteuid():raiseSecurityError('Still root uid after drop privileges!')ifgidandnotos.getgid()andnotos.getegid():raiseSecurityError('Still root gid after drop privileges!')
def_setuid(uid,gid):# If GID isn't defined, get the primary GID of the user.ifnotgidandpwd:gid=pwd.getpwuid(uid).pw_gid# Must set the GID before initgroups(), as setgid()# is known to zap the group list on some platforms.# setgid must happen before setuid (otherwise the setgid operation# may fail because of insufficient privileges and possibly stay# in a privileged group).setgid(gid)initgroups(uid,gid)# at last:setuid(uid)# ... and make sure privileges cannot be restored:try:setuid(0)exceptOSErrorasexc:ifexc.errno!=errno.EPERM:raise# we should get here: cannot restore privileges,# everything was fine.else:raiseSecurityError('non-root user able to restore privileges after setuid.')ifhasattr(_signal,'setitimer'):def_arm_alarm(seconds):_signal.setitimer(_signal.ITIMER_REAL,seconds)else:def_arm_alarm(seconds):_signal.alarm(math.ceil(seconds))classSignals:"""Convenience interface to :mod:`signals`. If the requested signal isn't supported on the current platform, the operation will be ignored. Example: >>> from celery.platforms import signals >>> from proj.handlers import my_handler >>> signals['INT'] = my_handler >>> signals['INT'] my_handler >>> signals.supported('INT') True >>> signals.signum('INT') 2 >>> signals.ignore('USR1') >>> signals['USR1'] == signals.ignored True >>> signals.reset('USR1') >>> signals['USR1'] == signals.default True >>> from proj.handlers import exit_handler, hup_handler >>> signals.update(INT=exit_handler, ... TERM=exit_handler, ... HUP=hup_handler) """ignored=_signal.SIG_IGNdefault=_signal.SIG_DFLdefarm_alarm(self,seconds):return_arm_alarm(seconds)defreset_alarm(self):return_signal.alarm(0)defsupported(self,name):"""Return true value if signal by ``name`` exists on this platform."""try:self.signum(name)exceptAttributeError:returnFalseelse:returnTruedefsignum(self,name):"""Get signal number by name."""ifisinstance(name,numbers.Integral):returnnameifnotisinstance(name,str) \
ornotname.isupper():raiseTypeError('signal name must be uppercase string.')ifnotname.startswith('SIG'):name='SIG'+namereturngetattr(_signal,name)defreset(self,*signal_names):"""Reset signals to the default signal handler. Does nothing if the platform has no support for signals, or the specified signal in particular. """self.update((sig,self.default)forsiginsignal_names)defignore(self,*names):"""Ignore signal using :const:`SIG_IGN`. Does nothing if the platform has no support for signals, or the specified signal in particular. """self.update((sig,self.ignored)forsiginnames)def__getitem__(self,name):return_signal.getsignal(self.signum(name))def__setitem__(self,name,handler):"""Install signal handler. Does nothing if the current platform has no support for signals, or the specified signal in particular. """try:_signal.signal(self.signum(name),handler)except(AttributeError,ValueError):passdefupdate(self,_d_=None,**sigmap):"""Set signal handlers from a mapping."""forname,handlerindict(_d_or{},**sigmap).items():self[name]=handlersignals=Signals()get_signal=signals.signum# compatinstall_signal_handler=signals.__setitem__# compatreset_signal=signals.reset# compatignore_signal=signals.ignore# compat
[文档]defsignal_name(signum):"""Return name of signal from signal number."""returnSIGMAP[signum][3:]
defstrargv(argv):arg_start=2if'manage'inargv[0]else1iflen(argv)>arg_start:return' '.join(argv[arg_start:])return''defset_pdeathsig(name):"""Sends signal ``name`` to process when parent process terminates."""ifsignals.supported('SIGKILL'):try:_set_pdeathsig(signals.signum('SIGKILL'))exceptOSError:# We ignore when OS does not support set_pdeathsigpass
[文档]defset_process_title(progname,info=None):"""Set the :command:`ps` name for the currently running process. Only works if :pypi:`setproctitle` is installed. """proctitle=f'[{progname}]'proctitle=f'{proctitle}{info}'ifinfoelseproctitleif_setproctitle:_setproctitle.setproctitle(safe_str(proctitle))returnproctitle
ifos.environ.get('NOSETPS'):# pragma: no coverdefset_mp_process_title(*a,**k):"""Disabled feature."""else:
[文档]defset_mp_process_title(progname,info=None,hostname=None):"""Set the :command:`ps` name from the current process name. Only works if :pypi:`setproctitle` is installed. """ifhostname:progname=f'{progname}: {hostname}'name=current_process().nameifcurrent_processelse'MainProcess'returnset_process_title(f'{progname}:{name}',info=info)
[文档]defget_errno_name(n):"""Get errno for string (e.g., ``ENOENT``)."""ifisinstance(n,str):returngetattr(errno,n)returnn
[文档]@contextmanagerdefignore_errno(*errnos,**kwargs):"""Context manager to ignore specific POSIX error codes. Takes a list of error codes to ignore: this can be either the name of the code, or the code integer itself:: >>> with ignore_errno('ENOENT'): ... with open('foo', 'r') as fh: ... return fh.read() >>> with ignore_errno(errno.ENOENT, errno.EPERM): ... pass Arguments: types (Tuple[Exception]): A tuple of exceptions to ignore (when the errno matches). Defaults to :exc:`Exception`. """types=kwargs.get('types')or(Exception,)errnos=[get_errno_name(errno)forerrnoinerrnos]try:yieldexcepttypesasexc:ifnothasattr(exc,'errno'):raiseifexc.errnonotinerrnos:raise
defcheck_privileges(accept_content):ifgrpisNoneorpwdisNone:returnpickle_or_serialize=('pickle'inaccept_contentor'application/group-python-serialize'inaccept_content)uid=os.getuid()ifhasattr(os,'getuid')else65535gid=os.getgid()ifhasattr(os,'getgid')else65535euid=os.geteuid()ifhasattr(os,'geteuid')else65535egid=os.getegid()ifhasattr(os,'getegid')else65535ifhasattr(os,'fchown'):ifnotall(hasattr(os,attr)forattrin('getuid','getgid','geteuid','getegid')):raiseSecurityError('suspicious platform, contact support')# Get the group database entry for the current user's group and effective# group id using grp.getgrgid() method# We must handle the case where either the gid or the egid are not found.try:gid_entry=grp.getgrgid(gid)egid_entry=grp.getgrgid(egid)exceptKeyError:warnings.warn(SecurityWarning(ASSUMING_ROOT))_warn_or_raise_security_error(egid,euid,gid,uid,pickle_or_serialize)return# Get the group and effective group name based on gidgid_grp_name=gid_entry[0]egid_grp_name=egid_entry[0]# Create lists to use in validation step later.gids_in_use=(gid_grp_name,egid_grp_name)groups_with_security_risk=('sudo','wheel')is_root=uid==0oreuid==0# Confirm that the gid and egid are not one that# can be used to escalate privileges.ifis_rootorany(groupingids_in_useforgroupingroups_with_security_risk):_warn_or_raise_security_error(egid,euid,gid,uid,pickle_or_serialize)def_warn_or_raise_security_error(egid,euid,gid,uid,pickle_or_serialize):c_force_root=os.environ.get('C_FORCE_ROOT',False)ifpickle_or_serializeandnotc_force_root:raiseSecurityError(ROOT_DISALLOWED.format(uid=uid,euid=euid,gid=gid,egid=egid,))warnings.warn(SecurityWarning(ROOT_DISCOURAGED.format(uid=uid,euid=euid,gid=gid,egid=egid,)))