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
|
from socket import AF_INET, AF_INET6
from pr2test.marks import require_root
pytestmark = [require_root()]
def test_peer_ipv4(context):
ifname = context.new_ifname
ipaddr = context.new_ipaddr
port = 9999
listen = 2525
peer_ip_1 = context.new_ipaddr
peer_ip_2 = context.new_ipaddr
allowed_ip_1 = str(context.ipnets[1])
allowed_ip_2 = str(context.ipnets[2])
(
context.ndb.interfaces.create(ifname=ifname, kind='wireguard')
.add_ip(f'{ipaddr}/24')
.set('state', 'up')
.commit()
)
peer_1 = {
'public_key': 'TGFHcm9zc2VCaWNoZV9DJ2VzdExhUGx1c0JlbGxlPDM=',
'endpoint_addr': peer_ip_1,
'endpoint_port': port,
'persistent_keepalive': 15,
'allowed_ips': [f'{allowed_ip_1}'],
}
peer_2 = {
'public_key': 'AGFHcm9zc2VCaWNoZV9DJ2VzdExhUGx1c0JlbGxlPDM=',
'endpoint_addr': peer_ip_2,
'endpoint_port': port,
'persistent_keepalive': 15,
'allowed_ips': [f'{allowed_ip_2}'],
}
(
context.wg.set(
ifname,
private_key='RCdhcHJlc0JpY2hlLEplU2VyYWlzTGFQbHVzQm9ubmU=',
fwmark=0x1337,
listen_port=listen,
peer=peer_1,
)
)
(
context.wg.set(
ifname,
private_key='RCdhcHJlc0JpY2hlLEplU2VyYWlzTGFQbHVzQm9ubmU=',
fwmark=0x1337,
listen_port=listen,
peer=peer_2,
)
)
for peer in context.wg.info(ifname)[0].get_attr('WGDEVICE_A_PEERS'):
endpoint = peer.get_attr('WGPEER_A_ENDPOINT')
allowed = peer.get_attr('WGPEER_A_ALLOWEDIPS')
assert endpoint['family'] == AF_INET
assert endpoint['port'] == port
assert endpoint['addr'] in (peer_ip_1, peer_ip_2)
assert allowed[0]['addr'] in (allowed_ip_1, allowed_ip_2)
def test_peer_ipv6(context):
ifname = context.new_ifname
ipaddr = context.new_ipaddr
port = 9999
listen = 2525
peer_ip_1 = '::fa'
peer_ip_2 = '::fb'
allowed_ip_1 = 'fa::/64'
allowed_ip_2 = 'fb::/64'
(
context.ndb.interfaces.create(ifname=ifname, kind='wireguard')
.add_ip(f'{ipaddr}/24')
.set('state', 'up')
.commit()
)
peer_1 = {
'public_key': 'TGFHcm9zc2VCaWNoZV9DJ2VzdExhUGx1c0JlbGxlPDM=',
'endpoint_addr': peer_ip_1,
'endpoint_port': port,
'persistent_keepalive': 15,
'allowed_ips': [f'{allowed_ip_1}'],
}
peer_2 = {
'public_key': 'AGFHcm9zc2VCaWNoZV9DJ2VzdExhUGx1c0JlbGxlPDM=',
'endpoint_addr': peer_ip_2,
'endpoint_port': port,
'persistent_keepalive': 15,
'allowed_ips': [f'{allowed_ip_2}'],
}
(
context.wg.set(
ifname,
private_key='RCdhcHJlc0JpY2hlLEplU2VyYWlzTGFQbHVzQm9ubmU=',
fwmark=0x1337,
listen_port=listen,
peer=peer_1,
)
)
(
context.wg.set(
ifname,
private_key='RCdhcHJlc0JpY2hlLEplU2VyYWlzTGFQbHVzQm9ubmU=',
fwmark=0x1337,
listen_port=listen,
peer=peer_2,
)
)
for peer in context.wg.info(ifname)[0].get_attr('WGDEVICE_A_PEERS'):
endpoint = peer.get_attr('WGPEER_A_ENDPOINT')
allowed = peer.get_attr('WGPEER_A_ALLOWEDIPS')
assert endpoint['family'] == AF_INET6
assert endpoint['port'] == port
assert endpoint['addr'] in (peer_ip_1, peer_ip_2)
assert allowed[0]['addr'] in (allowed_ip_1, allowed_ip_2)
|
