althaea.zone
ALPHA: wire is a tool to deploy nixos systems
wire.althaea.zone/
1# SPDX-License-Identifier: AGPL-3.0-or-later
2# Copyright 2024-2025 wire Contributors
3
4from typing import TYPE_CHECKING
5
6if TYPE_CHECKING:
7 from test_driver.machine import Machine
8 from tools import collect_store_objects, assert_store_not_poisoned
9
10 deployer: Machine = None # type: ignore[invalid-assignment]
11 receiver: Machine = None # type: ignore[invalid-assignment]
12 TEST_DIR = ""
13
14# typing-end
15
16deployer_so = collect_store_objects(deployer)
17receiver_so = collect_store_objects(receiver)
18
19# build receiver with no keys
20deployer.succeed(
21 f"wire apply --no-progress --on receiver --path {TEST_DIR}/hive.nix --no-keys --ssh-accept-host -vvv >&2"
22)
23
24receiver.wait_for_unit("sshd.service")
25
26# --no-keys should never push a key
27receiver.fail("test -f /run/keys/source_string_name")
28deployer.fail("test -f /run/keys/source_string_name")
29
30# key services are created
31receiver.succeed("systemctl cat source_string_name-key.service")
32
33_, is_failed = receiver.execute("systemctl is-failed source_string_name-key.service")
34assert is_failed == "inactive\n", (
35 f"source_string_name-key.service must be inactive before key exists ({is_failed})"
36)
37
38
39def test_keys(target, target_object, non_interactive):
40 if non_interactive:
41 deployer.succeed(
42 f"wire apply keys --on {target} --no-progress --path {TEST_DIR}/hive.nix --non-interactive --ssh-accept-host -vvv >&2"
43 )
44 else:
45 deployer.succeed(
46 f"wire apply keys --on {target} --no-progress --path {TEST_DIR}/hive.nix --ssh-accept-host -vvv >&2"
47 )
48
49 keys = [
50 (
51 "/run/keys/source_string_name",
52 "hello_world_source",
53 "root root 600",
54 "source_string_name",
55 ),
56 ("/etc/keys/file", "hello_world_file", "root root 644", "file"),
57 (
58 "/home/owner/some/deep/path/command",
59 "hello_world_command",
60 "owner owner 644",
61 "command",
62 ),
63 (
64 "/run/keys/environment",
65 "string_from_environment",
66 "root root 600",
67 "environment",
68 ),
69 ]
70
71 for path, value, permissions, name in keys:
72 # test existence & value
73 source_string = target_object.succeed(f"cat {path}")
74 assert value in source_string, f"{path} has correct contents ({target})"
75
76 stat = target_object.succeed(f"stat -c '%U %G %a' {path}").rstrip()
77 assert permissions == stat, f"{path} has correct permissions ({target})"
78
79
80def perform_routine(target, target_object, non_interactive):
81 test_keys(target, target_object, non_interactive)
82
83 # only check systemd units on receiver since deployer apply's are one time only
84 if target == "receiver":
85 target_object.succeed("systemctl start source_string_name-key.path")
86 target_object.succeed("systemctl start command-key.path")
87 target_object.wait_for_unit("source_string_name-key.service")
88 target_object.wait_for_unit("command-key.service")
89
90 # Mess with the keys to make sure that every push refreshes the permissions
91 target_object.succeed("echo 'incorrect_value' > /run/keys/source_string")
92 target_object.succeed("chown 600 /etc/keys/file")
93 # Test having a key that doesn't exist mixed with keys that do
94 target_object.succeed("rm /home/owner/some/deep/path/command")
95
96 if target == "receiver":
97 _, is_failed = target_object.execute("systemctl is-active command-key.service")
98 assert is_failed == "failed\n", (
99 f"command-key.service is failed after deletion ({is_failed})"
100 )
101
102 # Test keys twice to ensure the operation is idempotent,
103 # especially around directory creation.
104 test_keys(target, target_object, non_interactive)
105
106
107perform_routine("receiver", receiver, True)
108perform_routine("deployer", deployer, True)
109perform_routine("receiver", receiver, False)
110perform_routine("deployer", deployer, False)
111
112new_deployer_store_objects = collect_store_objects(deployer).difference(deployer_so)
113new_receiver_store_objects = collect_store_objects(receiver).difference(receiver_so)
114
115# no one should have any keys introduced by the operation
116for node, objects in [
117 (deployer, new_deployer_store_objects),
118 (receiver, new_receiver_store_objects),
119]:
120 assert_store_not_poisoned(node, "hello_world_source", objects)
121 assert_store_not_poisoned(node, "hello_world_file", objects)
122 assert_store_not_poisoned(node, "hello_world_command", objects)
123 assert_store_not_poisoned(node, "string_from_environment", objects)