Function check_privileges

celery/platforms.py:788–830

(accept_content)

def check_privileges(accept_content):
    if grp is None or pwd is None:
        return
    pickle_or_serialize = ('pickle' in accept_content
                           or 'application/group-python-serialize' in 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')

    # 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)
    except KeyError:
        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 gid
    gid_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 == 0 or euid == 0
    # Confirm that the gid and egid are not one that
    # can be used to escalate privileges.
    if is_root or any(group in gids_in_use
                      for group in groups_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):