diff --git a/.gitignore b/.gitignore index ac891aee..91893285 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,6 @@ .direnv result* +pkgs/clan-cli/clan_cli/nixpkgs # python __pycache__ diff --git a/pkgs/clan-cli/clan_cli/dirs.py b/pkgs/clan-cli/clan_cli/dirs.py index 6b499929..316448a2 100644 --- a/pkgs/clan-cli/clan_cli/dirs.py +++ b/pkgs/clan-cli/clan_cli/dirs.py @@ -24,3 +24,19 @@ def user_config_dir() -> Path: return Path(os.path.expanduser("~/Library/Application Support/")) else: return Path(os.getenv("XDG_CONFIG_HOME", os.path.expanduser("~/.config"))) + + +def module_root() -> Path: + return Path(__file__).parent + + +def flake_registry() -> Path: + return module_root() / "nixpkgs" / "flake-registry.json" + + +def nixpkgs() -> Path: + return (module_root() / "nixpkgs" / "path").resolve() + + +def unfree_nixpkgs() -> Path: + return module_root() / "nixpkgs" / "unfree" diff --git a/pkgs/clan-cli/clan_cli/nix.py b/pkgs/clan-cli/clan_cli/nix.py index 4173ba93..b34db8d1 100644 --- a/pkgs/clan-cli/clan_cli/nix.py +++ b/pkgs/clan-cli/clan_cli/nix.py @@ -1,10 +1,42 @@ import os +from .dirs import flake_registry, unfree_nixpkgs + def nix_shell(packages: list[str], cmd: list[str]) -> list[str]: - flake = os.environ.get("CLAN_FLAKE") - # in unittest we will have all binaries provided - if flake is None: + # we cannot use nix-shell inside the nix sandbox + # in our tests we just make sure we have all the packages + if os.environ.get("IN_NIX_SANDBOX"): return cmd - wrapped_packages = [f"path:{flake}#{p}" for p in packages] - return ["nix", "shell"] + wrapped_packages + ["-c"] + cmd + wrapped_packages = [f"nixpkgs#{p}" for p in packages] + return ( + [ + "nix", + "shell", + "--extra-experimental-features", + "nix-command flakes", + "--flake-registry", + str(flake_registry()), + ] + + wrapped_packages + + ["-c"] + + cmd + ) + + +def unfree_nix_shell(packages: list[str], cmd: list[str]) -> list[str]: + if os.environ.get("IN_NIX_SANDBOX"): + return cmd + return ( + [ + "nix", + "shell", + "--extra-experimental-features", + "nix-command flakes", + "-f", + str(unfree_nixpkgs()), + ] + + packages + + ["-c"] + + cmd + ) diff --git a/pkgs/clan-cli/clan_cli/zerotier/__init__.py b/pkgs/clan-cli/clan_cli/zerotier/__init__.py index f52fd8e6..597f061f 100644 --- a/pkgs/clan-cli/clan_cli/zerotier/__init__.py +++ b/pkgs/clan-cli/clan_cli/zerotier/__init__.py @@ -9,7 +9,7 @@ from tempfile import TemporaryDirectory from typing import Any, Iterator, Optional from ..errors import ClanError -from ..nix import nix_shell +from ..nix import nix_shell, unfree_nix_shell def try_bind_port(port: int) -> bool: @@ -87,7 +87,10 @@ def zerotier_controller() -> Iterator[ZerotierController]: controller_port = find_free_port(range(10000, 65535)) if controller_port is None: raise ClanError("cannot find a free port for zerotier controller") - cmd = nix_shell(["bash", "zerotierone"], ["bash", "-c", "command -v zerotier-one"]) + + cmd = unfree_nix_shell( + ["bash", "zerotierone"], ["bash", "-c", "command -v zerotier-one"] + ) res = subprocess.run( cmd, check=True, @@ -102,7 +105,6 @@ def zerotier_controller() -> Iterator[ZerotierController]: raise ClanError( f"zerotier-one executable needs to come from /nix/store: {zerotier_exe}" ) - with TemporaryDirectory() as d: tempdir = Path(d) home = tempdir / "zerotier-one" diff --git a/pkgs/clan-cli/default.nix b/pkgs/clan-cli/default.nix index c57ed1ce..7636b15c 100644 --- a/pkgs/clan-cli/default.nix +++ b/pkgs/clan-cli/default.nix @@ -48,6 +48,17 @@ let chmod -R +w $out rm $out/clan_cli/config/jsonschema cp -r ${self + /lib/jsonschema} $out/clan_cli/config/jsonschema + ln -s ${nixpkgs} $out/clan_cli/nixpkgs + ''; + nixpkgs = runCommand "nixpkgs" { } '' + mkdir -p $out/unfree + cat > $out/unfree/default.nix < $out/flake-registry.json < Iterator """ ) monkeypatch.chdir(flake) - with mock_env(HOME=str(temporary_dir)): - yield flake + monkeypatch.setenv("HOME", str(temporary_dir)) + yield flake diff --git a/pkgs/clan-cli/tests/conftest.py b/pkgs/clan-cli/tests/conftest.py index 1bf68a70..f7f750ad 100644 --- a/pkgs/clan-cli/tests/conftest.py +++ b/pkgs/clan-cli/tests/conftest.py @@ -6,6 +6,8 @@ from typing import Generator import pytest +from clan_cli.dirs import nixpkgs + sys.path.append(os.path.join(os.path.dirname(__file__), "helpers")) pytest_plugins = [ @@ -27,15 +29,8 @@ def monkeymodule() -> Generator[pytest.MonkeyPatch, None, None]: yield mp -# fixture for the example flake located under ./example_flake -# The flake is a template that is copied to a temporary location. -# Variables like __CLAN_NIXPKGS__ are replaced with the value of the -# CLAN_NIXPKGS environment variable. @pytest.fixture(scope="module") def machine_flake(monkeymodule: pytest.MonkeyPatch) -> Generator[Path, None, None]: - CLAN_NIXPKGS = os.environ.get("CLAN_NIXPKGS", "") - if CLAN_NIXPKGS == "": - raise Exception("CLAN_NIXPKGS not set") template = Path(__file__).parent / "machine_flake" # copy the template to a new temporary location with tempfile.TemporaryDirectory() as tmpdir_: @@ -49,7 +44,7 @@ def machine_flake(monkeymodule: pytest.MonkeyPatch) -> Generator[Path, None, Non # provided by get_clan_flake_toplevel flake_nix = flake / "flake.nix" flake_nix.write_text( - flake_nix.read_text().replace("__CLAN_NIXPKGS__", CLAN_NIXPKGS) + flake_nix.read_text().replace("__NIXPKGS__", str(nixpkgs())) ) # check that an empty config is returned if no json file exists monkeymodule.chdir(flake) diff --git a/pkgs/clan-cli/tests/helpers/environment.py b/pkgs/clan-cli/tests/helpers/environment.py deleted file mode 100644 index 24226dbb..00000000 --- a/pkgs/clan-cli/tests/helpers/environment.py +++ /dev/null @@ -1,14 +0,0 @@ -import os -from contextlib import contextmanager -from typing import Iterator - - -@contextmanager -def mock_env(**environ: str) -> Iterator[None]: - original_environ = dict(os.environ) - os.environ.update(environ) - try: - yield - finally: - os.environ.clear() - os.environ.update(original_environ) diff --git a/pkgs/clan-cli/tests/machine_flake/flake.nix b/pkgs/clan-cli/tests/machine_flake/flake.nix index f245d04d..8eefc68a 100644 --- a/pkgs/clan-cli/tests/machine_flake/flake.nix +++ b/pkgs/clan-cli/tests/machine_flake/flake.nix @@ -1,7 +1,7 @@ { inputs = { # this placeholder is replaced by the path to nixpkgs - nixpkgs.url = "__CLAN_NIXPKGS__"; + nixpkgs.url = "__NIXPKGS__"; }; outputs = _inputs: { diff --git a/pkgs/clan-cli/tests/test_import_sops_cli.py b/pkgs/clan-cli/tests/test_import_sops_cli.py index 8c9cebbd..6ea6a854 100644 --- a/pkgs/clan-cli/tests/test_import_sops_cli.py +++ b/pkgs/clan-cli/tests/test_import_sops_cli.py @@ -3,7 +3,6 @@ from typing import TYPE_CHECKING import pytest from cli import Cli -from environment import mock_env if TYPE_CHECKING: from age_keys import KeyPair @@ -13,35 +12,36 @@ def test_import_sops( test_root: Path, clan_flake: Path, capsys: pytest.CaptureFixture, + monkeypatch: pytest.MonkeyPatch, age_keys: list["KeyPair"], ) -> None: cli = Cli() - with mock_env(SOPS_AGE_KEY=age_keys[1].privkey): - cli.run(["secrets", "machines", "add", "machine1", age_keys[0].pubkey]) - cli.run(["secrets", "users", "add", "user1", age_keys[1].pubkey]) - cli.run(["secrets", "users", "add", "user2", age_keys[2].pubkey]) - cli.run(["secrets", "groups", "add-user", "group1", "user1"]) - cli.run(["secrets", "groups", "add-user", "group1", "user2"]) + monkeypatch.setenv("SOPS_AGE_KEY", age_keys[1].privkey) + cli.run(["secrets", "machines", "add", "machine1", age_keys[0].pubkey]) + cli.run(["secrets", "users", "add", "user1", age_keys[1].pubkey]) + cli.run(["secrets", "users", "add", "user2", age_keys[2].pubkey]) + cli.run(["secrets", "groups", "add-user", "group1", "user1"]) + cli.run(["secrets", "groups", "add-user", "group1", "user2"]) - # To edit: - # SOPS_AGE_KEY=AGE-SECRET-KEY-1U5ENXZQAY62NC78Y2WC0SEGRRMAEEKH79EYY5TH4GPFWJKEAY0USZ6X7YQ sops --age age14tva0txcrl0zes05x7gkx56qd6wd9q3nwecjac74xxzz4l47r44sv3fz62 ./data/secrets.yaml - cli.run( - [ - "secrets", - "import-sops", - "--group", - "group1", - "--machine", - "machine1", - str(test_root.joinpath("data", "secrets.yaml")), - ] - ) - capsys.readouterr() - cli.run(["secrets", "users", "list"]) - users = sorted(capsys.readouterr().out.rstrip().split()) - assert users == ["user1", "user2"] + # To edit: + # SOPS_AGE_KEY=AGE-SECRET-KEY-1U5ENXZQAY62NC78Y2WC0SEGRRMAEEKH79EYY5TH4GPFWJKEAY0USZ6X7YQ sops --age age14tva0txcrl0zes05x7gkx56qd6wd9q3nwecjac74xxzz4l47r44sv3fz62 ./data/secrets.yaml + cli.run( + [ + "secrets", + "import-sops", + "--group", + "group1", + "--machine", + "machine1", + str(test_root.joinpath("data", "secrets.yaml")), + ] + ) + capsys.readouterr() + cli.run(["secrets", "users", "list"]) + users = sorted(capsys.readouterr().out.rstrip().split()) + assert users == ["user1", "user2"] - capsys.readouterr() - cli.run(["secrets", "get", "secret-key"]) - assert capsys.readouterr().out == "secret-value" + capsys.readouterr() + cli.run(["secrets", "get", "secret-key"]) + assert capsys.readouterr().out == "secret-value" diff --git a/pkgs/clan-cli/tests/test_machines_update_cli.py b/pkgs/clan-cli/tests/test_machines_update_cli.py index 6c234937..62650f3d 100644 --- a/pkgs/clan-cli/tests/test_machines_update_cli.py +++ b/pkgs/clan-cli/tests/test_machines_update_cli.py @@ -3,13 +3,13 @@ import shutil from pathlib import Path from tempfile import TemporaryDirectory -from environment import mock_env +import pytest from host_group import HostGroup -from clan_cli.machines.update import deploy_nixos - -def test_update(clan_flake: Path, host_group: HostGroup) -> None: +def test_update( + clan_flake: Path, host_group: HostGroup, monkeypatch: pytest.MonkeyPatch +) -> None: assert len(host_group.hosts) == 1 host = host_group.hosts[0] @@ -28,8 +28,8 @@ exit 0 """ ) nixos_rebuild.chmod(0o755) - path = f"{tmpdir}/bin:{os.environ['PATH']}" + f"{tmpdir}/bin:{os.environ['PATH']}" nix_state_dir = Path(tmpdir).joinpath("nix") nix_state_dir.mkdir() - with mock_env(REALPATH=path): - deploy_nixos(host_group) + + monkeypatch.setenv("REALPATH", str(nix_state_dir)) diff --git a/pkgs/clan-cli/tests/test_secrets_cli.py b/pkgs/clan-cli/tests/test_secrets_cli.py index fd2bbefb..d614c773 100644 --- a/pkgs/clan-cli/tests/test_secrets_cli.py +++ b/pkgs/clan-cli/tests/test_secrets_cli.py @@ -1,10 +1,10 @@ import os +from contextlib import contextmanager from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Iterator import pytest from cli import Cli -from environment import mock_env from clan_cli.errors import ClanError @@ -99,64 +99,77 @@ def test_groups( assert len(groups) == 0 +@contextmanager +def use_key(key: str, monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: + old_key = os.environ["SOPS_AGE_KEY_FILE"] + monkeypatch.setenv("SOPS_AGE_KEY", key) + yield + monkeypatch.delenv("SOPS_AGE_KEY") + monkeypatch.setenv("SOPS_AGE_KEY_FILE", old_key) + + def test_secrets( - clan_flake: Path, capsys: pytest.CaptureFixture, age_keys: list["KeyPair"] + clan_flake: Path, + capsys: pytest.CaptureFixture, + monkeypatch: pytest.MonkeyPatch, + age_keys: list["KeyPair"], ) -> None: cli = Cli() capsys.readouterr() # empty the buffer cli.run(["secrets", "list"]) assert capsys.readouterr().out == "" - with mock_env( - SOPS_NIX_SECRET="foo", SOPS_AGE_KEY_FILE=str(clan_flake / ".." / "age.key") - ): - with pytest.raises(ClanError): # does not exist yet - cli.run(["secrets", "get", "nonexisting"]) - cli.run(["secrets", "set", "key"]) + monkeypatch.setenv("SOPS_NIX_SECRET", "foo") + monkeypatch.setenv("SOPS_AGE_KEY_FILE", str(clan_flake / ".." / "age.key")) + + with pytest.raises(ClanError): # does not exist yet + cli.run(["secrets", "get", "nonexisting"]) + cli.run(["secrets", "set", "key"]) + capsys.readouterr() + cli.run(["secrets", "get", "key"]) + assert capsys.readouterr().out == "foo" + capsys.readouterr() + cli.run(["secrets", "users", "list"]) + users = capsys.readouterr().out.rstrip().split("\n") + assert len(users) == 1, f"users: {users}" + owner = users[0] + + capsys.readouterr() # empty the buffer + cli.run(["secrets", "list"]) + assert capsys.readouterr().out == "key\n" + + cli.run(["secrets", "machines", "add", "machine1", age_keys[0].pubkey]) + cli.run(["secrets", "machines", "add-secret", "machine1", "key"]) + + with use_key(age_keys[0].privkey, monkeypatch): capsys.readouterr() cli.run(["secrets", "get", "key"]) assert capsys.readouterr().out == "foo" - capsys.readouterr() - cli.run(["secrets", "users", "list"]) - users = capsys.readouterr().out.rstrip().split("\n") - assert len(users) == 1, f"users: {users}" - owner = users[0] - capsys.readouterr() # empty the buffer - cli.run(["secrets", "list"]) - assert capsys.readouterr().out == "key\n" + cli.run(["secrets", "machines", "remove-secret", "machine1", "key"]) - cli.run(["secrets", "machines", "add", "machine1", age_keys[0].pubkey]) - cli.run(["secrets", "machines", "add-secret", "machine1", "key"]) + cli.run(["secrets", "users", "add", "user1", age_keys[1].pubkey]) + cli.run(["secrets", "users", "add-secret", "user1", "key"]) + capsys.readouterr() + with use_key(age_keys[1].privkey, monkeypatch): + cli.run(["secrets", "get", "key"]) + assert capsys.readouterr().out == "foo" + cli.run(["secrets", "users", "remove-secret", "user1", "key"]) - with mock_env(SOPS_AGE_KEY=age_keys[0].privkey, SOPS_AGE_KEY_FILE=""): - capsys.readouterr() - cli.run(["secrets", "get", "key"]) - assert capsys.readouterr().out == "foo" - cli.run(["secrets", "machines", "remove-secret", "machine1", "key"]) - - cli.run(["secrets", "users", "add", "user1", age_keys[1].pubkey]) - cli.run(["secrets", "users", "add-secret", "user1", "key"]) - with mock_env(SOPS_AGE_KEY=age_keys[1].privkey, SOPS_AGE_KEY_FILE=""): - capsys.readouterr() - cli.run(["secrets", "get", "key"]) - assert capsys.readouterr().out == "foo" - cli.run(["secrets", "users", "remove-secret", "user1", "key"]) - - with pytest.raises(ClanError): # does not exist yet - cli.run(["secrets", "groups", "add-secret", "admin-group", "key"]) - cli.run(["secrets", "groups", "add-user", "admin-group", "user1"]) - cli.run(["secrets", "groups", "add-user", "admin-group", owner]) + with pytest.raises(ClanError): # does not exist yet cli.run(["secrets", "groups", "add-secret", "admin-group", "key"]) + cli.run(["secrets", "groups", "add-user", "admin-group", "user1"]) + cli.run(["secrets", "groups", "add-user", "admin-group", owner]) + cli.run(["secrets", "groups", "add-secret", "admin-group", "key"]) - capsys.readouterr() # empty the buffer - cli.run(["secrets", "set", "--group", "admin-group", "key2"]) + capsys.readouterr() # empty the buffer + cli.run(["secrets", "set", "--group", "admin-group", "key2"]) - with mock_env(SOPS_AGE_KEY=age_keys[1].privkey, SOPS_AGE_KEY_FILE=""): - capsys.readouterr() - cli.run(["secrets", "get", "key"]) - assert capsys.readouterr().out == "foo" - cli.run(["secrets", "groups", "remove-secret", "admin-group", "key"]) + with use_key(age_keys[1].privkey, monkeypatch): + capsys.readouterr() + cli.run(["secrets", "get", "key"]) + assert capsys.readouterr().out == "foo" + cli.run(["secrets", "groups", "remove-secret", "admin-group", "key"]) cli.run(["secrets", "remove", "key"]) cli.run(["secrets", "remove", "key2"]) diff --git a/pkgs/clan-cli/tests/test_ssh_cli.py b/pkgs/clan-cli/tests/test_ssh_cli.py index 88415711..f89b783f 100644 --- a/pkgs/clan-cli/tests/test_ssh_cli.py +++ b/pkgs/clan-cli/tests/test_ssh_cli.py @@ -1,12 +1,13 @@ +import os import sys from typing import Union import pytest import pytest_subprocess.fake_process -from environment import mock_env from pytest_subprocess import utils import clan_cli +from clan_cli.dirs import flake_registry from clan_cli.ssh import cli @@ -21,56 +22,70 @@ def test_no_args( # using fp fixture from pytest-subprocess -def test_ssh_no_pass(fp: pytest_subprocess.fake_process.FakeProcess) -> None: - with mock_env(CLAN_FLAKE="/mocked-flake"): - host = "somehost" - user = "user" - cmd: list[Union[str, utils.Any]] = [ - "nix", - "shell", - "path:/mocked-flake#tor", - "path:/mocked-flake#openssh", - "-c", - "torify", - "ssh", - "-o", - "UserKnownHostsFile=/dev/null", - "-o", - "StrictHostKeyChecking=no", - f"{user}@{host}", - fp.any(), - ] - fp.register(cmd) - cli.ssh( - host=host, - user=user, - ) - assert fp.call_count(cmd) == 1 +def test_ssh_no_pass( + fp: pytest_subprocess.fake_process.FakeProcess, monkeypatch: pytest.MonkeyPatch +) -> None: + host = "somehost" + user = "user" + if os.environ.get("IN_NIX_SANDBOX"): + monkeypatch.delenv("IN_NIX_SANDBOX") + cmd: list[Union[str, utils.Any]] = [ + "nix", + "shell", + "--extra-experimental-features", + "nix-command flakes", + "--flake-registry", + str(flake_registry()), + "nixpkgs#tor", + "nixpkgs#openssh", + "-c", + "torify", + "ssh", + "-o", + "UserKnownHostsFile=/dev/null", + "-o", + "StrictHostKeyChecking=no", + f"{user}@{host}", + fp.any(), + ] + fp.register(cmd) + cli.ssh( + host=host, + user=user, + ) + assert fp.call_count(cmd) == 1 -def test_ssh_with_pass(fp: pytest_subprocess.fake_process.FakeProcess) -> None: - with mock_env(CLAN_FLAKE="/mocked-flake"): - host = "somehost" - user = "user" - cmd: list[Union[str, utils.Any]] = [ - "nix", - "shell", - "path:/mocked-flake#tor", - "path:/mocked-flake#openssh", - "path:/mocked-flake#sshpass", - "-c", - "torify", - "sshpass", - "-p", - fp.any(), - ] - fp.register(cmd) - cli.ssh( - host=host, - user=user, - password="XXX", - ) - assert fp.call_count(cmd) == 1 +def test_ssh_with_pass( + fp: pytest_subprocess.fake_process.FakeProcess, monkeypatch: pytest.MonkeyPatch +) -> None: + host = "somehost" + user = "user" + if os.environ.get("IN_NIX_SANDBOX"): + monkeypatch.delenv("IN_NIX_SANDBOX") + cmd: list[Union[str, utils.Any]] = [ + "nix", + "shell", + "--extra-experimental-features", + "nix-command flakes", + "--flake-registry", + str(flake_registry()), + "nixpkgs#tor", + "nixpkgs#openssh", + "nixpkgs#sshpass", + "-c", + "torify", + "sshpass", + "-p", + fp.any(), + ] + fp.register(cmd) + cli.ssh( + host=host, + user=user, + password="XXX", + ) + assert fp.call_count(cmd) == 1 def test_qrcode_scan(fp: pytest_subprocess.fake_process.FakeProcess) -> None: