yadunut/clan-core
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge pull request 'networking_3' (#4507) from networking_3 into main

Browse Source 
Reviewed-on: https://git.clan.lol/clan/clan-core/pulls/4507
This commit is contained in:
DavHau
2025-08-13 05:20:03 +00:00
parent 4f60345ba7 068b5d4c1e
commit 4f78a8ff94
20 changed files with 545 additions and 368 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pkgs/clan-cli/clan_cli/facts/upload.py
View File

@@ -6,9 +6,9 @@ from tempfile import TemporaryDirectory
from clan_lib.flake import require_flake
from clan_lib.machines.machines import Machine
from clan_lib.ssh.host import Host
from clan_lib.ssh.upload import upload
from clan_cli.completions import add_dynamic_completer, complete_machines
from clan_cli.ssh.upload import upload
log = logging.getLogger(__name__)

71
pkgs/clan-cli/clan_cli/machines/install.py
View File

@@ -1,6 +1,8 @@
import argparse
import json
import logging
import sys
from contextlib import ExitStack
from pathlib import Path
from typing import get_args
@@ -8,6 +10,7 @@ from clan_lib.errors import ClanError
from clan_lib.flake import require_flake
from clan_lib.machines.install import BuildOn, InstallOptions, run_machine_install
from clan_lib.machines.machines import Machine
from clan_lib.network.qr_code import read_qr_image, read_qr_json
from clan_lib.ssh.host_key import HostKeyCheck
from clan_lib.ssh.remote import Remote
@@ -17,7 +20,6 @@ from clan_cli.completions import (
complete_target_host,
)
from clan_cli.machines.hardware import HardwareConfig
from clan_cli.ssh.deploy_info import DeployInfo, find_reachable_host, ssh_command_parse
log = logging.getLogger(__name__)
@@ -27,39 +29,33 @@ def install_command(args: argparse.Namespace) -> None:
flake = require_flake(args.flake)
# Only if the caller did not specify a target_host via args.target_host
# Find a suitable target_host that is reachable
target_host_str = args.target_host
deploy_info: DeployInfo | None = (
ssh_command_parse(args) if target_host_str is None else None
with ExitStack() as stack:
remote: Remote
if args.target_host:
# TODO add network support here with either --network or some url magic
remote = Remote.from_ssh_uri(
machine_name=args.machine, address=args.target_host
)
use_tor = False
if deploy_info:
host = find_reachable_host(deploy_info)
if host is None or host.socks_port:
use_tor = True
target_host_str = deploy_info.tor.target
elif args.png:
data = read_qr_image(Path(args.png))
qr_code = read_qr_json(data, args.flake)
remote = stack.enter_context(qr_code.get_best_remote())
elif args.json:
json_file = Path(args.json)
if json_file.is_file():
data = json.loads(json_file.read_text())
else:
target_host_str = host.target
data = json.loads(args.json)
if args.password:
password = args.password
elif deploy_info and deploy_info.addrs[0].password:
password = deploy_info.addrs[0].password
qr_code = read_qr_json(data, args.flake)
remote = stack.enter_context(qr_code.get_best_remote())
else:
password = None
msg = "No --target-host, --json or --png data provided"
raise ClanError(msg)
machine = Machine(name=args.machine, flake=flake)
host_key_check = args.host_key_check
if target_host_str is not None:
target_host = Remote.from_ssh_uri(
machine_name=machine.name, address=target_host_str
).override(host_key_check=host_key_check)
else:
target_host = machine.target_host().override(host_key_check=host_key_check)
if args.identity_file:
target_host = target_host.override(private_key=args.identity_file)
if args.host_key_check:
remote.override(host_key_check=args.host_key_check)
if machine._class_ == "darwin":
msg = "Installing macOS machines is not yet supported"
@@ -68,7 +64,7 @@ def install_command(args: argparse.Namespace) -> None:
if not args.yes:
while True:
ask = (
input(f"Install {args.machine} to {target_host.target}? [y/N] ")
input(f"Install {args.machine} to {remote.target}? [y/N] ")
.strip()
.lower()
)
@@ -76,18 +72,15 @@ def install_command(args: argparse.Namespace) -> None:
break
if ask == "n" or ask == "":
return None
print(f"Invalid input '{ask}'. Please enter 'y' for yes or 'n' for no.")
print(
f"Invalid input '{ask}'. Please enter 'y' for yes or 'n' for no."
)
if args.identity_file:
target_host = target_host.override(private_key=args.identity_file)
remote = remote.override(private_key=args.identity_file)
if password:
target_host = target_host.override(password=password)
if use_tor:
target_host = target_host.override(
socks_port=9050, socks_wrapper=["torify"]
)
if args.password:
remote = remote.override(password=args.password)
return run_machine_install(
InstallOptions(
@@ -99,7 +92,7 @@ def install_command(args: argparse.Namespace) -> None:
build_on=args.build_on if args.build_on is not None else None,
update_hardware_config=HardwareConfig(args.update_hardware_config),
),
target_host=target_host,
target_host=remote,
)
except KeyboardInterrupt:
log.warning("Interrupted by user")

3
pkgs/clan-cli/clan_cli/network/overview.py
View File

@@ -16,6 +16,9 @@ def overview_command(args: argparse.Namespace) -> None:
for peer_name, peer in network["peers"].items():
print(f"\t{peer_name}: {'[OFFLINE]' if not peer else f'[{peer}]'}")
if not overview:
print("No networks found.")
def register_overview_parser(parser: argparse.ArgumentParser) -> None:
parser.set_defaults(func=overview_command)

4
pkgs/clan-cli/clan_cli/network/ping.py
View File

@@ -16,8 +16,8 @@ def ping_command(args: argparse.Namespace) -> None:
networks = networks_from_flake(flake)
if not networks:
print("No networks found in the flake")
print("No networks found")
return
# If network is specified, only check that network
if network_name:
networks_to_check = [(network_name, networks[network_name])]

212
pkgs/clan-cli/clan_cli/ssh/deploy_info.py
View File

@@ -1,17 +1,14 @@
import argparse
import contextlib
import json
import logging
import textwrap
from dataclasses import dataclass
from contextlib import ExitStack
from pathlib import Path
from typing import Any, get_args
from typing import get_args
from clan_lib.cmd import run
from clan_lib.errors import ClanError
from clan_lib.machines.machines import Machine
from clan_lib.network.tor.lib import spawn_tor
from clan_lib.nix import nix_shell
from clan_lib.network.network import get_best_remote
from clan_lib.network.qr_code import read_qr_image, read_qr_json
from clan_lib.ssh.remote import HostKeyCheck, Remote
from clan_cli.completions import (
@@ -22,180 +19,57 @@ from clan_cli.completions import (
log = logging.getLogger(__name__)
@dataclass
class DeployInfo:
addrs: list[Remote]
def get_tor_remote(remotes: list[Remote]) -> Remote:
"""Get the Remote configured for SOCKS5 proxy (Tor)."""
tor_remotes = [r for r in remotes if r.socks_port]
@property
def tor(self) -> Remote:
"""Return a list of Remote objects that are configured for SOCKS5 proxy."""
addrs = [addr for addr in self.addrs if addr.socks_port]
if not addrs:
if not tor_remotes:
msg = "No socks5 proxy address provided, please provide a socks5 proxy address."
raise ClanError(msg)
if len(addrs) > 1:
if len(tor_remotes) > 1:
msg = "Multiple socks5 proxy addresses provided, expected only one."
raise ClanError(msg)
return addrs[0]
def overwrite_remotes(
self,
host_key_check: HostKeyCheck | None = None,
private_key: Path | None = None,
ssh_options: dict[str, str] | None = None,
) -> "DeployInfo":
"""Return a new DeployInfo with all Remotes overridden with the given host_key_check."""
return DeployInfo(
addrs=[
addr.override(
host_key_check=host_key_check,
private_key=private_key,
ssh_options=ssh_options,
)
for addr in self.addrs
]
)
@staticmethod
def from_json(data: dict[str, Any], host_key_check: HostKeyCheck) -> "DeployInfo":
addrs = []
password = data.get("pass")
for addr in data.get("addrs", []):
if isinstance(addr, str):
remote = Remote.from_ssh_uri(
machine_name="clan-installer",
address=addr,
).override(host_key_check=host_key_check, password=password)
addrs.append(remote)
else:
msg = f"Invalid address format: {addr}"
raise ClanError(msg)
if tor_addr := data.get("tor"):
remote = Remote.from_ssh_uri(
machine_name="clan-installer",
address=tor_addr,
).override(
host_key_check=host_key_check,
socks_port=9050,
socks_wrapper=["torify"],
password=password,
)
addrs.append(remote)
return DeployInfo(addrs=addrs)
@staticmethod
def from_qr_code(picture_file: Path, host_key_check: HostKeyCheck) -> "DeployInfo":
cmd = nix_shell(
["zbar"],
[
"zbarimg",
"--quiet",
"--raw",
str(picture_file),
],
)
res = run(cmd)
data = res.stdout.strip()
return DeployInfo.from_json(json.loads(data), host_key_check=host_key_check)
def find_reachable_host(deploy_info: DeployInfo) -> Remote | None:
# If we only have one address, we have no choice but to use it.
if len(deploy_info.addrs) == 1:
return deploy_info.addrs[0]
for addr in deploy_info.addrs:
with contextlib.suppress(ClanError):
addr.check_machine_ssh_reachable()
return addr
return None
def ssh_shell_from_deploy(
deploy_info: DeployInfo, command: list[str] | None = None
) -> None:
if command and len(command) == 1 and command[0].count(" ") > 0:
msg = (
textwrap.dedent("""
It looks like you quoted the remote command.
The first argument should be the command to run, not a quoted string.
""")
.lstrip("\n")
.rstrip("\n")
)
raise ClanError(msg)
if host := find_reachable_host(deploy_info):
host.interactive_ssh(command)
return
log.info("Could not reach host via clearnet 'addrs'")
log.info(f"Trying to reach host via tor '{deploy_info}'")
tor_addrs = [addr for addr in deploy_info.addrs if addr.socks_port]
if not tor_addrs:
msg = "No tor address provided, please provide a tor address."
raise ClanError(msg)
with spawn_tor():
for tor_addr in tor_addrs:
log.info(f"Trying to reach host via tor address: {tor_addr}")
with contextlib.suppress(ClanError):
tor_addr.check_machine_ssh_reachable()
log.info(
"Host reachable via tor address, starting interactive ssh session."
)
tor_addr.interactive_ssh(command)
return
log.error("Could not reach host via tor address.")
def ssh_command_parse(args: argparse.Namespace) -> DeployInfo | None:
host_key_check = args.host_key_check
deploy = None
if args.json:
json_file = Path(args.json)
if json_file.is_file():
data = json.loads(json_file.read_text())
return DeployInfo.from_json(data, host_key_check)
data = json.loads(args.json)
deploy = DeployInfo.from_json(data, host_key_check)
elif args.png:
deploy = DeployInfo.from_qr_code(Path(args.png), host_key_check)
elif hasattr(args, "machine") and args.machine:
machine = Machine(args.machine, args.flake)
target = machine.target_host().override(
command_prefix=machine.name, host_key_check=host_key_check
)
deploy = DeployInfo(addrs=[target])
else:
return None
ssh_options = None
if hasattr(args, "ssh_option") and args.ssh_option:
for name, value in args.ssh_option:
ssh_options = {}
ssh_options[name] = value
deploy = deploy.overwrite_remotes(ssh_options=ssh_options)
return deploy
return tor_remotes[0]
def ssh_command(args: argparse.Namespace) -> None:
deploy_info = ssh_command_parse(args)
if not deploy_info:
with ExitStack() as stack:
remote: Remote
if hasattr(args, "machine") and args.machine:
machine = Machine(args.machine, args.flake)
remote = stack.enter_context(get_best_remote(machine))
elif args.png:
data = read_qr_image(Path(args.png))
qr_code = read_qr_json(data, args.flake)
remote = stack.enter_context(qr_code.get_best_remote())
elif args.json:
json_file = Path(args.json)
if json_file.is_file():
data = json.loads(json_file.read_text())
else:
data = json.loads(args.json)
qr_code = read_qr_json(data, args.flake)
remote = stack.enter_context(qr_code.get_best_remote())
else:
msg = "No MACHINE, --json or --png data provided"
raise ClanError(msg)
ssh_shell_from_deploy(deploy_info, args.remote_command)
# Convert ssh_option list to dictionary
ssh_options = {}
if args.ssh_option:
for name, value in args.ssh_option:
ssh_options[name] = value
remote = remote.override(
host_key_check=args.host_key_check, ssh_options=ssh_options
)
if args.remote_command:
remote.interactive_ssh(args.remote_command)
else:
remote.interactive_ssh()
def register_parser(parser: argparse.ArgumentParser) -> None:

120
pkgs/clan-cli/clan_cli/ssh/test_deploy_info.py
View File

@@ -3,15 +3,17 @@ from pathlib import Path
import pytest
from clan_lib.cmd import RunOpts, run
from clan_lib.flake import Flake
from clan_lib.network.qr_code import read_qr_image, read_qr_json
from clan_lib.nix import nix_shell
from clan_lib.ssh.remote import Remote
from clan_cli.ssh.deploy_info import DeployInfo, find_reachable_host
from clan_cli.tests.fixtures_flakes import ClanFlake
from clan_cli.tests.helpers import cli
def test_qrcode_scan(temp_dir: Path) -> None:
@pytest.mark.with_core
def test_qrcode_scan(temp_dir: Path, flake: ClanFlake) -> None:
data = '{"pass":"scabbed-defender-headlock","tor":"qjeerm4r6t55hcfum4pinnvscn5njlw2g3k7ilqfuu7cdt3ahaxhsbid.onion","addrs":["192.168.122.86"]}'
img_path = temp_dir / "qrcode.png"
cmd = nix_shell(
@@ -25,63 +27,93 @@ def test_qrcode_scan(temp_dir: Path) -> None:
run(cmd, RunOpts(input=data.encode()))
# Call the qrcode_scan function
deploy_info = DeployInfo.from_qr_code(img_path, "none")
json_data = read_qr_image(img_path)
qr_code = read_qr_json(json_data, Flake(str(flake.path)))
host = deploy_info.addrs[0]
assert host.address == "192.168.122.86"
assert host.user == "root"
assert host.password == "scabbed-defender-headlock"
# Check addresses
addresses = qr_code.addresses
assert len(addresses) >= 2 # At least direct and tor
tor_host = deploy_info.addrs[1]
# Find direct connection
direct_remote = None
for addr in addresses:
if addr.network.module_name == "clan_lib.network.direct":
direct_remote = addr.remote
break
assert direct_remote is not None
assert direct_remote.address == "192.168.122.86"
assert direct_remote.user == "root"
assert direct_remote.password == "scabbed-defender-headlock"
# Find tor connection
tor_remote = None
for addr in addresses:
if addr.network.module_name == "clan_lib.network.tor":
tor_remote = addr.remote
break
assert tor_remote is not None
assert (
tor_host.address
== "qjeerm4r6t55hcfum4pinnvscn5njlw2g3k7ilqfuu7cdt3ahaxhsbid.onion"
)
assert tor_host.socks_port == 9050
assert tor_host.password == "scabbed-defender-headlock"
assert tor_host.user == "root"
assert (
tor_host.address
tor_remote.address
== "qjeerm4r6t55hcfum4pinnvscn5njlw2g3k7ilqfuu7cdt3ahaxhsbid.onion"
)
assert tor_remote.socks_port == 9050
assert tor_remote.password == "scabbed-defender-headlock"
assert tor_remote.user == "root"
def test_from_json() -> None:
def test_from_json(temp_dir: Path) -> None:
data = '{"pass":"scabbed-defender-headlock","tor":"qjeerm4r6t55hcfum4pinnvscn5njlw2g3k7ilqfuu7cdt3ahaxhsbid.onion","addrs":["192.168.122.86"]}'
deploy_info = DeployInfo.from_json(json.loads(data), "none")
flake = Flake(str(temp_dir))
qr_code = read_qr_json(json.loads(data), flake)
host = deploy_info.addrs[0]
assert host.password == "scabbed-defender-headlock"
assert host.address == "192.168.122.86"
# Check addresses
addresses = qr_code.addresses
assert len(addresses) >= 2 # At least direct and tor
tor_host = deploy_info.addrs[1]
# Find direct connection
direct_remote = None
for addr in addresses:
if addr.network.module_name == "clan_lib.network.direct":
direct_remote = addr.remote
break
assert direct_remote is not None
assert direct_remote.password == "scabbed-defender-headlock"
assert direct_remote.address == "192.168.122.86"
# Find tor connection
tor_remote = None
for addr in addresses:
if addr.network.module_name == "clan_lib.network.tor":
tor_remote = addr.remote
break
assert tor_remote is not None
assert (
tor_host.address
== "qjeerm4r6t55hcfum4pinnvscn5njlw2g3k7ilqfuu7cdt3ahaxhsbid.onion"
)
assert tor_host.socks_port == 9050
assert tor_host.password == "scabbed-defender-headlock"
assert tor_host.user == "root"
assert (
tor_host.address
tor_remote.address
== "qjeerm4r6t55hcfum4pinnvscn5njlw2g3k7ilqfuu7cdt3ahaxhsbid.onion"
)
assert tor_remote.socks_port == 9050
assert tor_remote.password == "scabbed-defender-headlock"
assert tor_remote.user == "root"
@pytest.mark.with_core
def test_find_reachable_host(hosts: list[Remote]) -> None:
host = hosts[0]
uris = ["172.19.1.2", host.ssh_url()]
remotes = [Remote.from_ssh_uri(machine_name="some", address=uri) for uri in uris]
deploy_info = DeployInfo(addrs=remotes)
assert deploy_info.addrs[0].address == "172.19.1.2"
remote = find_reachable_host(deploy_info=deploy_info)
assert remote is not None
assert remote.ssh_url() == host.ssh_url()
# TODO: This test needs to be updated to use get_best_remote from clan_lib.network.network
# @pytest.mark.with_core
# def test_find_reachable_host(hosts: list[Remote]) -> None:
# host = hosts[0]
#
# uris = ["172.19.1.2", host.ssh_url()]
# remotes = [Remote.from_ssh_uri(machine_name="some", address=uri) for uri in uris]
#
# assert remotes[0].address == "172.19.1.2"
#
# remote = find_reachable_host(remotes=remotes)
#
# assert remote is not None
# assert remote.ssh_url() == host.ssh_url()
@pytest.mark.with_core

2
pkgs/clan-cli/clan_cli/tests/test_upload_single_file.py
View File

@@ -1,8 +1,8 @@
from pathlib import Path
import pytest
from clan_cli.ssh.upload import upload
from clan_lib.ssh.remote import Remote
from clan_lib.ssh.upload import upload
@pytest.mark.with_core

2
pkgs/clan-cli/clan_cli/vars/secret_modules/password_store.py
View File

@@ -6,11 +6,11 @@ from collections.abc import Iterable
from pathlib import Path
from tempfile import TemporaryDirectory
from clan_cli.ssh.upload import upload
from clan_cli.vars._types import StoreBase
from clan_cli.vars.generate import Generator, Var
from clan_lib.flake import Flake
from clan_lib.ssh.host import Host
from clan_lib.ssh.upload import upload
log = logging.getLogger(__name__)

2
pkgs/clan-cli/clan_cli/vars/secret_modules/sops.py
View File

@@ -22,13 +22,13 @@ from clan_cli.secrets.secrets import (
has_secret,
)
from clan_cli.secrets.sops import load_age_plugins
from clan_cli.ssh.upload import upload
from clan_cli.vars._types import StoreBase
from clan_cli.vars.generate import Generator
from clan_cli.vars.var import Var
from clan_lib.errors import ClanError
from clan_lib.flake import Flake
from clan_lib.ssh.host import Host
from clan_lib.ssh.upload import upload
@dataclass

6
pkgs/clan-cli/clan_lib/machines/install.py
View File

@@ -187,11 +187,13 @@ def run_machine_install(opts: InstallOptions, target_host: Remote) -> None:
cmd.append(target_host.target)
if target_host.socks_port:
# nix copy does not support socks5 proxy, use wrapper command
wrapper_cmd = target_host.socks_wrapper or ["torify"]
wrapper = target_host.socks_wrapper
wrapper_cmd = wrapper.cmd if wrapper else []
wrapper_packages = wrapper.packages if wrapper else []
cmd = nix_shell(
[
"nixos-anywhere",
*wrapper_cmd,
*wrapper_packages,
],
[*wrapper_cmd, *cmd],
)

14
pkgs/clan-cli/clan_lib/machines/machines.py
View File

@@ -10,7 +10,6 @@ from clan_cli.facts import secret_modules as facts_secret_modules
from clan_cli.vars._types import StoreBase
from clan_lib.api import API
from clan_lib.errors import ClanError
from clan_lib.flake import ClanSelectError, Flake
from clan_lib.nix_models.clan import InventoryMachine
from clan_lib.ssh.remote import Remote
@@ -125,15 +124,10 @@ class Machine:
return self.flake.path
def target_host(self) -> Remote:
remote = get_machine_host(self.name, self.flake, field="targetHost")
if remote is None:
msg = f"'targetHost' is not set for machine '{self.name}'"
raise ClanError(
msg,
description="See https://docs.clan.lol/guides/getting-started/update/#setting-the-target-host for more information.",
)
data = remote.data
return data
from clan_lib.network.network import get_best_remote
with get_best_remote(self) as remote:
return remote
def build_host(self) -> Remote | None:
"""

6
pkgs/clan-cli/clan_lib/network/direct.py
View File

@@ -19,12 +19,10 @@ class NetworkTechnology(NetworkTechnologyBase):
"""Direct connections are always 'running' as they don't require a daemon"""
return True
def ping(self, peer: Peer) -> None | float:
def ping(self, remote: Remote) -> None | float:
if self.is_running():
try:
# Parse the peer's host address to create a Remote object, use peer here since we don't have the machine_name here
remote = Remote.from_ssh_uri(machine_name="peer", address=peer.host)
# Use the existing SSH reachability check
now = time.time()
@@ -33,7 +31,7 @@ class NetworkTechnology(NetworkTechnologyBase):
return (time.time() - now) * 1000
except ClanError as e:
log.debug(f"Error checking peer {peer.host}: {e}")
log.debug(f"Error checking peer {remote}: {e}")
return None
return None

127
pkgs/clan-cli/clan_lib/network/network.py
View File

@@ -12,9 +12,10 @@ from clan_cli.vars.get import get_machine_var
from clan_lib.errors import ClanError
from clan_lib.flake import Flake
from clan_lib.import_utils import ClassSource, import_with_source
from clan_lib.ssh.remote import Remote
if TYPE_CHECKING:
from clan_lib.ssh.remote import Remote
from clan_lib.machines.machines import Machine
log = logging.getLogger(__name__)
@@ -51,7 +52,7 @@ class Peer:
.lstrip("\n")
)
raise ClanError(msg)
return var.value.decode()
return var.value.decode().strip()
msg = f"Unknown Var Type {self._host}"
raise ClanError(msg)
@@ -75,7 +76,7 @@ class Network:
return self.module.is_running()
def ping(self, peer: str) -> float | None:
return self.module.ping(self.peers[peer])
return self.module.ping(self.remote(peer))
def remote(self, peer: str) -> "Remote":
# TODO raise exception if peer is not in peers
@@ -95,7 +96,7 @@ class NetworkTechnologyBase(ABC):
pass
@abstractmethod
def ping(self, peer: Peer) -> None | float:
def ping(self, remote: "Remote") -> None | float:
pass
@contextmanager
@@ -108,12 +109,18 @@ def networks_from_flake(flake: Flake) -> dict[str, Network]:
# TODO more precaching, for example for vars
flake.precache(
[
"clan.exports.instances.*.networking",
"clan.?exports.instances.*.networking",
]
)
networks: dict[str, Network] = {}
networks_ = flake.select("clan.exports.instances.*.networking")
for network_name, network in networks_.items():
networks_ = flake.select("clan.?exports.instances.*.networking")
if "exports" not in networks_:
msg = """You are not exporting the clan exports through your flake.
Please add exports next to clanInternals and nixosConfiguration into the global flake.
"""
log.warning(msg)
return {}
for network_name, network in networks_["exports"].items():
if network:
peers: dict[str, Peer] = {}
for _peer in network["peers"].values():
@@ -128,15 +135,103 @@ def networks_from_flake(flake: Flake) -> dict[str, Network]:
return networks
def get_best_network(machine_name: str, networks: dict[str, Network]) -> Network | None:
for network_name, network in sorted(
networks.items(), key=lambda network: -network[1].priority
):
if machine_name in network.peers:
if network.is_running() and network.ping(machine_name):
print(f"connecting via {network_name}")
return network
return None
@contextmanager
def get_best_remote(machine: "Machine") -> Iterator["Remote"]:
"""
Context manager that yields the best remote connection for a machine following this priority:
1. If machine has targetHost in inventory, return a direct connection
2. Return the highest priority network where machine is reachable
3. If no network works, try to get targetHost from machine nixos config
Args:
machine: Machine instance to connect to
Yields:
Remote object for connecting to the machine
Raises:
ClanError: If no connection method works
"""
# Step 1: Check if targetHost is set in inventory
inv_machine = machine.get_inv_machine()
target_host = inv_machine.get("deploy", {}).get("targetHost")
if target_host:
log.debug(f"Using targetHost from inventory for {machine.name}: {target_host}")
# Create a direct network with just this machine
try:
remote = Remote.from_ssh_uri(machine_name=machine.name, address=target_host)
yield remote
return
except Exception as e:
log.debug(f"Inventory targetHost not reachable for {machine.name}: {e}")
# Step 2: Try existing networks by priority
try:
networks = networks_from_flake(machine.flake)
sorted_networks = sorted(networks.items(), key=lambda x: -x[1].priority)
for network_name, network in sorted_networks:
if machine.name not in network.peers:
continue
# Check if network is running and machine is reachable
log.debug(f"trying to connect via {network_name}")
if network.is_running():
try:
ping_time = network.ping(machine.name)
if ping_time is not None:
log.info(
f"Machine {machine.name} reachable via {network_name} network"
)
yield network.remote(machine.name)
return
except Exception as e:
log.debug(f"Failed to reach {machine.name} via {network_name}: {e}")
else:
try:
log.debug(f"Establishing connection for network {network_name}")
with network.module.connection(network) as connected_network:
ping_time = connected_network.ping(machine.name)
if ping_time is not None:
log.info(
f"Machine {machine.name} reachable via {network_name} network after connection"
)
yield connected_network.remote(machine.name)
return
except Exception as e:
log.debug(
f"Failed to establish connection to {machine.name} via {network_name}: {e}"
)
except Exception as e:
log.debug(f"Failed to use networking modules to determine machines remote: {e}")
# Step 3: Try targetHost from machine nixos config
try:
target_host = machine.select('config.clan.core.networking."targetHost"')
if target_host:
log.debug(
f"Using targetHost from machine config for {machine.name}: {target_host}"
)
# Check if reachable
try:
remote = Remote.from_ssh_uri(
machine_name=machine.name, address=target_host
)
yield remote
return
except Exception as e:
log.debug(
f"Machine config targetHost not reachable for {machine.name}: {e}"
)
except Exception as e:
log.debug(f"Could not get targetHost from machine config: {e}")
# No connection method found
msg = f"Could not find any way to connect to machine '{machine.name}'. No targetHost configured and machine not reachable via any network."
raise ClanError(msg)
def get_network_overview(networks: dict[str, Network]) -> dict:

4
pkgs/clan-cli/clan_lib/network/network_test.py
View File

@@ -26,6 +26,7 @@ def test_networks_from_flake(mock_get_machine_var: MagicMock) -> None:
# Define the expected return value from flake.select
mock_networking_data = {
"exports": {
"vpn-network": {
"peers": {
"machine1": {
@@ -67,6 +68,7 @@ def test_networks_from_flake(mock_get_machine_var: MagicMock) -> None:
"priority": 500,
},
}
}
# Mock the select method
flake.select.return_value = mock_networking_data
@@ -75,7 +77,7 @@ def test_networks_from_flake(mock_get_machine_var: MagicMock) -> None:
networks = networks_from_flake(flake)
# Verify the flake.select was called with the correct pattern
flake.select.assert_called_once_with("clan.exports.instances.*.networking")
flake.select.assert_called_once_with("clan.?exports.instances.*.networking")
# Verify the returned networks
assert len(networks) == 2

166
pkgs/clan-cli/clan_lib/network/qr_code.py Normal file
View File

@@ -0,0 +1,166 @@
import json
import logging
from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from clan_lib.cmd import run
from clan_lib.errors import ClanError
from clan_lib.flake import Flake
from clan_lib.network.network import Network, Peer
from clan_lib.nix import nix_shell
from clan_lib.ssh.remote import Remote
from clan_lib.ssh.socks_wrapper import tor_wrapper
log = logging.getLogger(__name__)
@dataclass(frozen=True)
class RemoteWithNetwork:
network: Network
remote: Remote
@dataclass(frozen=True)
class QRCodeData:
addresses: list[RemoteWithNetwork]
@contextmanager
def get_best_remote(self) -> Iterator[Remote]:
for address in self.addresses:
try:
log.debug(f"Establishing connection via {address}")
with address.network.module.connection(
address.network
) as connected_network:
ping_time = connected_network.module.ping(address.remote)
if ping_time is not None:
log.info(f"reachable via {address} after connection")
yield address.remote
except Exception as e:
log.debug(f"Failed to establish connection via {address}: {e}")
def read_qr_json(qr_data: dict[str, Any], flake: Flake) -> QRCodeData:
"""
Parse QR code JSON contents and output a dict of networks with remotes.
Args:
qr_data: JSON data from QR code containing network information
flake: Flake instance for creating peers
Returns:
Dictionary mapping network type to dict with "network" and "remote" keys
Example input:
{
"pass": "password123",
"tor": "ssh://user@hostname.onion",
"addrs": ["ssh://user@192.168.1.100", "ssh://user@example.com"]
}
Example output:
{
"direct": {
"network": Network(...),
"remote": Remote(...)
},
"tor": {
"network": Network(...),
"remote": Remote(...)
}
}
"""
addresses: list[RemoteWithNetwork] = []
password = qr_data.get("pass")
# Process clearnet addresses
clearnet_addrs = qr_data.get("addrs", [])
if clearnet_addrs:
for addr in clearnet_addrs:
if isinstance(addr, str):
peer = Peer(name="installer", _host={"plain": addr}, flake=flake)
network = Network(
peers={"installer": peer},
module_name="clan_lib.network.direct",
priority=1000,
)
# Create the remote with password
remote = Remote.from_ssh_uri(
machine_name="installer",
address=addr,
).override(password=password)
addresses.append(RemoteWithNetwork(network=network, remote=remote))
else:
msg = f"Invalid address format: {addr}"
raise ClanError(msg)
# Process tor address
if tor_addr := qr_data.get("tor"):
peer = Peer(name="installer-tor", _host={"plain": tor_addr}, flake=flake)
network = Network(
peers={"installer-tor": peer},
module_name="clan_lib.network.tor",
priority=500,
)
# Create the remote with password and tor settings
remote = Remote.from_ssh_uri(
machine_name="installer-tor",
address=tor_addr,
).override(
password=password,
socks_port=9050,
socks_wrapper=tor_wrapper,
)
addresses.append(RemoteWithNetwork(network=network, remote=remote))
return QRCodeData(addresses=addresses)
def read_qr_image(image_path: Path) -> dict[str, Any]:
"""
Parse a QR code image and extract the JSON data.
Args:
image_path: Path to the QR code image file
Returns:
Parsed JSON data from the QR code
Raises:
ClanError: If the QR code cannot be read or contains invalid JSON
"""
if not image_path.exists():
msg = f"QR code image file not found: {image_path}"
raise ClanError(msg)
cmd = nix_shell(
["zbar"],
[
"zbarimg",
"--quiet",
"--raw",
str(image_path),
],
)
try:
res = run(cmd)
data = res.stdout.strip()
if not data:
msg = f"No QR code found in image: {image_path}"
raise ClanError(msg)
return json.loads(data)
except json.JSONDecodeError as e:
msg = f"Invalid JSON in QR code: {e}"
raise ClanError(msg) from e
except Exception as e:
msg = f"Failed to read QR code from {image_path}: {e}"
raise ClanError(msg) from e

10
pkgs/clan-cli/clan_lib/network/tor/__init__.py
View File

@@ -8,6 +8,8 @@ from typing import TYPE_CHECKING
from clan_lib.errors import ClanError
from clan_lib.network import Network, NetworkTechnologyBase, Peer
from clan_lib.network.tor.lib import is_tor_running, spawn_tor
from clan_lib.ssh.remote import Remote
from clan_lib.ssh.socks_wrapper import tor_wrapper
if TYPE_CHECKING:
from clan_lib.ssh.remote import Remote
@@ -27,11 +29,9 @@ class NetworkTechnology(NetworkTechnologyBase):
"""Check if Tor is running by sending HTTP request to SOCKS port."""
return is_tor_running(self.proxy)
def ping(self, peer: Peer) -> None | float:
def ping(self, remote: Remote) -> None | float:
if self.is_running():
try:
remote = self.remote(peer)
# Use the existing SSH reachability check
now = time.time()
remote.check_machine_ssh_reachable()
@@ -39,7 +39,7 @@ class NetworkTechnology(NetworkTechnologyBase):
return (time.time() - now) * 1000
except ClanError as e:
log.debug(f"Error checking peer {peer.host}: {e}")
log.debug(f"Error checking peer {remote}: {e}")
return None
return None
@@ -58,5 +58,5 @@ class NetworkTechnology(NetworkTechnologyBase):
address=peer.host,
command_prefix=peer.name,
socks_port=self.proxy,
socks_wrapper=["torify"],
socks_wrapper=tor_wrapper,
)

1
pkgs/clan-cli/clan_lib/nix/allowed-packages.json
View File

@@ -28,6 +28,7 @@
"sops",
"sshpass",
"tor",
"torsocks",
"util-linux",
"virt-viewer",
"virtiofsd",

5
pkgs/clan-cli/clan_lib/ssh/remote.py
View File

@@ -18,6 +18,7 @@ from clan_lib.errors import ClanError, indent_command # Assuming these are avai
from clan_lib.nix import nix_shell
from clan_lib.ssh.host_key import HostKeyCheck, hostkey_to_ssh_opts
from clan_lib.ssh.parse import parse_ssh_uri
from clan_lib.ssh.socks_wrapper import SocksWrapper
from clan_lib.ssh.sudo_askpass_proxy import SudoAskpassProxy
if TYPE_CHECKING:
@@ -42,7 +43,7 @@ class Remote:
verbose_ssh: bool = False
ssh_options: dict[str, str] = field(default_factory=dict)
socks_port: int | None = None
socks_wrapper: list[str] | None = None
socks_wrapper: SocksWrapper | None = None
_control_path_dir: Path | None = None
_askpass_path: str | None = None
@@ -63,7 +64,7 @@ class Remote:
private_key: Path | None = None,
password: str | None = None,
socks_port: int | None = None,
socks_wrapper: list[str] | None = None,
socks_wrapper: SocksWrapper | None = None,
command_prefix: str | None = None,
port: int | None = None,
ssh_options: dict[str, str] | None = None,

16
pkgs/clan-cli/clan_lib/ssh/socks_wrapper.py Normal file
View File

@@ -0,0 +1,16 @@
from dataclasses import dataclass
@dataclass(frozen=True)
class SocksWrapper:
"""Configuration for SOCKS proxy wrapper commands."""
# The command to execute for wrapping network connections through SOCKS (e.g., ["torify"])
cmd: list[str]
# Nix packages required to provide the wrapper command (e.g., ["tor", "torsocks"])
packages: list[str]
# Pre-configured Tor wrapper instance
tor_wrapper = SocksWrapper(cmd=["torify"], packages=["tor", "torsocks"])

0
pkgs/clan-cli/clan_cli/ssh/upload.py → pkgs/clan-cli/clan_lib/ssh/upload.py
View File