使用Python的安全命令执行：subprocess.Popen

作者：Maximilian Batz | 2020-05-15

在开发picockpit-client时，安全问题对我很重要。

以下内容适用于Linux系统（但可能适用于所有类似Unix的系统，包括MacOS）。

Python允许使用子进程模块来运行外部命令。

输入subprocess

在即将到来的PiCockpit版本中，用户将能够创建自己的按钮（只需在Pi上编辑一个JSON文件），以便运行命令。

PiCockpit出厂时默认有三个按钮。

这些命令都需要root权限，picockpit-client默认是以root权限运行在你的系统上。

但你想执行的其他命令，应该以较低权限的用户运行（我们默认为用户 "pi"）。事实上，有些命令，比如Chromium浏览器会拒绝以root身份运行。

简而言之，我们需要一个解决方案，使用不同的用户来运行这些进程。

这是可能的，这要归功于 预先执行_fn popen的参数。

以下是我的代码。

def thread_run(self):
     def sanitize(input_string):
         #https://maxharp3r.wordpress.com/2008/05/15/pythons-minidom-xml-and-illegal-unicode-characters/
         RE_XML_ILLEGAL = u'([\u0000-\u0008\u000b-\u000c\u000e\u001f\ufffe-\uffff])' + \\
                          u'|' + \
                          u'([%s-%s][^%s-%s])|([^%s-%s][%s-%s])|([%s-%s]$)|(^[%s-%s])’ % \
                          (Chr(0xd800), chr(0xdbff), chr(0xdc00), chr(0xdfff)。
                           chr(0xd800), chr(0xdbff), chr(0xdc00), chr(0xdfff)。
                           chr(0xd800), chr(0xdbff), chr(0xdc00), chr(0xdfff))
         return re.sub(RE_XML_ILLEGAL, "", input_string)。

    def demote(user, uid, gid):
         def demote_function():
             print("starting")
             print ('uid, gid = %d, %d' % (os.getuid(), os.getgid() ))
             print (os.getgroups())
             #的initgroups必须在我们失去设置权限之前运行!
             os.initgroups(user, gid)
             os.setgid(gid)
             # 这个必须最后运行
             os.setuid(uid)
             print("完成降级")。
             print('uid, gid = %d, %d' % (os.getuid(), os.getgid() )
             print (os.getgroups())
         返回demote_function

    def run_cmd(cmd, cmd_def):
         return_code = 0
         return_error = ""
         return_result = ""
         timeout = None
         用户 = "pi"
         如果cmd_def中的 "timeout"。
             如果isinstance(cmd_def["timeout"], int)。
                 如果cmd_def["timeout"] > 0:
                     timeout = cmd_def["timeout"]
         如果cmd_def中的 "user"。
             如果isinstance(cmd_def["user"], str)。
                 user = cmd_def["user"]
         尝试。
             print("picontrol :: setup of user rights")
             uid = pwd.getpwnam(user).pw_uid
             gid = pwd.getpwnam(user).pw_gid
             print("picontrol :: requested user " + user + " uid " + str(uid) + " gid " + str(gid))
             print("picontrol :: starting command ... " )
             print(cmd)
             proc = subprocess.Popen(
                 cmd。
                 stdout=subprocess.PIPE。
                 stderr=subprocess.PIPE。
                 preexec_fn=demote(user, uid, gid)
             )
             # TODO：以不同方式处理超时问题
                 # timeout=timeout)
             print("picontrol :: AFTER subprocess.Popen")
         除非出现 "找不到文件 "的错误。
             return_code = errno.ENOENT
             return_error = "在你的系统中没有找到命令" + " ".join(cmd)
         # TODO：这个东西究竟会不会被调用？
         除了subprocess.TimeoutExpired。
             return_code = errno.ETIME
             return_error = "命令超时过期（" + str(cmd_def["timeout"]) + " sec）"
         除例外情况外，如e。
             如果 hasattr(e, 'errno')。
                 return_code = e.errno
             否则。
                 return_code = 1
             如果 hasattr(e, 'message')。
                 return_error = str(e.message)
             否则。
                 return_error = str(e)
         否则。
             #进程可以正常执行，启动时没有例外
             process_running = True
             而process_running。
                 如果proc.poll()不是 None。
                     process_running = False
                 如果self.stop_flag.is_set()。
                     print("在process_running中收到停止标志，将process_running设置为false")
                     process_running = False
                     #，之后还要做主动终止的步骤。
                 #半秒的分辨率
                 时间.睡眠(0.5)
             return_code = proc.returncode
             return_error = sanitize(proc.stderr.read().decode('UTF-8').rstrip("`r\n'))
             return_result = sanitize(proc.stdout.read().decode('UTF-8').rstrip("r\n"))

        返回（return_code, return_error, return_result）。

这段代码是一个进展中的工作，所以它的一部分可能无法工作。然而，我想在这篇博文中讨论的部分确实可以工作。

Linux下的用户

一个用户名，在Linux下，将被分配一个用户ID，一个用户组，而用户将是一个会员的其他群体。

/etc/passwd 包含系统中定义的用户。

这些条目可以被解码如下。

user : password : user id (uid) : group id (gid) : full name of user (GECOS) : user home directory : login shell

有趣的附注：在/etc/passwd中存储密码（以加密的形式）是可选的（如果不存储密码，就会有一个 x 存储在这里）；由于这个文件是世界可读的，许多系统选择使用根可读的/etc/shadow代替。

其中确实包含了密码。

对于picockpit，我们从JSON设置文件中获得用户名，并希望相应地设置其他一切。所以我们一开始就用。

uid = pwd.getpwnam(user).pw_uid
gid = pwd.getpwnam(user).pw_gid

(请务必 输入pwd).

pwd模块允许我们使用Python访问/etc/passwd文件。 getpwnam通过用户的名字查找相应的行，在我们的例子中是 "pi"，或 "root"（或任何你想要的，任何在JSON文件中设置的）。

这将帮助我们获得用户的用户ID（uid）和他们的组ID（gid）。

接下来我们调用subprocess.Popen。

proc = subprocess.Popen(
     cmd。
     stdout=subprocess.PIPE。
     stderr=subprocess.PIPE。
     preexec_fn=demote(user, uid, gid)
)

stdout和stderr是PIPEd，所以我们可以稍后捕获它们。

preexec_fn被设置为demote(user, uid, gid)。

重要的是!

这是很重要的一点。 preexec_fn期望有一个函数 句柄，它将在它所启动的新进程的上下文中执行（因此该进程将自我降级！）。这个函数将在实际代码之前运行。 因此，我们可以确定，代码将在我们希望它执行的情况下执行。.

我们在这里传递demote的方式，看起来好像是在调用这个函数。

但请看降级的定义。

def demote(user, uid, gid):
     def demote_function():
        print("starting")
         print ('uid, gid = %d, %d' % (os.getuid(), os.getgid() ))
         print (os.getgroups())
         #的initgroups必须在我们失去设置权限之前运行!
         os.initgroups(user, gid)
         os.setgid(gid)
         # 这个必须最后运行
         os.setuid(uid)
         print("完成降级")。
         print('uid, gid = %d, %d' % (os.getuid(), os.getgid() )
         print (os.getgroups())
     返回demote_function

这个函数将另一个函数包裹在其内部，并将其返回。使用上面规定的语法，我们实际上是在 返回一个函数 但也能向它传递参数。

有点像在吃蛋糕，也在吃它。很好笑一笑