Merge branch 'stable-2.16' into stable-2.17
[ganeti-github.git] / lib / backend.py
index 1d4d2e4..5b83290 100644 (file)
@@ -1467,7 +1467,9 @@ def AddNodeSshKey(node_uuid, node_name,
                   pub_key_file=pathutils.SSH_PUB_KEYS,
                   ssconf_store=None,
                   noded_cert_file=pathutils.NODED_CERT_FILE,
-                  run_cmd_fn=ssh.RunSshCmdWithStdin):
+                  run_cmd_fn=ssh.RunSshCmdWithStdin,
+                  ssh_update_debug=False,
+                  ssh_update_verbose=False):
   """Distributes a node's public SSH key across the cluster.
 
   Note that this function should only be executed on the master node, which
@@ -1503,7 +1505,9 @@ def AddNodeSshKey(node_uuid, node_name,
                            pub_key_file=pub_key_file,
                            ssconf_store=ssconf_store,
                            noded_cert_file=noded_cert_file,
-                           run_cmd_fn=run_cmd_fn)
+                           run_cmd_fn=run_cmd_fn,
+                           ssh_update_debug=ssh_update_debug,
+                           ssh_update_verbose=ssh_update_verbose)
 
 
 # Node info named tuple specifically for the use with AddNodeSshKeyBulk
@@ -1521,7 +1525,9 @@ def AddNodeSshKeyBulk(node_list,
                       pub_key_file=pathutils.SSH_PUB_KEYS,
                       ssconf_store=None,
                       noded_cert_file=pathutils.NODED_CERT_FILE,
-                      run_cmd_fn=ssh.RunSshCmdWithStdin):
+                      run_cmd_fn=ssh.RunSshCmdWithStdin,
+                      ssh_update_debug=False,
+                      ssh_update_verbose=False):
   """Distributes a node's public SSH key across the cluster.
 
   Note that this function should only be executed on the master node, which
@@ -1599,13 +1605,14 @@ def AddNodeSshKeyBulk(node_list,
         (constants.SSHS_OVERRIDE, all_keys)
 
       try:
+        backoff = 5  # seconds
         utils.RetryByNumberOfTimes(
-            constants.SSHS_MAX_RETRIES,
+            constants.SSHS_MAX_RETRIES, backoff,
             errors.SshUpdateError,
             run_cmd_fn, cluster_name, node_info.name, pathutils.SSH_UPDATE,
             ssh_port_map.get(node_info.name), node_data,
-            debug=False, verbose=False, use_cluster_key=False,
-            ask_key=False, strict_host_check=False)
+            debug=ssh_update_debug, verbose=ssh_update_verbose,
+            use_cluster_key=False, ask_key=False, strict_host_check=False)
       except errors.SshUpdateError as e:
         # Clean up the master's public key file if adding key fails
         if node_info.to_public_keys:
@@ -1645,13 +1652,13 @@ def AddNodeSshKeyBulk(node_list,
     if node in potential_master_candidates:
       logging.debug("Updating SSH key files of node '%s'.", node)
       try:
+        backoff = 5  # seconds
         utils.RetryByNumberOfTimes(
-            constants.SSHS_MAX_RETRIES,
-            errors.SshUpdateError,
+            constants.SSHS_MAX_RETRIES, backoff, errors.SshUpdateError,
             run_cmd_fn, cluster_name, node, pathutils.SSH_UPDATE,
             ssh_port_map.get(node), pot_mc_data,
-            debug=False, verbose=False, use_cluster_key=False,
-            ask_key=False, strict_host_check=False)
+            debug=ssh_update_debug, verbose=ssh_update_verbose,
+            use_cluster_key=False, ask_key=False, strict_host_check=False)
       except errors.SshUpdateError as last_exception:
         error_msg = ("When adding the key of node '%s', updating SSH key"
                      " files of node '%s' failed after %s retries."
@@ -1667,12 +1674,15 @@ def AddNodeSshKeyBulk(node_list,
       if to_authorized_keys:
         run_cmd_fn(cluster_name, node, pathutils.SSH_UPDATE,
                    ssh_port_map.get(node), base_data,
-                   debug=False, verbose=False, use_cluster_key=False,
-                   ask_key=False, strict_host_check=False)
+                   debug=ssh_update_debug, verbose=ssh_update_verbose,
+                   use_cluster_key=False, ask_key=False,
+                   strict_host_check=False)
 
   return node_errors
 
 
+# TODO: will be fixed with pending patch series.
+# pylint: disable=R0913
 def RemoveNodeSshKey(node_uuid, node_name,
                      master_candidate_uuids,
                      potential_master_candidates,
@@ -1686,7 +1696,9 @@ def RemoveNodeSshKey(node_uuid, node_name,
                      ssconf_store=None,
                      noded_cert_file=pathutils.NODED_CERT_FILE,
                      readd=False,
-                     run_cmd_fn=ssh.RunSshCmdWithStdin):
+                     run_cmd_fn=ssh.RunSshCmdWithStdin,
+                     ssh_update_debug=False,
+                     ssh_update_verbose=False):
   """Removes the node's SSH keys from the key files and distributes those.
 
   Note that at least one of the flags C{from_authorized_keys},
@@ -1740,7 +1752,9 @@ def RemoveNodeSshKey(node_uuid, node_name,
                               ssconf_store=ssconf_store,
                               noded_cert_file=noded_cert_file,
                               readd=readd,
-                              run_cmd_fn=run_cmd_fn)
+                              run_cmd_fn=run_cmd_fn,
+                              ssh_update_debug=ssh_update_debug,
+                              ssh_update_verbose=ssh_update_verbose)
 
 
 # Node info named tuple specifically for the use with RemoveNodeSshKeyBulk
@@ -1763,7 +1777,9 @@ def RemoveNodeSshKeyBulk(node_list,
                          ssconf_store=None,
                          noded_cert_file=pathutils.NODED_CERT_FILE,
                          readd=False,
-                         run_cmd_fn=ssh.RunSshCmdWithStdin):
+                         run_cmd_fn=ssh.RunSshCmdWithStdin,
+                         ssh_update_debug=False,
+                         ssh_update_verbose=False):
   """Removes the node's SSH keys from the key files and distributes those.
 
   Note that at least one of the flags C{from_authorized_keys},
@@ -1911,13 +1927,13 @@ def RemoveNodeSshKeyBulk(node_list,
             node_desc = "normal"
           logging.debug("Updating key setup of %s node %s.", node_desc, node)
           try:
+            backoff = 5  # seconds
             utils.RetryByNumberOfTimes(
-                constants.SSHS_MAX_RETRIES,
-                errors.SshUpdateError,
+                constants.SSHS_MAX_RETRIES, backoff, errors.SshUpdateError,
                 run_cmd_fn, cluster_name, node, pathutils.SSH_UPDATE,
                 ssh_port, pot_mc_data,
-                debug=False, verbose=False, use_cluster_key=False,
-                ask_key=False, strict_host_check=False)
+                debug=ssh_update_debug, verbose=ssh_update_verbose,
+                use_cluster_key=False, ask_key=False, strict_host_check=False)
           except errors.SshUpdateError as last_exception:
             error_msg = error_msg_final % (
                 node_info.name, node, last_exception)
@@ -1965,13 +1981,14 @@ def RemoveNodeSshKeyBulk(node_list,
       logging.debug("Updating SSH key setup of target node '%s'.",
                     node_info.name)
       try:
+        backoff = 5  # seconds
         utils.RetryByNumberOfTimes(
-            constants.SSHS_MAX_RETRIES,
+            constants.SSHS_MAX_RETRIES, backoff,
             errors.SshUpdateError,
             run_cmd_fn, cluster_name, node_info.name, pathutils.SSH_UPDATE,
             ssh_port, data,
-            debug=False, verbose=False, use_cluster_key=False,
-            ask_key=False, strict_host_check=False)
+            debug=ssh_update_debug, verbose=ssh_update_verbose,
+            use_cluster_key=False, ask_key=False, strict_host_check=False)
       except errors.SshUpdateError as last_exception:
         result_msgs.append(
             (node_info.name,
@@ -1984,18 +2001,52 @@ def RemoveNodeSshKeyBulk(node_list,
       ssh.RemovePublicKey(node_uuid, key_file=pub_key_file)
 
   return result_msgs
+# pylint: enable=R0913
+
+
+def RemoveSshKeyFromPublicKeyFile(node_name,
+                                  pub_key_file=pathutils.SSH_PUB_KEYS,
+                                  ssconf_store=None):
+  """Removes a SSH key from the master's public key file.
+
+  This is an operation that is only used to clean up after failed operations
+  (for example failed hooks before adding a node). To avoid abuse of this
+  function (and the matching RPC call), we add a safety check to make sure
+  that only stray keys can be removed that belong to nodes that are not
+  in the cluster (anymore).
+
+  @type node_name: string
+  @param node_name: the name of the node whose key is removed
+
+  """
+  if not ssconf_store:
+    ssconf_store = ssconf.SimpleStore()
+
+  node_list = ssconf_store.GetNodeList()
+
+  if node_name in node_list:
+    raise errors.SshUpdateError("Cannot remove key of node '%s',"
+                                " because it still belongs to the cluster."
+                                % node_name)
+
+  keys_by_name = ssh.QueryPubKeyFile([node_name], key_file=pub_key_file)
+  if not keys_by_name or node_name not in keys_by_name:
+    logging.info("The node '%s' whose key is supposed to be removed does not"
+                 " have an entry in the public key file. Hence, there is"
+                 " nothing left to do.", node_name)
 
+  ssh.RemovePublicKey(node_name, key_file=pub_key_file)
 
-def _GenerateNodeSshKey(node_uuid, node_name, ssh_port_map, ssh_key_type,
-                        ssh_key_bits, pub_key_file=pathutils.SSH_PUB_KEYS,
+
+def _GenerateNodeSshKey(node_name, ssh_port_map, ssh_key_type, ssh_key_bits,
                         ssconf_store=None,
                         noded_cert_file=pathutils.NODED_CERT_FILE,
                         run_cmd_fn=ssh.RunSshCmdWithStdin,
-                        suffix=""):
+                        suffix="",
+                        ssh_update_debug=False,
+                        ssh_update_verbose=False):
   """Generates the root SSH key pair on the node.
 
-  @type node_uuid: str
-  @param node_uuid: UUID of the node whose key is removed
   @type node_name: str
   @param node_name: name of the node whose key is remove
   @type ssh_port_map: dict of str to int
@@ -2009,12 +2060,6 @@ def _GenerateNodeSshKey(node_uuid, node_name, ssh_port_map, ssh_key_type,
   if not ssconf_store:
     ssconf_store = ssconf.SimpleStore()
 
-  keys_by_uuid = ssh.QueryPubKeyFile([node_uuid], key_file=pub_key_file)
-  if not keys_by_uuid or node_uuid not in keys_by_uuid:
-    raise errors.SshUpdateError("Node %s (UUID: %s) whose key is requested to"
-                                " be regenerated is not registered in the"
-                                " public keys file." % (node_name, node_uuid))
-
   data = {}
   _InitSshUpdateData(data, noded_cert_file, ssconf_store)
   cluster_name = data[constants.SSHS_CLUSTER_NAME]
@@ -2022,8 +2067,8 @@ def _GenerateNodeSshKey(node_uuid, node_name, ssh_port_map, ssh_key_type,
 
   run_cmd_fn(cluster_name, node_name, pathutils.SSH_UPDATE,
              ssh_port_map.get(node_name), data,
-             debug=False, verbose=False, use_cluster_key=False,
-             ask_key=False, strict_host_check=False)
+             debug=ssh_update_debug, verbose=ssh_update_verbose,
+             use_cluster_key=False, ask_key=False, strict_host_check=False)
 
 
 def _GetMasterNodeUUID(node_uuid_name_map, master_node_name):
@@ -2047,58 +2092,15 @@ def _GetOldMasterKeys(master_node_uuid, pub_key_file):
   return old_master_keys_by_uuid
 
 
-def _GetNewMasterKey(root_keyfiles, master_node_uuid):
-  new_master_keys = []
-  for (_, (_, public_key_file)) in root_keyfiles.items():
-    public_key_dir = os.path.dirname(public_key_file)
-    public_key_file_tmp_filename = \
-        os.path.splitext(os.path.basename(public_key_file))[0] \
-        + constants.SSHS_MASTER_SUFFIX + ".pub"
-    public_key_path_tmp = os.path.join(public_key_dir,
-                                       public_key_file_tmp_filename)
-    if os.path.exists(public_key_path_tmp):
-      # for some key types, there might not be any keys
-      key = utils.ReadFile(public_key_path_tmp)
-      new_master_keys.append(key)
-  if not new_master_keys:
-    raise errors.SshUpdateError("Cannot find any type of temporary SSH key.")
-  return {master_node_uuid: new_master_keys}
-
-
-def _ReplaceMasterKeyOnMaster(root_keyfiles):
-  number_of_moves = 0
-  for (_, (private_key_file, public_key_file)) in root_keyfiles.items():
-    key_dir = os.path.dirname(public_key_file)
-    private_key_file_tmp = \
-      os.path.basename(private_key_file) + constants.SSHS_MASTER_SUFFIX
-    public_key_file_tmp = private_key_file_tmp + ".pub"
-    private_key_path_tmp = os.path.join(key_dir,
-                                        private_key_file_tmp)
-    public_key_path_tmp = os.path.join(key_dir,
-                                       public_key_file_tmp)
-    if os.path.exists(public_key_file):
-      utils.CreateBackup(public_key_file)
-      utils.RemoveFile(public_key_file)
-    if os.path.exists(private_key_file):
-      utils.CreateBackup(private_key_file)
-      utils.RemoveFile(private_key_file)
-    if os.path.exists(public_key_path_tmp) and \
-        os.path.exists(private_key_path_tmp):
-      # for some key types, there might not be any keys
-      shutil.move(public_key_path_tmp, public_key_file)
-      shutil.move(private_key_path_tmp, private_key_file)
-      number_of_moves += 1
-  if not number_of_moves:
-    raise errors.SshUpdateError("Could not move at least one master SSH key.")
-
-
 def RenewSshKeys(node_uuids, node_names, master_candidate_uuids,
                  potential_master_candidates, old_key_type, new_key_type,
                  new_key_bits,
                  ganeti_pub_keys_file=pathutils.SSH_PUB_KEYS,
                  ssconf_store=None,
                  noded_cert_file=pathutils.NODED_CERT_FILE,
-                 run_cmd_fn=ssh.RunSshCmdWithStdin):
+                 run_cmd_fn=ssh.RunSshCmdWithStdin,
+                 ssh_update_debug=False,
+                 ssh_update_verbose=False):
   """Renews all SSH keys and updates authorized_keys and ganeti_pub_keys.
 
   @type node_uuids: list of str
@@ -2136,11 +2138,9 @@ def RenewSshKeys(node_uuids, node_names, master_candidate_uuids,
     raise errors.ProgrammerError("List of nodes UUIDs and node names"
                                  " does not match in length.")
 
-  (_, root_keyfiles) = \
-    ssh.GetAllUserFiles(constants.SSH_LOGIN_USER, mkdir=False, dircheck=False)
-  (_, old_pub_keyfile) = root_keyfiles[old_key_type]
-  (_, new_pub_keyfile) = root_keyfiles[new_key_type]
-  old_master_key = utils.ReadFile(old_pub_keyfile)
+  old_pub_keyfile = ssh.GetSshPubKeyFilename(old_key_type)
+  new_pub_keyfile = ssh.GetSshPubKeyFilename(new_key_type)
+  old_master_key = ssh.ReadLocalSshPubKeys([old_key_type])
 
   node_uuid_name_map = zip(node_uuids, node_names)
 
@@ -2171,20 +2171,13 @@ def RenewSshKeys(node_uuids, node_names, master_candidate_uuids,
     node_list.append((node_uuid, node_name, master_candidate,
                       potential_master_candidate))
 
-    keys_by_uuid = ssh.QueryPubKeyFile([node_uuid],
-                                       key_file=ganeti_pub_keys_file)
-    if not keys_by_uuid:
-      raise errors.SshUpdateError("No public key of node %s (UUID %s) found,"
-                                  " not generating a new key."
-                                  % (node_name, node_uuid))
-
     if master_candidate:
       logging.debug("Fetching old SSH key from node '%s'.", node_name)
-      old_pub_key = ssh.ReadRemoteSshPubKeys(old_pub_keyfile,
-                                             node_name, cluster_name,
-                                             ssh_port_map[node_name],
-                                             False, # ask_key
-                                             False) # key_check
+      old_pub_key = ssh.ReadRemoteSshPubKey(old_pub_keyfile,
+                                            node_name, cluster_name,
+                                            ssh_port_map[node_name],
+                                            False, # ask_key
+                                            False) # key_check
       if old_pub_key != old_master_key:
         # If we are already in a multi-key setup (that is past Ganeti 2.12),
         # we can safely remove the old key of the node. Otherwise, we cannot
@@ -2208,7 +2201,13 @@ def RenewSshKeys(node_uuids, node_names, master_candidate_uuids,
         node_info_to_remove,
         master_candidate_uuids,
         potential_master_candidates,
-        master_uuid=master_node_uuid)
+        master_uuid=master_node_uuid,
+        pub_key_file=ganeti_pub_keys_file,
+        ssconf_store=ssconf_store,
+        noded_cert_file=noded_cert_file,
+        run_cmd_fn=run_cmd_fn,
+        ssh_update_debug=ssh_update_debug,
+        ssh_update_verbose=ssh_update_verbose)
     if node_errors:
       all_node_errors = all_node_errors + node_errors
 
@@ -2216,19 +2215,20 @@ def RenewSshKeys(node_uuids, node_names, master_candidate_uuids,
       in node_list:
 
     logging.debug("Generating new SSH key for node '%s'.", node_name)
-    _GenerateNodeSshKey(node_uuid, node_name, ssh_port_map, new_key_type,
-                        new_key_bits, pub_key_file=ganeti_pub_keys_file,
+    _GenerateNodeSshKey(node_name, ssh_port_map, new_key_type, new_key_bits,
                         ssconf_store=ssconf_store,
                         noded_cert_file=noded_cert_file,
-                        run_cmd_fn=run_cmd_fn)
+                        run_cmd_fn=run_cmd_fn,
+                        ssh_update_verbose=ssh_update_verbose,
+                        ssh_update_debug=ssh_update_debug)
 
     try:
       logging.debug("Fetching newly created SSH key from node '%s'.", node_name)
-      pub_key = ssh.ReadRemoteSshPubKeys(new_pub_keyfile,
-                                         node_name, cluster_name,
-                                         ssh_port_map[node_name],
-                                         False, # ask_key
-                                         False) # key_check
+      pub_key = ssh.ReadRemoteSshPubKey(new_pub_keyfile,
+                                        node_name, cluster_name,
+                                        ssh_port_map[node_name],
+                                        False, # ask_key
+                                        False) # key_check
     except:
       raise errors.SshUpdateError("Could not fetch key of node %s"
                                   " (UUID %s)" % (node_name, node_uuid))
@@ -2248,7 +2248,9 @@ def RenewSshKeys(node_uuids, node_names, master_candidate_uuids,
       node_keys_to_add, potential_master_candidates,
       pub_key_file=ganeti_pub_keys_file, ssconf_store=ssconf_store,
       noded_cert_file=noded_cert_file,
-      run_cmd_fn=run_cmd_fn)
+      run_cmd_fn=run_cmd_fn,
+      ssh_update_debug=ssh_update_debug,
+      ssh_update_verbose=ssh_update_verbose)
   if node_errors:
     all_node_errors = all_node_errors + node_errors
 
@@ -2260,19 +2262,21 @@ def RenewSshKeys(node_uuids, node_names, master_candidate_uuids,
 
   # Generate a new master key with a suffix, don't touch the old one for now
   logging.debug("Generate new ssh key of master.")
-  _GenerateNodeSshKey(master_node_uuid, master_node_name, ssh_port_map,
+  _GenerateNodeSshKey(master_node_name, ssh_port_map,
                       new_key_type, new_key_bits,
-                      pub_key_file=ganeti_pub_keys_file,
                       ssconf_store=ssconf_store,
                       noded_cert_file=noded_cert_file,
                       run_cmd_fn=run_cmd_fn,
-                      suffix=constants.SSHS_MASTER_SUFFIX)
+                      suffix=constants.SSHS_MASTER_SUFFIX,
+                      ssh_update_debug=ssh_update_debug,
+                      ssh_update_verbose=ssh_update_verbose)
   # Read newly created master key
-  new_master_key_dict = _GetNewMasterKey(root_keyfiles, master_node_uuid)
+  new_master_keys = ssh.ReadLocalSshPubKeys(
+      [new_key_type], suffix=constants.SSHS_MASTER_SUFFIX)
 
   # Replace master key in the master nodes' public key file
   ssh.RemovePublicKey(master_node_uuid, key_file=ganeti_pub_keys_file)
-  for pub_key in new_master_key_dict[master_node_uuid]:
+  for pub_key in new_master_keys:
     ssh.AddPublicKey(master_node_uuid, pub_key, key_file=ganeti_pub_keys_file)
 
   # Add new master key to all node's public and authorized keys
@@ -2282,12 +2286,15 @@ def RenewSshKeys(node_uuids, node_names, master_candidate_uuids,
       to_authorized_keys=True, to_public_keys=True,
       get_public_keys=False, pub_key_file=ganeti_pub_keys_file,
       ssconf_store=ssconf_store, noded_cert_file=noded_cert_file,
-      run_cmd_fn=run_cmd_fn)
+      run_cmd_fn=run_cmd_fn,
+      ssh_update_debug=ssh_update_debug,
+      ssh_update_verbose=ssh_update_verbose)
   if node_errors:
     all_node_errors = all_node_errors + node_errors
 
   # Remove the old key file and rename the new key to the non-temporary filename
-  _ReplaceMasterKeyOnMaster(root_keyfiles)
+  ssh.ReplaceSshKeys(new_key_type, new_key_type,
+                     src_key_suffix=constants.SSHS_MASTER_SUFFIX)
 
   # Remove old key from authorized keys
   (auth_key_file, _) = \
@@ -2302,7 +2309,13 @@ def RenewSshKeys(node_uuids, node_names, master_candidate_uuids,
       potential_master_candidates,
       keys_to_remove=old_master_keys_by_uuid, from_authorized_keys=True,
       from_public_keys=False, clear_authorized_keys=False,
-      clear_public_keys=False)
+      clear_public_keys=False,
+      pub_key_file=ganeti_pub_keys_file,
+      ssconf_store=ssconf_store,
+      noded_cert_file=noded_cert_file,
+      run_cmd_fn=run_cmd_fn,
+      ssh_update_debug=ssh_update_debug,
+      ssh_update_verbose=ssh_update_verbose)
   if node_errors:
     all_node_errors = all_node_errors + node_errors
 
@@ -5772,18 +5785,25 @@ def _PrepareRestrictedCmd(path, cmd,
   return _verify_cmd(path, cmd)
 
 
-def RunRestrictedCmd(cmd,
-                     _lock_timeout=_RCMD_LOCK_TIMEOUT,
-                     _lock_file=pathutils.RESTRICTED_COMMANDS_LOCK_FILE,
-                     _path=pathutils.RESTRICTED_COMMANDS_DIR,
-                     _sleep_fn=time.sleep,
-                     _prepare_fn=_PrepareRestrictedCmd,
-                     _runcmd_fn=utils.RunCmd,
-                     _enabled=constants.ENABLE_RESTRICTED_COMMANDS):
-  """Executes a restricted command after performing strict tests.
+def RunConstrainedCmd(cmd,
+                      lock_file,
+                      path,
+                      inp=None,
+                      _lock_timeout=_RCMD_LOCK_TIMEOUT,
+                      _sleep_fn=time.sleep,
+                      _prepare_fn=_PrepareRestrictedCmd,
+                      _runcmd_fn=utils.RunCmd,
+                      _enabled=constants.ENABLE_RESTRICTED_COMMANDS):
+  """Executes a command after performing strict tests.
 
   @type cmd: string
   @param cmd: Command name
+  @type lock_file: string
+  @param lock_file: path to the lock file
+  @type path: string
+  @param path: path to the directory in which the command is present
+  @type inp: string
+  @param inp: Input to be passed to the command
   @rtype: string
   @return: Command output
   @raise RPCFail: In case of an error
@@ -5798,14 +5818,24 @@ def RunRestrictedCmd(cmd,
   try:
     cmdresult = None
     try:
-      lock = utils.FileLock.Open(_lock_file)
+      lock = utils.FileLock.Open(lock_file)
       lock.Exclusive(blocking=True, timeout=_lock_timeout)
 
-      (status, value) = _prepare_fn(_path, cmd)
+      (status, value) = _prepare_fn(path, cmd)
 
       if status:
+        if inp:
+          input_fd = tempfile.TemporaryFile()
+          input_fd.write(inp)
+          input_fd.flush()
+          input_fd.seek(0)
+        else:
+          input_fd = None
         cmdresult = _runcmd_fn([value], env={}, reset_env=True,
-                               postfork_fn=lambda _: lock.Unlock())
+                               postfork_fn=lambda _: lock.Unlock(),
+                               input_fd=input_fd)
+        if input_fd:
+          input_fd.close()
       else:
         logging.error(value)
     except Exception: # pylint: disable=W0703