1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
|
"""Integration test for the wireguard module."""
import pytest
from pycloudlib.lxd.instance import LXDInstance
from cloudinit.subp import subp
from tests.integration_tests.instances import IntegrationInstance
ASCII_TEXT = "ASCII text"
USER_DATA = """\
#cloud-config
wireguard:
interfaces:
- name: wg0
config_path: /etc/wireguard/wg0.conf
content: |
[Interface]
Address = 192.168.254.1/32
ListenPort = 51820
PrivateKey = iNlmgtGo6yiFhD9TuVnx/qJSp+C5Cwg4wwPmOJwlZXI=
[Peer]
PublicKey = 6PewunPjxlUq/0xvbVxklN2p73YIytfjxpoIEohCukY=
AllowedIPs = 192.168.254.2/32
- name: wg1
config_path: /etc/wireguard/wg1.conf
content: |
[Interface]
PrivateKey = GGLU4+5vIcK9lGyfz4AJn9fR5/FN/6sf4Fd5chZ16Vc=
Address = 192.168.254.2/24
[Peer]
PublicKey = 2as8z3EDjSsfFEkvOQGVnJ1Hv+h1jRAh2BKJg+DHvGk=
Endpoint = 127.0.0.1:51820
AllowedIPs = 0.0.0.0/0
readinessprobe:
- ping -qc 5 192.168.254.1 2>&1 > /dev/null
- echo $? > /tmp/ping
# wg-quick configures the system interfaces and routes, but we need to ssh in
# stop the service at the end of cloud-init
runcmd:
- [systemctl, stop, wg-quick@wg0.service]
- [systemctl, stop, wg-quick@wg1.service]
"""
def load_wireguard_kernel_module_lxd(instance: LXDInstance):
subp(
"lxc config set {} linux.kernel_modules wireguard".format(
instance.name
).split()
)
@pytest.mark.ci
@pytest.mark.user_data(USER_DATA)
@pytest.mark.lxd_vm
@pytest.mark.gce
@pytest.mark.ec2
@pytest.mark.azure
@pytest.mark.openstack
@pytest.mark.oci
@pytest.mark.ubuntu
class TestWireguard:
@pytest.mark.parametrize(
"cmd,expected_out",
(
# check if wireguard module is loaded
("lsmod | grep '^wireguard' | awk '{print $1}'", "wireguard"),
# test if file was written for wg0
(
"stat -c '%N' /etc/wireguard/wg0.conf",
r"'/etc/wireguard/wg0.conf'",
),
# check permissions for wg0
("stat -c '%U %a' /etc/wireguard/wg0.conf", r"root 600"),
# ASCII check wg1
("file /etc/wireguard/wg1.conf", ASCII_TEXT),
# md5sum check wg1
(
"md5sum </etc/wireguard/wg1.conf",
"cff31c9879da0967313d3f561aed766b",
),
# sha256sum check
(
"sha256sum </etc/wireguard/wg1.conf",
"8443055d1442d051588beb03f7895b58"
"269196eb9916617969dc5220c1a90d54",
),
# check if systemd started wg0
(
"systemctl is-failed wg-quick@wg0; test $? -eq 1",
"inactive",
),
# check if systemd started wg1
(
"systemctl is-failed wg-quick@wg1; test $? -eq 1",
"inactive",
),
# check readiness probe (ping wg0)
("cat /tmp/ping", "0"),
),
)
def test_wireguard(
self, cmd, expected_out, class_client: IntegrationInstance
):
result = class_client.execute(cmd)
assert result.ok
assert expected_out in result.stdout
def test_wireguard_tools_installed(
self, class_client: IntegrationInstance
):
"""Test that 'wg version' succeeds, indicating installation."""
assert class_client.execute("wg version").ok
@pytest.mark.ci
@pytest.mark.user_data(USER_DATA)
@pytest.mark.lxd_setup.with_args(load_wireguard_kernel_module_lxd)
@pytest.mark.lxd_container
@pytest.mark.ubuntu
class TestWireguardWithoutKmod:
def test_wireguard_tools_installed(
self, class_client: IntegrationInstance
):
"""Test that 'wg version' succeeds, indicating installation."""
assert class_client.execute("wg version").ok
|
