nixos-rebuild-ng: improve IPv6 support (#469709)
This commit is contained in:
@@ -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",
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user