diff --git a/pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/process.py b/pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/process.py index 1f57111a72ef..902797189c69 100644 --- a/pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/process.py +++ b/pkgs/by-name/ni/nixos-rebuild-ng/src/nixos_rebuild/process.py @@ -7,6 +7,7 @@ import shlex import subprocess from collections.abc import Sequence from dataclasses import dataclass +from ipaddress import AddressValueError, IPv6Address from typing import Final, Self, TypedDict, Unpack from . import tmpdir @@ -63,6 +64,23 @@ class Remote: "'--ask-sudo-password' option instead" ) + def ssh_host(self) -> str: + """Fix up host string for SSH. + + It returns an SSH-compatible host string if the host is using an + IPv6 address, otherwise it returns the passed host string unmodified. + """ + try: + host_split = self.host.split("@") + host_fixed = host_split[-1].strip("[]").replace("%25", "%") + IPv6Address(host_fixed) + if len(host_split) > 1: + return host_split[0] + "@" + host_fixed + else: + return host_fixed + except AddressValueError: + return self.host + # Not exhaustive, but we can always extend it later. class RunKwargs(TypedDict, total=False): @@ -112,7 +130,7 @@ def run_wrapper( "ssh", *remote.opts, *SSH_DEFAULT_OPTS, - remote.host, + remote.ssh_host(), "--", # SSH will join the parameters here and pass it to the shell, so we # need to quote it to avoid issues. @@ -187,7 +205,7 @@ def _kill_long_running_ssh_process(args: Args, remote: Remote) -> None: "ssh", *remote.opts, *SSH_DEFAULT_OPTS, - remote.host, + remote.ssh_host(), "--", "pkill", "--signal", diff --git a/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_process.py b/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_process.py index 84eba633b21b..84b5c084bf38 100644 --- a/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_process.py +++ b/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_process.py @@ -140,3 +140,23 @@ def test_remote_from_name(monkeypatch: MonkeyPatch) -> None: opts=["-f", "foo", "-b", "bar", "-t"], sudo_password="password", ) + + +def test_ssh_host() -> None: + remotes = { + "user@[fe80::1%25eth0]": "user@fe80::1%eth0", + "[fe80::c98b%25enp4s0]": "fe80::c98b%enp4s0", + "user@[2001::5fce:a:198]": "user@2001::5fce:a:198", + "[2001::dead:beef]": "2001::dead:beef", + "root@192.168.178.1": "root@192.168.178.1", + "192.168.178.1": "192.168.178.1", + "user@localhost": "user@localhost", + "localhost": "localhost", + "user@example.org": "user@example.org", + "example.org": "example.org", + } + + for host_input, expected in remotes.items(): + remote = m.Remote.from_arg(host_input, None, False) + assert remote is not None + assert remote.ssh_host() == expected