[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