reformatted

This commit is contained in:
Jakub Kropáček 2025-01-07 16:18:04 +01:00
parent be8f8fbbf9
commit 6b54f0bc42

View file

@ -2,15 +2,18 @@
import argparse import argparse
import json import json
import ipaddress import ipaddress
import time
from subprocess import check_output from subprocess import check_output
def _get_available_machines() -> list: def _get_available_machines() -> list:
output = check_output(['nix', 'flake', 'show', '--json']) output = check_output(["nix", "flake", "show", "--json"])
parsed_output = json.loads(output) parsed_output = json.loads(output)
machines = parsed_output.get('nixosConfigurations', dict()).keys() machines = parsed_output.get("nixosConfigurations", dict()).keys()
return list(machines) return list(machines)
def _validate_ip(ip: str) -> bool: def _validate_ip(ip: str) -> bool:
try: try:
ipaddress.ip_address(ip) ipaddress.ip_address(ip)
@ -18,25 +21,27 @@ def _validate_ip(ip: str) -> bool:
except ValueError: except ValueError:
return False return False
def _check_ssh_connection(ip: str) -> bool: def _check_ssh_connection(ip: str) -> bool:
try: try:
check_output(['ssh', f'root@{ip}', 'echo', 'Connected']) check_output(["ssh", f"root@{ip}", "echo", "Connected"])
return True return True
except Exception: except Exception:
return False return False
def bootstrap_machine(ip: str): def bootstrap_machine(ip: str):
check_output( check_output(
[ [
'nix', "nix",
'run', "run",
'github:nix-community/nixos-anywhere', "github:nix-community/nixos-anywhere",
'--', "--",
'--flake', "--flake",
'".#bootstrap"', '".#bootstrap"',
'--target-host', "--target-host",
f'root@{ip}', f"root@{ip}",
'--build-on-remote' "--build-on-remote",
] ]
) )
@ -48,22 +53,33 @@ def get_ssh_key(ip: str) -> str:
def get_machine_config(machine_name: str) -> dict: def get_machine_config(machine_name: str) -> dict:
output = check_output(['nix', 'eval', '--json', f'.#nixosConfigurations.{machine_name}.config.kropcloud']) output = check_output(
[
"nix",
"eval",
"--json",
f".#nixosConfigurations.{machine_name}.config.kropcloud",
]
)
return json.loads(output) return json.loads(output)
def main() -> int: def main() -> int:
parser = argparse.ArgumentParser(description='Install a machine') parser = argparse.ArgumentParser(description="Install a machine")
parser.add_argument('machine_name', type=str, help='The name of the machine to install') parser.add_argument(
parser.add_argument('machine_ip', type=str, help='The ip of the machine to install') "machine_name", type=str, help="The name of the machine to install"
)
parser.add_argument("machine_ip", type=str, help="The ip of the machine to install")
args = parser.parse_args() args = parser.parse_args()
machine_name = args.machine_name machine_name = args.machine_name
if not machine_name in _get_available_machines(): if machine_name not in _get_available_machines():
raise ValueError(f'Machine {machine_name} not found, available machines are: {_get_available_machines()}') raise ValueError(
f"Machine {machine_name} not found, available machines are: {_get_available_machines()}"
)
if _validate_ip(args.machine_ip): if _validate_ip(args.machine_ip):
raise ValueError(f'Invalid IP address {args.machine_ip}') raise ValueError(f"Invalid IP address {args.machine_ip}")
machine_config = get_machine_config(machine_name) machine_config = get_machine_config(machine_name)
print(machine_config) print(machine_config)
@ -75,7 +91,7 @@ def main() -> int:
# # connect and get ssh keys # # connect and get ssh keys
ssh_key = get_ssh_key() # ssh_key = get_ssh_key()
# Add the ssh key to keys in secrets/secrets.nix # Add the ssh key to keys in secrets/secrets.nix
# and rekey the secrets # and rekey the secrets
@ -84,5 +100,6 @@ def main() -> int:
return 0 return 0
if __name__ == '__main__':
if __name__ == "__main__":
raise SystemExit(main()) raise SystemExit(main())