Merge pull request #1464 from hercules-ci/internal-known-hosts-handling

Always provide our known_hosts in addition to user known_hosts

nixops/backends/__init__.py

@@ -6,6 +6,7 @@
  Mapping,
  Match,
  Any,
+ Dict,
  List,
  Optional,
  Union,
@@ -453,15 +454,57 @@ def get_keys(self):
  return self.keys
 
  def get_ssh_name(self) -> str:
+ """
+ In ssh terminology, this is the "Host", which part of the "destination"
+ but not necessarily the same as the "Hostname".
+ The ssh config file can set Hostname for specific Hosts, effectively
+ rewriting the destination into the final hostname or ip.
+ """
  assert False
 
- def get_ssh_flags(self, scp: bool = False) -> List[str]:
- if scp:
- return ["-P", str(self.ssh_port)] if self.ssh_port is not None else []
+ def get_ssh_host_keys(self) -> Optional[str]:
+ """
+ Return the public host key in known_hosts format or None if not known.
+ """
+ return None
+
+ def _get_ssh_ambient_options(self) -> Dict[str, str]:
+ with subprocess.Popen(
+ ["ssh", "-G", self.get_ssh_name()], stdout=subprocess.PIPE, text=True
+ ) as proc:
+ assert proc.stdout is not None
+ opts: Dict[str, str] = {}
+ for line in proc.stdout:
+ s = line.rstrip("\r\n").split(" ", 1)
+ if len(s) == 2:
+ opts[s[0].lower()] = s[1]
+
+ return opts
+
+ def get_known_hosts_file(self, *args, **kwargs) -> Optional[str]:
+ k = self.get_ssh_host_keys()
+ if k is not None:
+ return self.write_ssh_known_hosts(k)
  else:
- return list(self.ssh_options) + (
- ["-p", str(self.ssh_port)] if self.ssh_port is not None else []
- )
+ return None
+
+ def get_ssh_flags(self, scp: bool = False) -> List[str]:
+ flags: List[str] = []
+
+ if self.ssh_port is not None:
+ flags = flags + ["-o", "Port=" + str(self.ssh_port)]
+
+ # We add our own public host key (if known) to GlobalKnownHostsFile.
+ # This way we don't override keys in ~/.ssh/known_hosts that some users
+ # may rely on. We don't set UserKnownHostsFile, because that file is
+ # supposed to be editable, whereas ours is generated and shouldn't be
+ # edited.
+ known_hosts_file = self.get_known_hosts_file()
+
+ if known_hosts_file is not None:
+ flags = flags + ["-o", "GlobalKnownHostsFile=" + known_hosts_file]
+
+ return flags
 
  def get_ssh_password(self):
  return None
@@ -505,6 +548,36 @@ def write_ssh_private_key(self, private_key: str) -> str:
  def get_ssh_private_key_file(self) -> Optional[str]:
  return None
 
+ def write_ssh_known_hosts(self, known_hosts: str) -> str:
+ """
+ Write a temporary file for a known_hosts file containing this machine's
+ host public key and the global known hosts entries.
+ """
+
+ # We copy the global known hosts files, because we can't pass multiple
+ # file names through NIX_SSHOPTS, because spaces are interpreted as
+ # option separators there.
+ ambientGlobalFilesStr = self._get_ssh_ambient_options().get(
+ "globalknownhostsfile"
+ )
+ if ambientGlobalFilesStr is None:
+ ambientGlobalFiles = []
+ else:
+ ambientGlobalFiles = ambientGlobalFilesStr.split()
+
+ for globalFile in ambientGlobalFiles:
+ if os.path.exists(globalFile):
+ with open(globalFile) as f:
+ contents = f.read()
+ known_hosts = (
+ known_hosts + f"\n\n# entries from {globalFile}\n" + contents
+ )
+
+ file = "{0}/known_host_nixops-{1}".format(self.depl.tempdir, self.name)
+ with os.fdopen(os.open(file, os.O_CREAT | os.O_WRONLY, 0o600), "w") as f:
+ f.write(known_hosts)
+ return file
+
  def _logged_exec(self, command: List[str], **kwargs) -> Union[str, int]:
  return nixops.util.logged_exec(command, self.logger, **kwargs)