[rhinstaller/anaconda/pulls/315 master] Do not use libuser

jkonecny12 installerbot-noreply at redhat.com
Fri Aug 21 06:47:36 UTC 2015


> @@ -286,179 +251,166 @@ def createUser(self, user_name, *args, **kwargs):
>                          in.  homedir will be interpreted relative to this.
>                          Defaults to /mnt/sysimage.
>             shell     -- The shell for the new user.  If none is given, the
> -                        libuser default is used.
> +                        login.defs default is used.
>             uid       -- The UID for the new user.  If none is given, the next
>                          available one is used.
>             gid       -- The GID for the new user.  If none is given, the next
>                          available one is used.
> +           mkmailspool -- Whether to create a /var/spool/mail file for the user.
> +                        Defaults to True.
>          """
> -        childpid = self._prepareChroot(kwargs.get("root", iutil.getSysroot()))
> -        if childpid == 0:
> -            if self.admin.lookupUserByName(user_name):
> -                log.error("User %s already exists, not creating.", user_name)
> -                os._exit(1)
> -
> -            userEnt = self.admin.initUser(user_name)
> -            groupEnt = self.admin.initGroup(user_name)
> -
> -            gid = kwargs.get("gid") or -1
> -            if gid >= 0:
> -                groupEnt.set(libuser.GIDNUMBER, kwargs["gid"])
> -
> -            grpLst = [grp for grp in map(self.admin.lookupGroupByName, kwargs.get("groups", [])) if grp]
> -            userEnt.set(libuser.GIDNUMBER, [groupEnt.get(libuser.GIDNUMBER)[0]] +
> -                        [grp.get(libuser.GIDNUMBER)[0] for grp in grpLst])
> -
> -            homedir = kwargs.get("homedir", None)
> -            if not homedir:
> -                homedir = "/home/" + user_name
> -            # libuser expects the parent directory tree to exist.
> -            parent_dir = iutil.parent_dir(homedir)
> -            if parent_dir:
> -                iutil.mkdirChain(parent_dir)
> -            userEnt.set(libuser.HOMEDIRECTORY, homedir)
> -
> -            if kwargs.get("shell", False):
> -                userEnt.set(libuser.LOGINSHELL, kwargs["shell"])
> -
> -            uid = kwargs.get("uid") or -1
> -            if uid >= 0:
> -                userEnt.set(libuser.UIDNUMBER, kwargs["uid"])
> -
> -            if kwargs.get("gecos", False):
> -                userEnt.set(libuser.GECOS, kwargs["gecos"])
> -
> -            # need to create home directory for the user or does it already exist?
> -            # userEnt.get returns lists (usually with a single item)
> -            mk_homedir = not os.path.exists(userEnt.get(libuser.HOMEDIRECTORY)[0])
>  
> -            try:
> -                self.admin.addUser(userEnt, mkmailspool=kwargs.get("mkmailspool", True),
> -                                   mkhomedir=mk_homedir)
> -            except RuntimeError as e:
> -                log.critical("Error when creating new user: %s", e)
> -                os._exit(1)
> +        root = kwargs.get("root", iutil.getSysroot())
>  
> -            try:
> -                self.admin.addGroup(groupEnt)
> -            except RuntimeError as e:
> -                log.critical("Error when creating new group: %s", e)
> -                os._exit(1)
> -
> -            if not mk_homedir:
> -                try:
> -                    stats = os.stat(userEnt.get(libuser.HOMEDIRECTORY)[0])
> -                    orig_uid = stats.st_uid
> -                    orig_gid = stats.st_gid
> -
> -                    log.info("Home directory for the user %s already existed, "
> -                             "fixing the owner and SELinux context.", user_name)
> -                    # home directory already existed, change owner of it properly
> -                    iutil.chown_dir_tree(userEnt.get(libuser.HOMEDIRECTORY)[0],
> -                                         userEnt.get(libuser.UIDNUMBER)[0],
> -                                         groupEnt.get(libuser.GIDNUMBER)[0],
> -                                         orig_uid, orig_gid)
> -                    iutil.execWithRedirect("restorecon",
> -                                           ["-r", userEnt.get(libuser.HOMEDIRECTORY)[0]])
> -                except OSError as e:
> -                    log.critical("Unable to change owner of existing home directory: %s",
> -                            os.strerror)
> -                    os._exit(1)
> -
> -            pw = kwargs.get("password", False)
> -            try:
> -                if pw:
> -                    if kwargs.get("isCrypted", False):
> -                        password = kwargs["password"]
> -                    else:
> -                        password = cryptPassword(kwargs["password"], algo=kwargs.get("algo", None))
> -                    self.admin.setpassUser(userEnt, password, True)
> -                    userEnt.set(libuser.SHADOWLASTCHANGE, "")
> -                    self.admin.modifyUser(userEnt)
> -                elif pw == "":
> -                    # Setup the account with *NO* password
> -                    self.admin.unlockUser(userEnt)
> -                    log.info("user account %s setup with no password", user_name)
> -
> -                if kwargs.get("lock", False):
> -                    self.admin.lockUser(userEnt)
> -                    log.info("user account %s locked", user_name)
> -            # setpassUser raises SystemError on failure, while unlockUser and lockUser
> -            # raise RuntimeError
> -            except (RuntimeError, SystemError) as e:
> -                log.critical("Unable to set password for new user: %s", e)
> -                os._exit(1)
> -
> -            # Add the user to all the groups they should be part of.
> -            grpLst.append(self.admin.lookupGroupByName(user_name))
> -            try:
> -                for grp in grpLst:
> -                    grp.add(libuser.MEMBERNAME, user_name)
> -                    self.admin.modifyGroup(grp)
> -            except RuntimeError as e:
> -                log.critical("Unable to add user to groups: %s", e)
> -                os._exit(1)
> -
> -            os._exit(0)
> -        else:
> -            return self._finishChroot(childpid)
> +        if self.checkUserExists(user_name, root):
> +            raise ValueError("User %s already exists" % user_name)
>  
> -    def checkUserExists(self, username, root=None):
> -        childpid = self._prepareChroot(root)
> +        args = ["-R", root]
>  
> -        if childpid == 0:
> -            if self.admin.lookupUserByName(username):
> -                os._exit(0)
> -            else:
> -                os._exit(1)
> +        # If a specific gid is requested, create the user group with that GID.
> +        # Otherwise let useradd do it automatically.
> +        if "gid" in kwargs:
> +            self.createGroup(user_name, gid=kwargs['gid'], root=root)
> +            args.extend(['-g', str(kwargs['gid'])])
> +        else:
> +            args.append('-U')
> +
> +        if "groups" in kwargs:
> +            args.extend(['-G', ",".join(kwargs["groups"])])
> +
> +        homedir = kwargs.get("homedir", "/home/" + user_name)
> +        # useradd expects the parent directory tree to exist.
> +        parent_dir = iutil.parent_dir(root + homedir)
> +        if parent_dir:
> +            iutil.mkdirChain(parent_dir)
> +        args.extend(["-d", homedir])
> +
> +        # Check whether the directory exists or if useradd should create it
> +        mk_homedir = not os.path.exists(root + homedir)
> +        if mk_homedir:
> +            args.append("-m")
>          else:
> -            return self._finishChroot(childpid)
> +            args.append("-M")
>  
> -    def setUserPassword(self, username, password, isCrypted, lock, algo=None):
> -        user = self.admin.lookupUserByName(username)
> +        if "shell" in kwargs:
> +            args.extend(["-s", kwargs["shell"]])
>  
> -        if isCrypted:
> -            self.admin.setpassUser(user, password, True)
> +        if "uid" in kwargs:
> +            args.extend(["-u", str(kwargs["uid"])])
> +
> +        if "gecos" in kwargs:
> +            args.extend(["-c", kwargs["gecos"]])
> +
> +        # The mail spool is annoying. We can control where it is created, but
> +        # not whether it is created. So, have useradd create it, maybe, in a
> +        # temp directory, remove that directory, and then if we actually wanted
> +        # a spool file just make an empty file in /var/spool/mail.
> +        spool_dir = tempfile.mkdtemp()
> +        args.extend(["-K", "MAIL_DIR=%s" % spool_dir])
> +
> +        args.append(user_name)
> +        try:
> +            with self._ensureLoginDefs(root):
> +                status = iutil.execWithRedirect("useradd", args)
> +
> +            if status == 4:
> +                raise ValueError("UID %s already exists" % kwargs.get("uid", -1))
> +            elif status == 6:
> +                raise ValueError("Invalid groups %s" % kwargs.get("groups", []))
> +            elif status == 9:
> +                raise ValueError("User %s already exists" % user_name)
> +            elif status != 0:
> +                raise OSError("Unable to create user %s: status=%s" % (user_name, status))
> +        finally:
> +            shutil.rmtree(spool_dir)
> +
> +        if kwargs.get("mkmailspool", True):
> +            iutil.mkdirChain(root + "/var/spool/mail")
> +            open(root + "/var/spool/mail/" + user_name, "w").close()
> +
> +        if not mk_homedir:
> +            try:
> +                stats = os.stat(root + homedir)
> +                orig_uid = stats.st_uid
> +                orig_gid = stats.st_gid
> +
> +                # Gett the UID and GID of the created user
> +                pwent = self._getpwnam(user_name, root)
> +
> +                log.info("Home directory for the user %s already existed, "
> +                         "fixing the owner and SELinux context.", user_name)
> +                # home directory already existed, change owner of it properly
> +                iutil.chown_dir_tree(root + homedir,
> +                                     pwent[2], pwent[3],
> +                                     orig_uid, orig_gid)
> +                iutil.execWithRedirect("restorecon", ["-r", root + homedir])
> +            except OSError as e:
> +                log.critical("Unable to change owner of existing home directory: %s", e.strerror)
> +                raise
> +
> +        pw = kwargs.get("password", "")
> +        crypted = kwargs.get("isCrypted", False)
> +        algo = kwargs.get("algo", None)
> +        lock = kwargs.get("lock", False)
> +
> +        self.setUserPassword(user_name, pw, crypted, lock, algo, root)
> +
> +    def checkUserExists(self, username, root=None):
> +        if self._getpwnam(username, root):
> +            return True
> +
> +        return False
> +
> +    def setUserPassword(self, username, password, isCrypted, lock, algo=None, root="/"):
> +        if password:
> +            if not isCrypted:
> +                password = cryptPassword(password, algo)
>          else:
> -            self.admin.setpassUser(user, cryptPassword(password, algo=algo), True)
> +            log.info("user account %s setup with no password", username)
>  
>          if lock:
> -            self.admin.lockUser(user)
> +            password = "!" + password
> +            log.info("user account %s locked", username)
>  
> -        user.set(libuser.SHADOWLASTCHANGE, "")
> -        return self.admin.modifyUser(user)
> +        proc = iutil.startProgram(["chpasswd", "-R", root, "-e"], stdin=subprocess.PIPE)
> +        proc.communicate(("%s:%s\n" % (username, password)).encode("utf-8"))
> +        if proc.returncode != 0:
> +            raise OSError("Unable to set password for new user: status=%s" % proc.returncode)
>  
> -    def setRootPassword(self, password, isCrypted=False, isLocked=False, algo=None):
> -        return self.setUserPassword("root", password, isCrypted, isLocked, algo)
> +        # Reset sp_lstchg to an empty string. On systems with no rtc, this
> +        # field can be set to 0, which has a special meaning that the password
> +        # must be reset on the next login.
> +        iutil.execWithRedirect("chage", ["-R", root, "-d", "", username])
> +
> +    def setRootPassword(self, password, isCrypted=False, isLocked=False, algo=None, root="/"):
> +        return self.setUserPassword("root", password, isCrypted, isLocked, algo, root)
>  
>      def setUserSshKey(self, username, key, **kwargs):
> -        childpid = self._prepareChroot(kwargs.get("root", iutil.getSysroot()))
> -
> -        if childpid == 0:
> -            user = self.admin.lookupUserByName(username)
> -            if not user:
> -                log.error("setUserSshKey: user %s does not exist", username)
> -                os._exit(1)
> -
> -            homedir = user.get(libuser.HOMEDIRECTORY)[0]
> -            if not os.path.exists(homedir):
> -                log.error("setUserSshKey: home directory for %s does not exist", username)
> -                os._exit(1)
> -
> -            sshdir = os.path.join(homedir, ".ssh")
> -            if not os.path.isdir(sshdir):
> -                os.mkdir(sshdir, 0o700)
> -                iutil.eintr_retry_call(os.chown, sshdir, user.get(libuser.UIDNUMBER)[0], user.get(libuser.GIDNUMBER)[0])
> -
> -            authfile = os.path.join(sshdir, "authorized_keys")
> -            authfile_existed = os.path.exists(authfile)
> -            with iutil.open_with_perm(authfile, "a", 0o600) as f:
> -                f.write(key + "\n")
> -
> -            # Only change ownership if we created it
> -            if not authfile_existed:
> -                iutil.eintr_retry_call(os.chown, authfile, user.get(libuser.UIDNUMBER)[0], user.get(libuser.GIDNUMBER)[0])
> -                iutil.execWithRedirect("restorecon", ["-r", sshdir])
> -            os._exit(0)
> -        else:
> -            return self._finishChroot(childpid)
> +        root = kwargs.get("root", iutil.getSysroot())
> +
> +        pwent = self._getpwnam(username, root)
> +        if not pwent:
> +            raise ValueError("setUserSshKey: user %s does not exist" % username)
> +
> +        homedir = root + pwent[5]
> +        if not os.path.exists(homedir):
> +            log.error("setUserSshKey: home directory for %s does not exist", username)
> +            os._exit(1)
> +
> +        uid = pwent[2]
> +        gid = pwent[3]
> +
> +        sshdir = os.path.join(homedir, ".ssh")
> +        if not os.path.isdir(sshdir):
> +            os.mkdir(sshdir, 0o700)
> +            iutil.eintr_retry_call(os.chown, sshdir, int(uid), int(gid))
> +
> +        authfile = os.path.join(sshdir, "authorized_keys")
> +        authfile_existed = os.path.exists(authfile)
> +        with iutil.open_with_perm(authfile, "a", 0o600) as f:
> +            f.write(key + "\n")
> +
> +        # Only change ownership if we created it
> +        if not authfile_existed:

Maybe I missed something but I think we want to change ownership only when the out file "is" created.

-- 
To view this pull request on github, visit https://github.com/rhinstaller/anaconda/pull/315#discussion_r37609977


More information about the anaconda-patches mailing list