Merge pull request 'nix_shell' (#3339) from nix_shell into main
Reviewed-on: https://git.clan.lol/clan/clan-core/pulls/3339
This commit is contained in:
@@ -6,7 +6,7 @@ from typing import Any, Literal
|
|||||||
|
|
||||||
from clan_cli.cmd import RunOpts
|
from clan_cli.cmd import RunOpts
|
||||||
from clan_cli.errors import ClanError
|
from clan_cli.errors import ClanError
|
||||||
from clan_cli.nix import nix_shell, run_no_stdout
|
from clan_cli.nix import nix_shell_legacy, run_no_stdout
|
||||||
|
|
||||||
from . import API
|
from . import API
|
||||||
|
|
||||||
@@ -126,7 +126,7 @@ def show_block_devices() -> Blockdevices:
|
|||||||
It must return a list of block devices.
|
It must return a list of block devices.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
cmd = nix_shell(
|
cmd = nix_shell_legacy(
|
||||||
["nixpkgs#util-linux"],
|
["nixpkgs#util-linux"],
|
||||||
[
|
[
|
||||||
"lsblk",
|
"lsblk",
|
||||||
|
|||||||
@@ -3,7 +3,7 @@ import re
|
|||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
|
|
||||||
from clan_cli.cmd import run_no_stdout
|
from clan_cli.cmd import run_no_stdout
|
||||||
from clan_cli.nix import nix_shell
|
from clan_cli.nix import nix_shell_legacy
|
||||||
|
|
||||||
from . import API
|
from . import API
|
||||||
|
|
||||||
@@ -89,7 +89,7 @@ def parse_avahi_output(output: str) -> DNSInfo:
|
|||||||
|
|
||||||
@API.register
|
@API.register
|
||||||
def show_mdns() -> DNSInfo:
|
def show_mdns() -> DNSInfo:
|
||||||
cmd = nix_shell(
|
cmd = nix_shell_legacy(
|
||||||
["nixpkgs#avahi"],
|
["nixpkgs#avahi"],
|
||||||
[
|
[
|
||||||
"avahi-browse",
|
"avahi-browse",
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
from clan_cli.cmd import run
|
from clan_cli.cmd import run
|
||||||
from clan_cli.nix import nix_shell
|
from clan_cli.nix import nix_shell_legacy
|
||||||
|
|
||||||
_works: bool | None = None
|
_works: bool | None = None
|
||||||
|
|
||||||
@@ -13,7 +13,7 @@ def bubblewrap_works() -> bool:
|
|||||||
|
|
||||||
def _bubblewrap_works() -> bool:
|
def _bubblewrap_works() -> bool:
|
||||||
# fmt: off
|
# fmt: off
|
||||||
cmd = nix_shell(
|
cmd = nix_shell_legacy(
|
||||||
[
|
[
|
||||||
"nixpkgs#bash",
|
"nixpkgs#bash",
|
||||||
"nixpkgs#bubblewrap",
|
"nixpkgs#bubblewrap",
|
||||||
|
|||||||
@@ -9,7 +9,7 @@ from clan_cli.cmd import CmdOut, RunOpts, run
|
|||||||
from clan_cli.errors import ClanError
|
from clan_cli.errors import ClanError
|
||||||
from clan_cli.flake import Flake
|
from clan_cli.flake import Flake
|
||||||
from clan_cli.inventory import Inventory, init_inventory
|
from clan_cli.inventory import Inventory, init_inventory
|
||||||
from clan_cli.nix import nix_command, nix_metadata, nix_shell
|
from clan_cli.nix import nix_command, nix_metadata, nix_shell_legacy
|
||||||
from clan_cli.templates import (
|
from clan_cli.templates import (
|
||||||
InputPrio,
|
InputPrio,
|
||||||
TemplateName,
|
TemplateName,
|
||||||
@@ -41,7 +41,7 @@ class CreateOptions:
|
|||||||
|
|
||||||
|
|
||||||
def git_command(directory: Path, *args: str) -> list[str]:
|
def git_command(directory: Path, *args: str) -> list[str]:
|
||||||
return nix_shell(["nixpkgs#git"], ["git", "-C", str(directory), *args])
|
return nix_shell_legacy(["nixpkgs#git"], ["git", "-C", str(directory), *args])
|
||||||
|
|
||||||
|
|
||||||
@API.register
|
@API.register
|
||||||
|
|||||||
@@ -18,7 +18,7 @@ from clan_cli.errors import ClanError
|
|||||||
from clan_cli.git import commit_files
|
from clan_cli.git import commit_files
|
||||||
from clan_cli.machines.inventory import get_all_machines, get_selected_machines
|
from clan_cli.machines.inventory import get_all_machines, get_selected_machines
|
||||||
from clan_cli.machines.machines import Machine
|
from clan_cli.machines.machines import Machine
|
||||||
from clan_cli.nix import nix_shell
|
from clan_cli.nix import nix_shell_legacy
|
||||||
|
|
||||||
from .check import check_secrets
|
from .check import check_secrets
|
||||||
from .public_modules import FactStoreBase
|
from .public_modules import FactStoreBase
|
||||||
@@ -39,7 +39,7 @@ def read_multiline_input(prompt: str = "Finish with Ctrl-D") -> str:
|
|||||||
|
|
||||||
def bubblewrap_cmd(generator: str, facts_dir: Path, secrets_dir: Path) -> list[str]:
|
def bubblewrap_cmd(generator: str, facts_dir: Path, secrets_dir: Path) -> list[str]:
|
||||||
# fmt: off
|
# fmt: off
|
||||||
return nix_shell(
|
return nix_shell_legacy(
|
||||||
[
|
[
|
||||||
"nixpkgs#bash",
|
"nixpkgs#bash",
|
||||||
"nixpkgs#bubblewrap",
|
"nixpkgs#bubblewrap",
|
||||||
|
|||||||
@@ -5,7 +5,7 @@ from typing import override
|
|||||||
|
|
||||||
from clan_cli.cmd import Log, RunOpts
|
from clan_cli.cmd import Log, RunOpts
|
||||||
from clan_cli.machines.machines import Machine
|
from clan_cli.machines.machines import Machine
|
||||||
from clan_cli.nix import nix_shell
|
from clan_cli.nix import nix_shell_legacy
|
||||||
|
|
||||||
from . import SecretStoreBase
|
from . import SecretStoreBase
|
||||||
|
|
||||||
@@ -18,7 +18,7 @@ class SecretStore(SecretStoreBase):
|
|||||||
self, service: str, name: str, value: bytes, groups: list[str]
|
self, service: str, name: str, value: bytes, groups: list[str]
|
||||||
) -> Path | None:
|
) -> Path | None:
|
||||||
subprocess.run(
|
subprocess.run(
|
||||||
nix_shell(
|
nix_shell_legacy(
|
||||||
["nixpkgs#pass"],
|
["nixpkgs#pass"],
|
||||||
["pass", "insert", "-m", f"machines/{self.machine.name}/{name}"],
|
["pass", "insert", "-m", f"machines/{self.machine.name}/{name}"],
|
||||||
),
|
),
|
||||||
@@ -29,7 +29,7 @@ class SecretStore(SecretStoreBase):
|
|||||||
|
|
||||||
def get(self, service: str, name: str) -> bytes:
|
def get(self, service: str, name: str) -> bytes:
|
||||||
return subprocess.run(
|
return subprocess.run(
|
||||||
nix_shell(
|
nix_shell_legacy(
|
||||||
["nixpkgs#pass"],
|
["nixpkgs#pass"],
|
||||||
["pass", "show", f"machines/{self.machine.name}/{name}"],
|
["pass", "show", f"machines/{self.machine.name}/{name}"],
|
||||||
),
|
),
|
||||||
@@ -51,7 +51,7 @@ class SecretStore(SecretStoreBase):
|
|||||||
hashes = []
|
hashes = []
|
||||||
hashes.append(
|
hashes.append(
|
||||||
subprocess.run(
|
subprocess.run(
|
||||||
nix_shell(
|
nix_shell_legacy(
|
||||||
["nixpkgs#git"],
|
["nixpkgs#git"],
|
||||||
[
|
[
|
||||||
"git",
|
"git",
|
||||||
@@ -71,7 +71,7 @@ class SecretStore(SecretStoreBase):
|
|||||||
if symlink.is_symlink():
|
if symlink.is_symlink():
|
||||||
hashes.append(
|
hashes.append(
|
||||||
subprocess.run(
|
subprocess.run(
|
||||||
nix_shell(
|
nix_shell_legacy(
|
||||||
["nixpkgs#git"],
|
["nixpkgs#git"],
|
||||||
[
|
[
|
||||||
"git",
|
"git",
|
||||||
|
|||||||
@@ -13,7 +13,7 @@ from clan_cli.errors import ClanError
|
|||||||
from clan_cli.facts.generate import generate_facts
|
from clan_cli.facts.generate import generate_facts
|
||||||
from clan_cli.facts.secret_modules import SecretStoreBase
|
from clan_cli.facts.secret_modules import SecretStoreBase
|
||||||
from clan_cli.machines.machines import Machine
|
from clan_cli.machines.machines import Machine
|
||||||
from clan_cli.nix import nix_shell
|
from clan_cli.nix import nix_shell_legacy
|
||||||
from clan_cli.vars.generate import generate_vars
|
from clan_cli.vars.generate import generate_vars
|
||||||
|
|
||||||
from .automount import pause_automounting
|
from .automount import pause_automounting
|
||||||
@@ -147,7 +147,7 @@ def flash_machine(
|
|||||||
disko_install.extend(["--option", "dry-run", "true"])
|
disko_install.extend(["--option", "dry-run", "true"])
|
||||||
disko_install.extend(extra_args)
|
disko_install.extend(extra_args)
|
||||||
|
|
||||||
cmd = nix_shell(
|
cmd = nix_shell_legacy(
|
||||||
["nixpkgs#disko"],
|
["nixpkgs#disko"],
|
||||||
disko_install,
|
disko_install,
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ from pathlib import Path
|
|||||||
from .cmd import Log, RunOpts, run
|
from .cmd import Log, RunOpts, run
|
||||||
from .errors import ClanError
|
from .errors import ClanError
|
||||||
from .locked_open import locked_open
|
from .locked_open import locked_open
|
||||||
from .nix import run_cmd
|
from .nix import nix_shell
|
||||||
|
|
||||||
|
|
||||||
def commit_file(
|
def commit_file(
|
||||||
@@ -72,7 +72,7 @@ def _commit_file_to_git(
|
|||||||
|
|
||||||
with locked_open(real_git_dir / "clan.lock", "w+"):
|
with locked_open(real_git_dir / "clan.lock", "w+"):
|
||||||
for file_path in file_paths:
|
for file_path in file_paths:
|
||||||
cmd = run_cmd(
|
cmd = nix_shell(
|
||||||
["git"],
|
["git"],
|
||||||
["git", "-C", str(repo_dir), "add", "--", str(file_path)],
|
["git", "-C", str(repo_dir), "add", "--", str(file_path)],
|
||||||
)
|
)
|
||||||
@@ -87,7 +87,7 @@ def _commit_file_to_git(
|
|||||||
)
|
)
|
||||||
|
|
||||||
# check if there is a diff
|
# check if there is a diff
|
||||||
cmd = run_cmd(
|
cmd = nix_shell(
|
||||||
["git"],
|
["git"],
|
||||||
["git", "-C", str(repo_dir), "diff", "--cached", "--exit-code", "--"]
|
["git", "-C", str(repo_dir), "diff", "--cached", "--exit-code", "--"]
|
||||||
+ [str(file_path) for file_path in file_paths],
|
+ [str(file_path) for file_path in file_paths],
|
||||||
@@ -98,7 +98,7 @@ def _commit_file_to_git(
|
|||||||
return
|
return
|
||||||
|
|
||||||
# commit only that file
|
# commit only that file
|
||||||
cmd = run_cmd(
|
cmd = nix_shell(
|
||||||
["git"],
|
["git"],
|
||||||
[
|
[
|
||||||
"git",
|
"git",
|
||||||
|
|||||||
@@ -13,7 +13,7 @@ from clan_cli.errors import ClanCmdError, ClanError
|
|||||||
from clan_cli.flake import Flake
|
from clan_cli.flake import Flake
|
||||||
from clan_cli.git import commit_file
|
from clan_cli.git import commit_file
|
||||||
from clan_cli.machines.machines import Machine
|
from clan_cli.machines.machines import Machine
|
||||||
from clan_cli.nix import nix_config, nix_eval, run_cmd
|
from clan_cli.nix import nix_config, nix_eval, nix_shell
|
||||||
|
|
||||||
from .types import machine_name_type
|
from .types import machine_name_type
|
||||||
|
|
||||||
@@ -144,7 +144,7 @@ def generate_machine_hardware_info(opts: HardwareGenerateOptions) -> HardwareCon
|
|||||||
if opts.password:
|
if opts.password:
|
||||||
deps += ["sshpass"]
|
deps += ["sshpass"]
|
||||||
|
|
||||||
cmd = run_cmd(
|
cmd = nix_shell(
|
||||||
deps,
|
deps,
|
||||||
[
|
[
|
||||||
*(["sshpass", "-p", opts.password] if opts.password else []),
|
*(["sshpass", "-p", opts.password] if opts.password else []),
|
||||||
|
|||||||
@@ -18,7 +18,7 @@ from clan_cli.errors import ClanError
|
|||||||
from clan_cli.facts.generate import generate_facts
|
from clan_cli.facts.generate import generate_facts
|
||||||
from clan_cli.machines.hardware import HardwareConfig
|
from clan_cli.machines.hardware import HardwareConfig
|
||||||
from clan_cli.machines.machines import Machine
|
from clan_cli.machines.machines import Machine
|
||||||
from clan_cli.nix import nix_shell
|
from clan_cli.nix import nix_shell_legacy
|
||||||
from clan_cli.ssh.deploy_info import DeployInfo, find_reachable_host, ssh_command_parse
|
from clan_cli.ssh.deploy_info import DeployInfo, find_reachable_host, ssh_command_parse
|
||||||
from clan_cli.ssh.host_key import HostKeyCheck
|
from clan_cli.ssh.host_key import HostKeyCheck
|
||||||
from clan_cli.vars.generate import generate_vars
|
from clan_cli.vars.generate import generate_vars
|
||||||
@@ -146,7 +146,7 @@ def install_machine(opts: InstallOptions) -> None:
|
|||||||
# cmd.append("--ssh-option")
|
# cmd.append("--ssh-option")
|
||||||
# cmd.append("ProxyCommand=nc -x 127.0.0.1:9050 -X 5 %h %p")
|
# cmd.append("ProxyCommand=nc -x 127.0.0.1:9050 -X 5 %h %p")
|
||||||
run(
|
run(
|
||||||
nix_shell(
|
nix_shell_legacy(
|
||||||
[
|
[
|
||||||
"nixpkgs#nixos-anywhere",
|
"nixpkgs#nixos-anywhere",
|
||||||
"nixpkgs#tor",
|
"nixpkgs#tor",
|
||||||
@@ -157,7 +157,7 @@ def install_machine(opts: InstallOptions) -> None:
|
|||||||
)
|
)
|
||||||
else:
|
else:
|
||||||
run(
|
run(
|
||||||
nix_shell(
|
nix_shell_legacy(
|
||||||
["nixpkgs#nixos-anywhere"],
|
["nixpkgs#nixos-anywhere"],
|
||||||
cmd,
|
cmd,
|
||||||
),
|
),
|
||||||
|
|||||||
@@ -20,7 +20,7 @@ from clan_cli.inventory import (
|
|||||||
patch_inventory_with,
|
patch_inventory_with,
|
||||||
)
|
)
|
||||||
from clan_cli.machines.hardware import HardwareConfig
|
from clan_cli.machines.hardware import HardwareConfig
|
||||||
from clan_cli.nix import nix_eval, nix_shell
|
from clan_cli.nix import nix_eval, nix_shell_legacy
|
||||||
from clan_cli.tags import list_nixos_machines_by_tags
|
from clan_cli.tags import list_nixos_machines_by_tags
|
||||||
|
|
||||||
log = logging.getLogger(__name__)
|
log = logging.getLogger(__name__)
|
||||||
@@ -126,7 +126,7 @@ def check_machine_online(
|
|||||||
|
|
||||||
timeout = opts.timeout if opts and opts.timeout else 20
|
timeout = opts.timeout if opts and opts.timeout else 20
|
||||||
|
|
||||||
cmd = nix_shell(
|
cmd = nix_shell_legacy(
|
||||||
["nixpkgs#util-linux", *(["nixpkgs#openssh"] if hostname else [])],
|
["nixpkgs#util-linux", *(["nixpkgs#openssh"] if hostname else [])],
|
||||||
[
|
[
|
||||||
"ssh",
|
"ssh",
|
||||||
|
|||||||
@@ -105,8 +105,8 @@ def nix_metadata(flake_url: str | Path) -> dict[str, Any]:
|
|||||||
return data
|
return data
|
||||||
|
|
||||||
|
|
||||||
# Deprecated: use run_cmd() instead
|
# Deprecated: use nix_shell() instead
|
||||||
def nix_shell(packages: list[str], cmd: list[str]) -> list[str]:
|
def nix_shell_legacy(packages: list[str], cmd: list[str]) -> list[str]:
|
||||||
# we cannot use nix-shell inside the nix sandbox
|
# we cannot use nix-shell inside the nix sandbox
|
||||||
# in our tests we just make sure we have all the packages
|
# in our tests we just make sure we have all the packages
|
||||||
if (
|
if (
|
||||||
@@ -124,46 +124,48 @@ def nix_shell(packages: list[str], cmd: list[str]) -> list[str]:
|
|||||||
|
|
||||||
|
|
||||||
# lazy loads list of allowed and static programs
|
# lazy loads list of allowed and static programs
|
||||||
class Programs:
|
class Packages:
|
||||||
allowed_programs: set[str] | None = None
|
allowed_packages: set[str] | None = None
|
||||||
static_programs: set[str] | None = None
|
static_packages: set[str] | None = None
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def ensure_allowed(cls: type["Programs"], program: str) -> None:
|
def ensure_allowed(cls: type["Packages"], package: str) -> None:
|
||||||
if cls.allowed_programs is None:
|
if cls.allowed_packages is None:
|
||||||
with (Path(__file__).parent / "allowed-programs.json").open() as f:
|
with (Path(__file__).parent / "allowed-packages.json").open() as f:
|
||||||
cls.allowed_programs = allowed_programs = set(json.load(f))
|
cls.allowed_packages = allowed_packages = set(json.load(f))
|
||||||
else:
|
else:
|
||||||
allowed_programs = cls.allowed_programs
|
allowed_packages = cls.allowed_packages
|
||||||
|
|
||||||
if program not in allowed_programs:
|
if package not in allowed_packages:
|
||||||
msg = f"Program not allowed: '{program}', allowed programs are:\n{'\n'.join(allowed_programs)}"
|
msg = f"Package not allowed: '{package}', allowed packages are:\n{'\n'.join(allowed_packages)}"
|
||||||
raise ClanError(msg)
|
raise ClanError(msg)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def is_static(cls: type["Programs"], program: str) -> bool:
|
def is_provided(cls: type["Packages"], program: str) -> bool:
|
||||||
"""
|
"""
|
||||||
Determines if a program is statically shipped with this clan distribution
|
Determines if a program is shipped with the clan package.
|
||||||
"""
|
"""
|
||||||
if cls.static_programs is None:
|
if cls.static_packages is None:
|
||||||
cls.static_programs = set(
|
cls.static_packages = set(
|
||||||
os.environ.get("CLAN_STATIC_PROGRAMS", "").split(":")
|
os.environ.get("CLAN_PROVIDED_PACKAGES", "").split(":")
|
||||||
)
|
)
|
||||||
return program in cls.static_programs
|
return program in cls.static_packages
|
||||||
|
|
||||||
|
|
||||||
# Alternative implementation of nix_shell() to replace nix_shell() at some point
|
# Alternative implementation of nix_shell() to replace nix_shell_legacy() at some point
|
||||||
# Features:
|
# Features:
|
||||||
# - allow list for programs (need to be specified in allowed-programs.json)
|
# - allow list for programs (need to be specified in allowed-packages.json)
|
||||||
# - be abe to compute a closure of all deps for testing
|
# - be abe to compute a closure of all deps for testing
|
||||||
# - build clan distributions that ship some or all packages (eg. clan-cli-full)
|
# - build clan distributions that ship some or all packages (eg. clan-cli-full)
|
||||||
def run_cmd(programs: list[str], cmd: list[str]) -> list[str]:
|
def nix_shell(packages: list[str], cmd: list[str]) -> list[str]:
|
||||||
for program in programs:
|
for program in packages:
|
||||||
Programs.ensure_allowed(program)
|
Packages.ensure_allowed(program)
|
||||||
if os.environ.get("IN_NIX_SANDBOX"):
|
if os.environ.get("IN_NIX_SANDBOX"):
|
||||||
return cmd
|
return cmd
|
||||||
missing_packages = [
|
missing_packages = [
|
||||||
f"nixpkgs#{program}" for program in programs if not Programs.is_static(program)
|
f"nixpkgs#{package}"
|
||||||
|
for package in packages
|
||||||
|
if not Packages.is_provided(package)
|
||||||
]
|
]
|
||||||
if not missing_packages:
|
if not missing_packages:
|
||||||
return cmd
|
return cmd
|
||||||
|
|||||||
@@ -11,7 +11,7 @@ from clan_cli.completions import (
|
|||||||
complete_users,
|
complete_users,
|
||||||
)
|
)
|
||||||
from clan_cli.errors import ClanError
|
from clan_cli.errors import ClanError
|
||||||
from clan_cli.nix import nix_shell
|
from clan_cli.nix import nix_shell_legacy
|
||||||
|
|
||||||
from .secrets import encrypt_secret, sops_secrets_folder
|
from .secrets import encrypt_secret, sops_secrets_folder
|
||||||
|
|
||||||
@@ -30,7 +30,7 @@ def import_sops(args: argparse.Namespace) -> None:
|
|||||||
if args.input_type:
|
if args.input_type:
|
||||||
cmd += ["--input-type", args.input_type]
|
cmd += ["--input-type", args.input_type]
|
||||||
cmd += ["--output-type", "json", "--decrypt", args.sops_file]
|
cmd += ["--output-type", "json", "--decrypt", args.sops_file]
|
||||||
cmd = nix_shell(["nixpkgs#sops", "nixpkgs#gnupg"], cmd)
|
cmd = nix_shell_legacy(["nixpkgs#sops", "nixpkgs#gnupg"], cmd)
|
||||||
|
|
||||||
res = run(cmd, RunOpts(error_msg=f"Could not import sops file {file}"))
|
res = run(cmd, RunOpts(error_msg=f"Could not import sops file {file}"))
|
||||||
secrets = json.loads(res.stdout)
|
secrets = json.loads(res.stdout)
|
||||||
|
|||||||
@@ -16,7 +16,7 @@ from clan_cli.api import API
|
|||||||
from clan_cli.cmd import Log, RunOpts, run
|
from clan_cli.cmd import Log, RunOpts, run
|
||||||
from clan_cli.dirs import user_config_dir
|
from clan_cli.dirs import user_config_dir
|
||||||
from clan_cli.errors import ClanError
|
from clan_cli.errors import ClanError
|
||||||
from clan_cli.nix import nix_shell
|
from clan_cli.nix import nix_shell_legacy
|
||||||
|
|
||||||
from .folders import sops_machines_folder, sops_users_folder
|
from .folders import sops_machines_folder, sops_users_folder
|
||||||
|
|
||||||
@@ -233,7 +233,7 @@ def sops_run(
|
|||||||
raise ClanError(msg)
|
raise ClanError(msg)
|
||||||
sops_cmd.append(str(secret_path))
|
sops_cmd.append(str(secret_path))
|
||||||
|
|
||||||
cmd = nix_shell(["nixpkgs#sops", "nixpkgs#gnupg"], sops_cmd)
|
cmd = nix_shell_legacy(["nixpkgs#sops", "nixpkgs#gnupg"], sops_cmd)
|
||||||
opts = (
|
opts = (
|
||||||
dataclasses.replace(run_opts, env=environ)
|
dataclasses.replace(run_opts, env=environ)
|
||||||
if run_opts
|
if run_opts
|
||||||
@@ -249,7 +249,7 @@ def sops_run(
|
|||||||
|
|
||||||
|
|
||||||
def get_public_age_key(privkey: str) -> str:
|
def get_public_age_key(privkey: str) -> str:
|
||||||
cmd = nix_shell(["nixpkgs#age"], ["age-keygen", "-y"])
|
cmd = nix_shell_legacy(["nixpkgs#age"], ["age-keygen", "-y"])
|
||||||
|
|
||||||
error_msg = "Failed to get public key for age private key. Is the key malformed?"
|
error_msg = "Failed to get public key for age private key. Is the key malformed?"
|
||||||
res = run(cmd, RunOpts(input=privkey.encode(), error_msg=error_msg))
|
res = run(cmd, RunOpts(input=privkey.encode(), error_msg=error_msg))
|
||||||
@@ -257,7 +257,7 @@ def get_public_age_key(privkey: str) -> str:
|
|||||||
|
|
||||||
|
|
||||||
def generate_private_key(out_file: Path | None = None) -> tuple[str, str]:
|
def generate_private_key(out_file: Path | None = None) -> tuple[str, str]:
|
||||||
cmd = nix_shell(["nixpkgs#age"], ["age-keygen"])
|
cmd = nix_shell_legacy(["nixpkgs#age"], ["age-keygen"])
|
||||||
try:
|
try:
|
||||||
proc = run(cmd)
|
proc = run(cmd)
|
||||||
res = proc.stdout.strip()
|
res = proc.stdout.strip()
|
||||||
|
|||||||
@@ -14,7 +14,7 @@ from clan_cli.completions import (
|
|||||||
)
|
)
|
||||||
from clan_cli.errors import ClanError
|
from clan_cli.errors import ClanError
|
||||||
from clan_cli.machines.machines import Machine
|
from clan_cli.machines.machines import Machine
|
||||||
from clan_cli.nix import nix_shell
|
from clan_cli.nix import nix_shell_legacy
|
||||||
from clan_cli.ssh.host import Host, is_ssh_reachable
|
from clan_cli.ssh.host import Host, is_ssh_reachable
|
||||||
from clan_cli.ssh.host_key import HostKeyCheck
|
from clan_cli.ssh.host_key import HostKeyCheck
|
||||||
from clan_cli.ssh.parse import parse_deployment_address
|
from clan_cli.ssh.parse import parse_deployment_address
|
||||||
@@ -65,7 +65,7 @@ def find_reachable_host(
|
|||||||
|
|
||||||
|
|
||||||
def qrcode_scan(picture_file: Path) -> str:
|
def qrcode_scan(picture_file: Path) -> str:
|
||||||
cmd = nix_shell(
|
cmd = nix_shell_legacy(
|
||||||
["nixpkgs#zbar"],
|
["nixpkgs#zbar"],
|
||||||
[
|
[
|
||||||
"zbarimg",
|
"zbarimg",
|
||||||
|
|||||||
@@ -12,7 +12,7 @@ from typing import Any
|
|||||||
from clan_cli.cmd import CmdOut, RunOpts, run
|
from clan_cli.cmd import CmdOut, RunOpts, run
|
||||||
from clan_cli.colors import AnsiColor
|
from clan_cli.colors import AnsiColor
|
||||||
from clan_cli.errors import ClanError
|
from clan_cli.errors import ClanError
|
||||||
from clan_cli.nix import run_cmd
|
from clan_cli.nix import nix_shell
|
||||||
from clan_cli.ssh.host_key import HostKeyCheck
|
from clan_cli.ssh.host_key import HostKeyCheck
|
||||||
|
|
||||||
cmdlog = logging.getLogger(__name__)
|
cmdlog = logging.getLogger(__name__)
|
||||||
@@ -216,7 +216,7 @@ class Host:
|
|||||||
*ssh_opts,
|
*ssh_opts,
|
||||||
]
|
]
|
||||||
|
|
||||||
return run_cmd(packages, cmd)
|
return nix_shell(packages, cmd)
|
||||||
|
|
||||||
def connect_ssh_shell(
|
def connect_ssh_shell(
|
||||||
self, *, password: str | None = None, tor_socks: bool = False
|
self, *, password: str | None = None, tor_socks: bool = False
|
||||||
|
|||||||
@@ -10,7 +10,7 @@ from dataclasses import dataclass
|
|||||||
from clan_cli.async_run import AsyncRuntime
|
from clan_cli.async_run import AsyncRuntime
|
||||||
from clan_cli.cmd import Log, RunOpts, run
|
from clan_cli.cmd import Log, RunOpts, run
|
||||||
from clan_cli.errors import TorConnectionError, TorSocksError
|
from clan_cli.errors import TorConnectionError, TorSocksError
|
||||||
from clan_cli.nix import nix_shell
|
from clan_cli.nix import nix_shell_legacy
|
||||||
|
|
||||||
log = logging.getLogger(__name__)
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
@@ -117,7 +117,7 @@ def spawn_tor(runtime: AsyncRuntime) -> None:
|
|||||||
"""Starts Tor process using nix-shell."""
|
"""Starts Tor process using nix-shell."""
|
||||||
cmd_args = ["tor", "--HardwareAccel", "1"]
|
cmd_args = ["tor", "--HardwareAccel", "1"]
|
||||||
packages = ["nixpkgs#tor"]
|
packages = ["nixpkgs#tor"]
|
||||||
cmd = nix_shell(packages, cmd_args)
|
cmd = nix_shell_legacy(packages, cmd_args)
|
||||||
runtime.async_run(None, run, cmd, RunOpts(log=Log.BOTH))
|
runtime.async_run(None, run, cmd, RunOpts(log=Log.BOTH))
|
||||||
log.debug("Attempting to start Tor")
|
log.debug("Attempting to start Tor")
|
||||||
|
|
||||||
|
|||||||
@@ -2,19 +2,21 @@ import subprocess
|
|||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
from clan_cli.nix import nix_shell
|
from clan_cli.nix import nix_shell_legacy
|
||||||
|
|
||||||
|
|
||||||
# fixture for git_repo
|
# fixture for git_repo
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def git_repo(temp_dir: Path) -> Path:
|
def git_repo(temp_dir: Path) -> Path:
|
||||||
# initialize a git repository
|
# initialize a git repository
|
||||||
cmd = nix_shell(["nixpkgs#git"], ["git", "init"])
|
cmd = nix_shell_legacy(["nixpkgs#git"], ["git", "init"])
|
||||||
subprocess.run(cmd, cwd=temp_dir, check=True)
|
subprocess.run(cmd, cwd=temp_dir, check=True)
|
||||||
# set user.name and user.email
|
# set user.name and user.email
|
||||||
cmd = nix_shell(["nixpkgs#git"], ["git", "config", "user.name", "test"])
|
cmd = nix_shell_legacy(["nixpkgs#git"], ["git", "config", "user.name", "test"])
|
||||||
subprocess.run(cmd, cwd=temp_dir, check=True)
|
subprocess.run(cmd, cwd=temp_dir, check=True)
|
||||||
cmd = nix_shell(["nixpkgs#git"], ["git", "config", "user.email", "test@test.test"])
|
cmd = nix_shell_legacy(
|
||||||
|
["nixpkgs#git"], ["git", "config", "user.email", "test@test.test"]
|
||||||
|
)
|
||||||
subprocess.run(cmd, cwd=temp_dir, check=True)
|
subprocess.run(cmd, cwd=temp_dir, check=True)
|
||||||
# return the path to the git repository
|
# return the path to the git repository
|
||||||
return temp_dir
|
return temp_dir
|
||||||
|
|||||||
@@ -6,7 +6,7 @@ from clan_cli.facts.secret_modules.password_store import SecretStore
|
|||||||
from clan_cli.flake import Flake
|
from clan_cli.flake import Flake
|
||||||
from clan_cli.machines.facts import machine_get_fact
|
from clan_cli.machines.facts import machine_get_fact
|
||||||
from clan_cli.machines.machines import Machine
|
from clan_cli.machines.machines import Machine
|
||||||
from clan_cli.nix import nix_shell
|
from clan_cli.nix import nix_shell_legacy
|
||||||
from clan_cli.ssh.host import Host
|
from clan_cli.ssh.host import Host
|
||||||
from clan_cli.tests.fixtures_flakes import ClanFlake
|
from clan_cli.tests.fixtures_flakes import ClanFlake
|
||||||
from clan_cli.tests.helpers import cli
|
from clan_cli.tests.helpers import cli
|
||||||
@@ -58,13 +58,13 @@ def test_upload_secret(
|
|||||||
"""
|
"""
|
||||||
)
|
)
|
||||||
subprocess.run(
|
subprocess.run(
|
||||||
nix_shell(
|
nix_shell_legacy(
|
||||||
["nixpkgs#gnupg"], ["gpg", "--batch", "--gen-key", str(gpg_key_spec)]
|
["nixpkgs#gnupg"], ["gpg", "--batch", "--gen-key", str(gpg_key_spec)]
|
||||||
),
|
),
|
||||||
check=True,
|
check=True,
|
||||||
)
|
)
|
||||||
subprocess.run(
|
subprocess.run(
|
||||||
nix_shell(["nixpkgs#pass"], ["pass", "init", "test@local"]), check=True
|
nix_shell_legacy(["nixpkgs#pass"], ["pass", "init", "test@local"]), check=True
|
||||||
)
|
)
|
||||||
cli.run(["facts", "generate", "vm1", "--flake", str(flake.path)])
|
cli.run(["facts", "generate", "vm1", "--flake", str(flake.path)])
|
||||||
|
|
||||||
|
|||||||
@@ -17,7 +17,7 @@ from clan_cli.completions import (
|
|||||||
from clan_cli.errors import ClanError
|
from clan_cli.errors import ClanError
|
||||||
from clan_cli.git import commit_files
|
from clan_cli.git import commit_files
|
||||||
from clan_cli.machines.inventory import get_all_machines, get_selected_machines
|
from clan_cli.machines.inventory import get_all_machines, get_selected_machines
|
||||||
from clan_cli.nix import nix_config, nix_shell, nix_test_store
|
from clan_cli.nix import nix_config, nix_shell_legacy, nix_test_store
|
||||||
from clan_cli.vars._types import StoreBase
|
from clan_cli.vars._types import StoreBase
|
||||||
|
|
||||||
from .check import check_vars
|
from .check import check_vars
|
||||||
@@ -84,7 +84,7 @@ def bubblewrap_cmd(generator: str, tmpdir: Path) -> list[str]:
|
|||||||
test_store = nix_test_store()
|
test_store = nix_test_store()
|
||||||
|
|
||||||
# fmt: off
|
# fmt: off
|
||||||
return nix_shell(
|
return nix_shell_legacy(
|
||||||
[
|
[
|
||||||
"nixpkgs#bash",
|
"nixpkgs#bash",
|
||||||
"nixpkgs#bubblewrap",
|
"nixpkgs#bubblewrap",
|
||||||
|
|||||||
@@ -9,7 +9,7 @@ from tempfile import TemporaryDirectory
|
|||||||
|
|
||||||
from clan_cli.cmd import CmdOut, Log, RunOpts, run
|
from clan_cli.cmd import CmdOut, Log, RunOpts, run
|
||||||
from clan_cli.machines.machines import Machine
|
from clan_cli.machines.machines import Machine
|
||||||
from clan_cli.nix import nix_shell
|
from clan_cli.nix import nix_shell_legacy
|
||||||
from clan_cli.ssh.upload import upload
|
from clan_cli.ssh.upload import upload
|
||||||
from clan_cli.vars._types import StoreBase
|
from clan_cli.vars._types import StoreBase
|
||||||
from clan_cli.vars.generate import Generator, Var
|
from clan_cli.vars.generate import Generator, Var
|
||||||
@@ -49,7 +49,9 @@ class SecretStore(StoreBase):
|
|||||||
return Path(self.entry_prefix) / self.rel_dir(generator, name)
|
return Path(self.entry_prefix) / self.rel_dir(generator, name)
|
||||||
|
|
||||||
def _run_pass(self, *args: str, options: RunOpts | None = None) -> CmdOut:
|
def _run_pass(self, *args: str, options: RunOpts | None = None) -> CmdOut:
|
||||||
cmd = nix_shell(packages=["nixpkgs#pass"], cmd=[self._store_backend, *args])
|
cmd = nix_shell_legacy(
|
||||||
|
packages=["nixpkgs#pass"], cmd=[self._store_backend, *args]
|
||||||
|
)
|
||||||
return run(cmd, options)
|
return run(cmd, options)
|
||||||
|
|
||||||
def _set(
|
def _set(
|
||||||
@@ -90,7 +92,7 @@ class SecretStore(StoreBase):
|
|||||||
hashes = []
|
hashes = []
|
||||||
hashes.append(
|
hashes.append(
|
||||||
run(
|
run(
|
||||||
nix_shell(
|
nix_shell_legacy(
|
||||||
["nixpkgs#git"],
|
["nixpkgs#git"],
|
||||||
[
|
[
|
||||||
"git",
|
"git",
|
||||||
@@ -118,7 +120,7 @@ class SecretStore(StoreBase):
|
|||||||
if symlink.is_symlink():
|
if symlink.is_symlink():
|
||||||
hashes.append(
|
hashes.append(
|
||||||
run(
|
run(
|
||||||
nix_shell(
|
nix_shell_legacy(
|
||||||
["nixpkgs#git"],
|
["nixpkgs#git"],
|
||||||
[
|
[
|
||||||
"git",
|
"git",
|
||||||
|
|||||||
@@ -19,7 +19,7 @@ from clan_cli.dirs import module_root, user_cache_dir, vm_state_dir
|
|||||||
from clan_cli.errors import ClanCmdError, ClanError
|
from clan_cli.errors import ClanCmdError, ClanError
|
||||||
from clan_cli.facts.generate import generate_facts
|
from clan_cli.facts.generate import generate_facts
|
||||||
from clan_cli.machines.machines import Machine
|
from clan_cli.machines.machines import Machine
|
||||||
from clan_cli.nix import nix_shell
|
from clan_cli.nix import nix_shell_legacy
|
||||||
from clan_cli.qemu.qga import QgaSession
|
from clan_cli.qemu.qga import QgaSession
|
||||||
from clan_cli.qemu.qmp import QEMUMonitorProtocol
|
from clan_cli.qemu.qmp import QEMUMonitorProtocol
|
||||||
|
|
||||||
@@ -96,7 +96,7 @@ def prepare_disk(
|
|||||||
file_name: str = "disk.img",
|
file_name: str = "disk.img",
|
||||||
) -> Path:
|
) -> Path:
|
||||||
disk_img = directory / file_name
|
disk_img = directory / file_name
|
||||||
cmd = nix_shell(
|
cmd = nix_shell_legacy(
|
||||||
["nixpkgs#qemu"],
|
["nixpkgs#qemu"],
|
||||||
[
|
[
|
||||||
"qemu-img",
|
"qemu-img",
|
||||||
@@ -127,7 +127,7 @@ def start_vm(
|
|||||||
) -> Iterator[subprocess.Popen]:
|
) -> Iterator[subprocess.Popen]:
|
||||||
env = os.environ.copy()
|
env = os.environ.copy()
|
||||||
env.update(extra_env)
|
env.update(extra_env)
|
||||||
cmd = nix_shell(packages, args)
|
cmd = nix_shell_legacy(packages, args)
|
||||||
machine.debug(f"Starting VM with command: {cmd}")
|
machine.debug(f"Starting VM with command: {cmd}")
|
||||||
|
|
||||||
with subprocess.Popen(
|
with subprocess.Popen(
|
||||||
|
|||||||
@@ -6,7 +6,7 @@ from collections.abc import Iterator
|
|||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
from clan_cli.errors import ClanError
|
from clan_cli.errors import ClanError
|
||||||
from clan_cli.nix import nix_shell
|
from clan_cli.nix import nix_shell_legacy
|
||||||
|
|
||||||
|
|
||||||
@contextlib.contextmanager
|
@contextlib.contextmanager
|
||||||
@@ -14,7 +14,7 @@ def start_virtiofsd(socket_path: Path) -> Iterator[None]:
|
|||||||
sandbox = "namespace"
|
sandbox = "namespace"
|
||||||
if shutil.which("newuidmap") is None:
|
if shutil.which("newuidmap") is None:
|
||||||
sandbox = "none"
|
sandbox = "none"
|
||||||
virtiofsd = nix_shell(
|
virtiofsd = nix_shell_legacy(
|
||||||
["nixpkgs#virtiofsd"],
|
["nixpkgs#virtiofsd"],
|
||||||
[
|
[
|
||||||
"virtiofsd",
|
"virtiofsd",
|
||||||
|
|||||||
@@ -6,7 +6,7 @@ import time
|
|||||||
from collections.abc import Iterator
|
from collections.abc import Iterator
|
||||||
|
|
||||||
from clan_cli.errors import ClanError
|
from clan_cli.errors import ClanError
|
||||||
from clan_cli.nix import nix_shell
|
from clan_cli.nix import nix_shell_legacy
|
||||||
|
|
||||||
VMADDR_CID_HYPERVISOR = 2
|
VMADDR_CID_HYPERVISOR = 2
|
||||||
|
|
||||||
@@ -29,7 +29,7 @@ def start_waypipe(cid: int | None, title_prefix: str) -> Iterator[None]:
|
|||||||
if cid is None:
|
if cid is None:
|
||||||
yield
|
yield
|
||||||
return
|
return
|
||||||
waypipe = nix_shell(
|
waypipe = nix_shell_legacy(
|
||||||
["nixpkgs#waypipe"],
|
["nixpkgs#waypipe"],
|
||||||
[
|
[
|
||||||
"waypipe",
|
"waypipe",
|
||||||
|
|||||||
@@ -37,7 +37,7 @@ let
|
|||||||
# load nixpkgs runtime dependencies from a json file
|
# load nixpkgs runtime dependencies from a json file
|
||||||
# This file represents an allow list at the same time that is checked by the run_cmd
|
# This file represents an allow list at the same time that is checked by the run_cmd
|
||||||
# implementation in nix.py
|
# implementation in nix.py
|
||||||
allDependencies = lib.importJSON ./clan_cli/nix/allowed-programs.json;
|
allDependencies = lib.importJSON ./clan_cli/nix/allowed-packages.json;
|
||||||
generateRuntimeDependenciesMap =
|
generateRuntimeDependenciesMap =
|
||||||
deps:
|
deps:
|
||||||
lib.filterAttrs (_: pkg: !pkg.meta.unsupported or false) (lib.genAttrs deps (name: pkgs.${name}));
|
lib.filterAttrs (_: pkg: !pkg.meta.unsupported or false) (lib.genAttrs deps (name: pkgs.${name}));
|
||||||
@@ -109,7 +109,7 @@ pythonRuntime.pkgs.buildPythonApplication {
|
|||||||
clan-core-path
|
clan-core-path
|
||||||
|
|
||||||
"--set"
|
"--set"
|
||||||
"CLAN_STATIC_PROGRAMS"
|
"CLAN_PROVIDED_PACKAGES"
|
||||||
(lib.concatStringsSep ":" (lib.attrNames bundledRuntimeDependenciesMap))
|
(lib.concatStringsSep ":" (lib.attrNames bundledRuntimeDependenciesMap))
|
||||||
];
|
];
|
||||||
|
|
||||||
|
|||||||
@@ -146,7 +146,7 @@
|
|||||||
clan-core-path = clanCoreWithVendoredDeps;
|
clan-core-path = clanCoreWithVendoredDeps;
|
||||||
templateDerivation = templateDerivation;
|
templateDerivation = templateDerivation;
|
||||||
pythonRuntime = pkgs.python3;
|
pythonRuntime = pkgs.python3;
|
||||||
includedRuntimeDeps = lib.importJSON ./clan_cli/nix/allowed-programs.json;
|
includedRuntimeDeps = lib.importJSON ./clan_cli/nix/allowed-packages.json;
|
||||||
};
|
};
|
||||||
clan-cli-docs = pkgs.stdenv.mkDerivation {
|
clan-cli-docs = pkgs.stdenv.mkDerivation {
|
||||||
name = "clan-cli-docs";
|
name = "clan-cli-docs";
|
||||||
|
|||||||
@@ -19,7 +19,7 @@ exclude = ["clan_cli.nixpkgs*", "result"]
|
|||||||
|
|
||||||
[tool.setuptools.package-data]
|
[tool.setuptools.package-data]
|
||||||
clan_cli = [
|
clan_cli = [
|
||||||
"**/allowed-programs.json",
|
"**/allowed-packages.json",
|
||||||
"py.typed",
|
"py.typed",
|
||||||
"templates/**/*",
|
"templates/**/*",
|
||||||
"vms/mimetypes/**/*",
|
"vms/mimetypes/**/*",
|
||||||
|
|||||||
@@ -25,7 +25,7 @@ mkShell {
|
|||||||
|
|
||||||
inputsFrom = [ self'.devShells.default ];
|
inputsFrom = [ self'.devShells.default ];
|
||||||
|
|
||||||
CLAN_STATIC_PROGRAMS = lib.concatStringsSep ":" (
|
CLAN_PROVIDED_PACKAGES = lib.concatStringsSep ":" (
|
||||||
lib.attrNames clan-cli-full.passthru.runtimeDependenciesMap
|
lib.attrNames clan-cli-full.passthru.runtimeDependenciesMap
|
||||||
);
|
);
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user