/usr/lib/python3/dist-packages/celery/platforms.py is in python3-celery 4.1.0-2ubuntu1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 | # -*- coding: utf-8 -*-
"""Platforms.
Utilities dealing with platform specifics: signals, daemonization,
users, groups, and so on.
"""
from __future__ import absolute_import, print_function, unicode_literals
import atexit
import errno
import math
import numbers
import os
import platform as _platform
import signal as _signal
import sys
import warnings
from collections import namedtuple
from billiard.compat import get_fdmax, close_open_fds
# fileno used to be in this module
from kombu.utils.compat import maybe_fileno
from kombu.utils.encoding import safe_str
from contextlib import contextmanager
from .exceptions import SecurityError
from .local import try_import
from .five import items, reraise, string_t
try:
from billiard.process import current_process
except ImportError: # pragma: no cover
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',
]
# exitcodes
EX_OK = getattr(os, 'EX_OK', 0)
EX_FAILURE = 1
EX_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_WRONLY
PIDFILE_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})"""
_range = namedtuple('_range', ('start', 'stop'))
C_FORCE_ROOT = os.environ.get('C_FORCE_ROOT', False)
ROOT_DISALLOWED = """\
Running a worker with superuser privileges when the
worker accepts messages serialized with pickle is a very bad idea!
If you really want to continue then you have to set the C_FORCE_ROOT
environment 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 is
absolutely not recommended!
Please specify a different user using the -u option.
User information: uid={uid} euid={euid} gid={gid} egid={egid}
"""
SIGNAMES = {
sig for sig in dir(_signal)
if sig.startswith('SIG') and '_' not in sig
}
SIGMAP = {getattr(_signal, name): name for name in SIGNAMES}
def isatty(fh):
"""Return true if the process has a controlling terminal."""
try:
return fh.isatty()
except AttributeError:
pass
def pyimplementation():
"""Return string identifying the current Python implementation."""
if hasattr(_platform, 'python_implementation'):
return _platform.python_implementation()
elif sys.platform.startswith('java'):
return 'Jython ' + sys.platform
elif hasattr(sys, 'pypy_version_info'):
v = '.'.join(str(p) for p in sys.pypy_version_info[:3])
if sys.pypy_version_info[3:]:
v += '-' + ''.join(str(p) for p in sys.pypy_version_info[3:])
return 'PyPy ' + v
else:
return 'CPython'
class LockFailed(Exception):
"""Raised if a PID lock can't be acquired."""
class Pidfile(object):
"""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 = None
def __init__(self, path):
self.path = os.path.abspath(path)
def acquire(self):
"""Acquire lock."""
try:
self.write_pid()
except OSError as exc:
reraise(LockFailed, LockFailed(str(exc)), sys.exc_info()[2])
return self
__enter__ = acquire
def is_locked(self):
"""Return true if the pid lock exists."""
return os.path.exists(self.path)
def release(self, *args):
"""Release lock."""
self.remove()
__exit__ = release
def read_pid(self):
"""Read and return the current pid."""
with ignore_errno('ENOENT'):
with open(self.path, 'r') as fh:
line = fh.readline()
if line.strip() == line: # must contain '\n'
raise ValueError(
'Partial or invalid pidfile {0.path}'.format(self))
try:
return int(line.strip())
except ValueError:
raise ValueError(
'pidfile {0.path} contents invalid.'.format(self))
def remove(self):
"""Remove the lock."""
with ignore_errno(errno.ENOENT, errno.EACCES):
os.unlink(self.path)
def remove_if_stale(self):
"""Remove the lock if the process isn't running.
I.e. process does not respons to signal.
"""
try:
pid = self.read_pid()
except ValueError as exc:
print('Broken pidfile found - Removing it.', file=sys.stderr)
self.remove()
return True
if not pid:
self.remove()
return True
try:
os.kill(pid, 0)
except os.error as exc:
if exc.errno == errno.ESRCH:
print('Stale pidfile exists - Removing it.', file=sys.stderr)
self.remove()
return True
return False
def write_pid(self):
pid = os.getpid()
content = '{0}\n'.format(pid)
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)
except AttributeError: # pragma: no cover
pass
finally:
pidfile.close()
rfh = open(self.path)
try:
if rfh.read() != content:
raise LockFailed(
"Inconsistency: Pidfile content doesn't match at re-read")
finally:
rfh.close()
PIDFile = Pidfile # noqa: E305 XXX compat alias
def create_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)
return pidlock
def _create_pidlock(pidfile):
pidlock = Pidfile(pidfile)
if pidlock.is_locked() and not pidlock.remove_if_stale():
print(PIDLOCKED.format(pidfile, pidlock.read_pid()), file=sys.stderr)
raise SystemExit(EX_CANTCREAT)
pidlock.acquire()
return pidlock
def fd_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()
for path in paths:
try:
fd = os.open(path, os.O_RDONLY)
except OSError:
continue
try:
stats.add(os.fstat(fd)[1:3])
finally:
os.close(fd)
def fd_in_stats(fd):
try:
return os.fstat(fd)[1:3] in stats
except OSError:
return False
return [_fd for _fd in range(get_fdmax(2048)) if fd_in_stats(_fd)]
class DaemonContext(object):
"""Context manager daemonizing the process."""
_is_open = False
def __init__(self, pidfile=None, workdir=None, umask=None,
fake=False, after_chdir=None, after_forkers=True,
**kwargs):
if isinstance(umask, string_t):
# octal or decimal, depending on initial zero.
umask = int(umask, 8 if umask.startswith('0') else 10)
self.workdir = workdir or DAEMON_WORKDIR
self.umask = umask
self.fake = fake
self.after_chdir = after_chdir
self.after_forkers = after_forkers
self.stdfds = (sys.stdin, sys.stdout, sys.stderr)
def redirect_to_null(self, fd):
if fd is not None:
dest = os.open(os.devnull, os.O_RDWR)
os.dup2(dest, fd)
def open(self):
if not self._is_open:
if not self.fake:
self._detach()
os.chdir(self.workdir)
if self.umask is not None:
os.umask(self.umask)
if self.after_chdir:
self.after_chdir()
if not self.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)
for fd in self.stdfds:
self.redirect_to_null(maybe_fileno(fd))
if self.after_forkers and mputil is not None:
mputil._run_after_forkers()
self._is_open = True
__enter__ = open
def close(self, *args):
if self._is_open:
self._is_open = False
__exit__ = close
def _detach(self):
if os.fork() == 0: # first child
os.setsid() # create new session
if os.fork() > 0: # pragma: no cover
# second child
os._exit(0)
else:
os._exit(0)
return self
def detached(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')
"""
if not resource:
raise RuntimeError('This platform does not support detach.')
workdir = os.getcwd() if workdir is None else workdir
signals.reset('SIGCLD') # Make sure SIGCLD is using the default handler.
maybe_drop_privileges(uid=uid, gid=gid)
def after_chdir_do():
# Since without stderr any errors will be silently suppressed,
# we need to know that we have access to the logfile.
logfile and open(logfile, 'a').close()
# Doesn't actually create the pidfile, but makes sure it's not stale.
if pidfile:
_create_pidlock(pidfile).release()
return DaemonContext(
umask=umask, workdir=workdir, fake=fake, after_chdir=after_chdir_do,
)
def parse_uid(uid):
"""Parse user id.
Arguments:
uid (str, int): Actual uid, or the username of a user.
Returns:
int: The actual uid.
"""
try:
return int(uid)
except ValueError:
try:
return pwd.getpwnam(uid).pw_uid
except (AttributeError, KeyError):
raise KeyError('User does not exist: {0}'.format(uid))
def parse_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:
return int(gid)
except ValueError:
try:
return grp.getgrnam(gid).gr_gid
except (AttributeError, KeyError):
raise KeyError('Group does not exist: {0}'.format(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[:]
while 1:
try:
return os.setgroups(groups)
except ValueError: # error from Python's check.
if len(groups) <= 1:
raise
groups[:] = groups[:-1]
except OSError as exc: # error from the OS.
if exc.errno != errno.EINVAL or len(groups) <= 1:
raise
groups[:] = groups[:-1]
def setgroups(groups):
"""Set active groups from a list of group ids."""
max_groups = None
try:
max_groups = os.sysconf('SC_NGROUPS_MAX')
except Exception: # pylint: disable=broad-except
pass
try:
return _setgroups_hack(groups[:max_groups])
except OSError as exc:
if exc.errno != errno.EPERM:
raise
if any(group not in groups for group in os.getgroups()):
# we shouldn't be allowed to change to this group.
raise
def initgroups(uid, gid):
"""Init process group permissions.
Compat version of :func:`os.initgroups` that was first
added to Python 2.7.
"""
if not pwd: # pragma: no cover
return
username = pwd.getpwuid(uid)[0]
if hasattr(os, 'initgroups'): # Python 2.7+
return os.initgroups(username, gid)
groups = [gr.gr_gid for gr in grp.getgrall()
if username in gr.gr_mem]
setgroups(groups)
def setgid(gid):
"""Version of :func:`os.setgid` supporting group names."""
os.setgid(parse_gid(gid))
def setuid(uid):
"""Version of :func:`os.setuid` supporting usernames."""
os.setuid(parse_uid(uid))
def maybe_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.
"""
if sys.platform == 'win32':
return
if os.geteuid():
# no point trying to setuid unless we're root.
if not os.getuid():
raise SecurityError('contact support')
uid = uid and parse_uid(uid)
gid = gid and parse_gid(gid)
if uid:
_setuid(uid, gid)
else:
gid and setgid(gid)
if uid and not os.getuid() and not os.geteuid():
raise SecurityError('Still root uid after drop privileges!')
if gid and not os.getgid() and not os.getegid():
raise SecurityError('Still root gid after drop privileges!')
def _setuid(uid, gid):
# If GID isn't defined, get the primary GID of the user.
if not gid and pwd:
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)
except OSError as exc:
if exc.errno != errno.EPERM:
raise
# we should get here: cannot restore privileges,
# everything was fine.
else:
raise SecurityError(
'non-root user able to restore privileges after setuid.')
class Signals(object):
"""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_IGN
default = _signal.SIG_DFL
if hasattr(_signal, 'setitimer'):
def arm_alarm(self, seconds):
_signal.setitimer(_signal.ITIMER_REAL, seconds)
else: # pragma: no cover
try:
from itimer import alarm as _itimer_alarm # noqa
except ImportError:
def arm_alarm(self, seconds): # noqa
_signal.alarm(math.ceil(seconds))
else: # pragma: no cover
def arm_alarm(self, seconds): # noqa
return _itimer_alarm(seconds) # noqa
def reset_alarm(self):
return _signal.alarm(0)
def supported(self, name):
"""Return true value if signal by ``name`` exists on this platform."""
try:
self.signum(name)
except AttributeError:
return False
else:
return True
def signum(self, name):
"""Get signal number by name."""
if isinstance(name, numbers.Integral):
return name
if not isinstance(name, string_t) \
or not name.isupper():
raise TypeError('signal name must be uppercase string.')
if not name.startswith('SIG'):
name = 'SIG' + name
return getattr(_signal, name)
def reset(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) for sig in signal_names)
def ignore(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) for sig in names)
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):
pass
def update(self, _d_=None, **sigmap):
"""Set signal handlers from a mapping."""
for name, handler in items(dict(_d_ or {}, **sigmap)):
self[name] = handler
signals = Signals()
get_signal = signals.signum # compat
install_signal_handler = signals.__setitem__ # compat
reset_signal = signals.reset # compat
ignore_signal = signals.ignore # compat
def signal_name(signum):
"""Return name of signal from signal number."""
return SIGMAP[signum][3:]
def strargv(argv):
arg_start = 2 if 'manage' in argv[0] else 1
if len(argv) > arg_start:
return ' '.join(argv[arg_start:])
return ''
def set_process_title(progname, info=None):
"""Set the :command:`ps` name for the currently running process.
Only works if :pypi:`setproctitle` is installed.
"""
proctitle = '[{0}]'.format(progname)
proctitle = '{0} {1}'.format(proctitle, info) if info else proctitle
if _setproctitle:
_setproctitle.setproctitle(safe_str(proctitle))
return proctitle
if os.environ.get('NOSETPS'): # pragma: no cover
def set_mp_process_title(*a, **k):
"""Disabled feature."""
pass
else:
def set_mp_process_title(progname, info=None, hostname=None): # noqa
"""Set the :command:`ps` name from the current process name.
Only works if :pypi:`setproctitle` is installed.
"""
if hostname:
progname = '{0}: {1}'.format(progname, hostname)
name = current_process().name if current_process else 'MainProcess'
return set_process_title('{0}:{1}'.format(progname, name), info=info)
def get_errno_name(n):
"""Get errno for string (e.g., ``ENOENT``)."""
if isinstance(n, string_t):
return getattr(errno, n)
return n
@contextmanager
def ignore_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) for errno in errnos]
try:
yield
except types as exc:
if not hasattr(exc, 'errno'):
raise
if exc.errno not in errnos:
raise
def check_privileges(accept_content):
uid = os.getuid() if hasattr(os, 'getuid') else 65535
gid = os.getgid() if hasattr(os, 'getgid') else 65535
euid = os.geteuid() if hasattr(os, 'geteuid') else 65535
egid = os.getegid() if hasattr(os, 'getegid') else 65535
if hasattr(os, 'fchown'):
if not all(hasattr(os, attr)
for attr in ['getuid', 'getgid', 'geteuid', 'getegid']):
raise SecurityError('suspicious platform, contact support')
if not uid or not gid or not euid or not egid:
if ('pickle' in accept_content or
'application/x-python-serialize' in accept_content):
if not C_FORCE_ROOT:
try:
print(ROOT_DISALLOWED.format(
uid=uid, euid=euid, gid=gid, egid=egid,
), file=sys.stderr)
finally:
os._exit(1)
warnings.warn(RuntimeWarning(ROOT_DISCOURAGED.format(
uid=uid, euid=euid, gid=gid, egid=egid,
)))
|