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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
|
#
# Copyright (c) 2021 Nutanix Inc. All rights reserved.
#
# Authors: John Levon <john.levon@nutanix.com>
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# * Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# * Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# * Neither the name of Nutanix nor the names of its contributors may be
# used to endorse or promote products derived from this software without
# specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL <COPYRIGHT HOLDER> BE LIABLE FOR ANY
# DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
# OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH
# DAMAGE.
#
from libvfio_user import *
import ctypes as c
import errno
ctx = None
def test_pci_cap_setup():
global ctx
ctx = vfu_create_ctx(flags=LIBVFIO_USER_FLAG_ATTACH_NB)
assert ctx != None
ret = vfu_pci_init(ctx, pci_type=VFU_PCI_TYPE_CONVENTIONAL)
assert ret == 0
ret = vfu_setup_region(ctx, index=VFU_PCI_DEV_CFG_REGION_IDX,
size=PCI_CFG_SPACE_SIZE, flags=VFU_REGION_FLAG_RW)
assert ret == 0
def test_pci_cap_bad_flags():
pos = vfu_pci_add_capability(ctx, pos=0, flags=999,
data=struct.pack("ccHH", to_byte(PCI_CAP_ID_PM), b'\0', 0, 0))
assert pos == -1
assert c.get_errno() == errno.EINVAL
def test_pci_cap_no_cb():
pos = vfu_pci_add_capability(ctx, pos=0, flags=VFU_CAP_FLAG_CALLBACK,
data=struct.pack("ccHH", to_byte(PCI_CAP_ID_PM), b'\0', 0, 0))
assert pos == -1
assert c.get_errno() == errno.EINVAL
def test_pci_cap_unknown_cap():
pos = vfu_pci_add_capability(ctx, pos=0, flags=0,
data=struct.pack("ccHH", b'\x81', b'\0', 0, 0))
assert pos == -1
assert c.get_errno() == errno.ENOTSUP
def test_pci_cap_bad_pos():
pos = vfu_pci_add_capability(ctx, pos=PCI_CFG_SPACE_SIZE, flags=0,
data=struct.pack("ccHH", to_byte(PCI_CAP_ID_PM), b'\0', 0, 0))
assert pos == -1
assert c.get_errno() == errno.EINVAL
@c.CFUNCTYPE(c.c_int, c.c_void_p, c.POINTER(c.c_char),
c.c_long, c.c_long, c.c_int)
def pci_region_cb(ctx, buf, count, offset, is_write):
if not is_write:
return read_pci_cfg_space(ctx, buf, count, offset)
return write_pci_cfg_space(ctx, buf, count, offset)
def test_pci_cap_setup_cb():
global ctx
vfu_destroy_ctx(ctx)
ctx = vfu_create_ctx(flags=LIBVFIO_USER_FLAG_ATTACH_NB)
assert ctx != None
ret = vfu_pci_init(ctx, pci_type=VFU_PCI_TYPE_CONVENTIONAL)
assert ret == 0
ret = vfu_setup_region(ctx, index=VFU_PCI_DEV_CFG_REGION_IDX,
size=PCI_CFG_SPACE_SIZE, cb=pci_region_cb,
flags=VFU_REGION_FLAG_RW)
assert ret == 0
cap_offsets = (
PCI_STD_HEADER_SIZEOF,
PCI_STD_HEADER_SIZEOF + PCI_PM_SIZEOF,
# NB: note 4-byte alignment of vsc2
PCI_STD_HEADER_SIZEOF + PCI_PM_SIZEOF + 8,
0x80,
0x90,
0xa0
)
def test_add_caps():
pos = vfu_pci_add_capability(ctx, pos=0, flags=0,
data=struct.pack("ccHH", to_byte(PCI_CAP_ID_PM), b'\0', 0, 0))
assert pos == cap_offsets[0]
data = b"abc"
cap = struct.pack("ccc%ds" % len(data), to_byte(PCI_CAP_ID_VNDR), b'\0',
to_byte(3 + len(data)), data)
pos = vfu_pci_add_capability(ctx, pos=0, flags=VFU_CAP_FLAG_READONLY,
data=cap)
assert pos == cap_offsets[1]
data = b"Hello world."
cap = struct.pack("ccc%ds" % len(data), to_byte(PCI_CAP_ID_VNDR), b'\0',
to_byte(3 + len(data)), data)
pos = vfu_pci_add_capability(ctx, pos=0, flags=VFU_CAP_FLAG_CALLBACK,
data=cap)
assert pos == cap_offsets[2]
pos = vfu_pci_add_capability(ctx, pos=cap_offsets[3], flags=0, data=cap)
assert pos == cap_offsets[3]
pos = vfu_pci_add_capability(ctx, pos=cap_offsets[4], flags=0, data=cap)
assert pos == cap_offsets[4]
ret = vfu_realize_ctx(ctx)
assert ret == 0
def test_find_caps():
offset = vfu_pci_find_capability(ctx, False, PCI_CAP_ID_PM)
assert offset == cap_offsets[0]
space = get_pci_cfg_space(ctx)
assert space[offset] == PCI_CAP_ID_PM
assert space[offset + PCI_CAP_LIST_NEXT] == cap_offsets[1]
offset = vfu_pci_find_next_capability(ctx, False, offset, PCI_CAP_ID_PM)
assert offset == 0
offset = vfu_pci_find_capability(ctx, False, PCI_CAP_ID_VNDR)
assert offset == cap_offsets[1]
assert space[offset] == PCI_CAP_ID_VNDR
assert space[offset + PCI_CAP_LIST_NEXT] == cap_offsets[2]
offset = vfu_pci_find_next_capability(ctx, False, offset, PCI_CAP_ID_PM)
assert offset == 0
offset = vfu_pci_find_next_capability(ctx, False, 0, PCI_CAP_ID_VNDR)
assert offset == cap_offsets[1]
assert space[offset] == PCI_CAP_ID_VNDR
assert space[offset + PCI_CAP_LIST_NEXT] == cap_offsets[2]
offset = vfu_pci_find_next_capability(ctx, False, offset, PCI_CAP_ID_VNDR)
assert offset == cap_offsets[2]
assert space[offset] == PCI_CAP_ID_VNDR
assert space[offset + PCI_CAP_LIST_NEXT] == cap_offsets[3]
offset = vfu_pci_find_next_capability(ctx, False, offset, PCI_CAP_ID_VNDR)
assert offset == cap_offsets[3]
offset = vfu_pci_find_next_capability(ctx, False, offset, PCI_CAP_ID_VNDR)
assert offset == cap_offsets[4]
offset = vfu_pci_find_next_capability(ctx, False, offset, PCI_CAP_ID_VNDR)
assert offset == 0
# check for invalid offsets
offset = vfu_pci_find_next_capability(ctx, False, 8192, PCI_CAP_ID_PM)
assert offset == 0
assert c.get_errno() == errno.EINVAL
offset = vfu_pci_find_next_capability(ctx, False, 256, PCI_CAP_ID_PM)
assert offset == 0
assert c.get_errno() == errno.EINVAL
offset = vfu_pci_find_next_capability(ctx, False, 255, PCI_CAP_ID_PM)
assert offset == 0
assert c.get_errno() == errno.EINVAL
offset = vfu_pci_find_next_capability(ctx, False,
PCI_STD_HEADER_SIZEOF +
PCI_PM_SIZEOF + 1,
PCI_CAP_ID_VNDR)
assert offset == 0
assert c.get_errno() == errno.ENOENT
def test_pci_cap_write_hdr():
sock = connect_client(ctx)
# offset of struct cap_hdr
offset=cap_offsets[0]
data=b'\x01'
write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset,
count=len(data), data=data, expect=errno.EPERM)
disconnect_client(ctx, sock)
def test_pci_cap_readonly():
sock = connect_client(ctx)
# start of vendor payload
offset=cap_offsets[1] + 2
data=b'\x01'
write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset,
count=len(data), data=data, expect=errno.EPERM)
# offsetof(struct vsc, data)
offset=cap_offsets[1] + 3
payload = read_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset,
count=3)
assert payload == b'abc'
disconnect_client(ctx, sock)
def test_pci_cap_callback():
sock = connect_client(ctx)
# offsetof(struct vsc, data)
offset=cap_offsets[2] + 3
data = b"Hello world."
payload = read_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset,
count=len(data))
assert payload == data
data = b"Bye world."
write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset,
count=len(data), data=data)
payload = read_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset,
count=len(data))
assert payload == data
disconnect_client(ctx, sock)
def test_pci_cap_write_pmcs():
sock = connect_client(ctx)
# struct pc
offset=cap_offsets[0] + 3
data=b'\x01\x02'
write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset,
count=len(data), data=data, expect=errno.EINVAL)
offset=cap_offsets[0] + 2
data=b'\x01'
write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset,
count=len(data), data=data, expect=errno.EINVAL)
offset=cap_offsets[0] + 2
data=b'\x01\x02'
write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset,
count=len(data), data=data, expect=errno.ENOTSUP)
# struct pmcs
offset=cap_offsets[0] + 5
data=b'\x01\x02'
write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset,
count=len(data), data=data, expect=errno.EINVAL)
offset=cap_offsets[0] + 4
data=b'\x01'
write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset,
count=len(data), data=data, expect=errno.EINVAL)
offset = cap_offsets[0] + 4
data=b'\x01\x02'
write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset,
count=len(data), data=data)
assert get_pci_cfg_space(ctx)[offset:offset+2] == data
# pmcsr_se
offset=cap_offsets[0] + 6
data=b'\x01'
write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset,
count=len(data), data=data, expect=errno.ENOTSUP)
# data
offset=cap_offsets[0] + 7
data=b'\x01'
write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset,
count=len(data), data=data, expect=errno.ENOTSUP)
disconnect_client(ctx, sock)
reset_flag = -1
@c.CFUNCTYPE(c.c_int, c.c_void_p, c.c_int)
def vfu_reset_cb(ctx, reset_type):
assert reset_type == VFU_RESET_PCI_FLR or reset_type == VFU_RESET_LOST_CONN
global reset_flag
reset_flag = reset_type
return 0
def test_pci_cap_write_px():
sock = connect_client(ctx)
ret = vfu_setup_device_reset_cb(ctx, vfu_reset_cb)
assert ret == 0
#flrc
cap = struct.pack("ccHHcc52c", to_byte(PCI_CAP_ID_EXP), b'\0', 0, 0, b'\0',
b'\x10', *[b'\0' for _ in range(52)])
pos = vfu_pci_add_capability(ctx, pos=cap_offsets[5], flags=0, data=cap)
assert pos == cap_offsets[5]
#iflr
offset = cap_offsets[5] + 8
data = b'\x00\x80'
write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset,
count=len(data), data=data)
assert reset_flag == VFU_RESET_PCI_FLR
disconnect_client(ctx, sock)
assert reset_flag == VFU_RESET_LOST_CONN
def test_pci_cap_write_msix():
# FIXME
pass
def test_pci_cap_cleanup():
vfu_destroy_ctx(ctx)
|
