1#!/usr/bin/env python3
2# group: rw auto
3#
4# Test LUKS volume with detached header
5#
6# Copyright (C) 2024 SmartX Inc.
7#
8# Authors:
9#     Hyman Huang <yong.huang@smartx.com>
10#
11# This program is free software; you can redistribute it and/or modify
12# it under the terms of the GNU General Public License as published by
13# the Free Software Foundation; either version 2 of the License, or
14# (at your option) any later version.
15#
16# This program is distributed in the hope that it will be useful,
17# but WITHOUT ANY WARRANTY; without even the implied warranty of
18# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
19# GNU General Public License for more details.
20#
21# You should have received a copy of the GNU General Public License
22# along with this program.  If not, see <http://www.gnu.org/licenses/>.
23#
24
25import os
26import json
27import iotests
28from iotests import (
29    imgfmt,
30    qemu_img_create,
31    qemu_img_info,
32    QMPTestCase,
33)
34
35
36image_size = 128 * 1024 * 1024
37
38luks_img = os.path.join(iotests.test_dir, "luks.img")
39detached_header_img1 = os.path.join(iotests.test_dir, "detached_header.img1")
40detached_header_img2 = os.path.join(iotests.test_dir, "detached_header.img2")
41detached_payload_raw_img = os.path.join(
42    iotests.test_dir, "detached_payload_raw.img"
43)
44detached_payload_qcow2_img = os.path.join(
45    iotests.test_dir, "detached_payload_qcow2.img"
46)
47detached_header_raw_img = "json:" + json.dumps(
48    {
49        "driver": "luks",
50        "file": {"filename": detached_payload_raw_img},
51        "header": {
52            "filename": detached_header_img1,
53        },
54    }
55)
56detached_header_qcow2_img = "json:" + json.dumps(
57    {
58        "driver": "luks",
59        "file": {"filename": detached_payload_qcow2_img},
60        "header": {"filename": detached_header_img2},
61    }
62)
63
64secret_obj = "secret,id=sec0,data=foo"
65luks_opts = "key-secret=sec0"
66
67
68class TestDetachedLUKSHeader(QMPTestCase):
69    def setUp(self) -> None:
70        self.vm = iotests.VM()
71        self.vm.add_object(secret_obj)
72        self.vm.launch()
73
74        # 1. Create the normal LUKS disk with 128M size
75        self.vm.blockdev_create(
76            {"driver": "file", "filename": luks_img, "size": 0}
77        )
78        self.vm.qmp_log(
79            "blockdev-add",
80            driver="file",
81            filename=luks_img,
82            node_name="luks-1-storage",
83        )
84        result = self.vm.blockdev_create(
85            {
86                "driver": imgfmt,
87                "file": "luks-1-storage",
88                "key-secret": "sec0",
89                "size": image_size,
90                "iter-time": 10,
91            }
92        )
93        # None is expected
94        self.assertEqual(result, None)
95
96        # 2. Create the LUKS disk with detached header (raw)
97
98        # Create detached LUKS header
99        self.vm.blockdev_create(
100            {"driver": "file", "filename": detached_header_img1, "size": 0}
101        )
102        self.vm.qmp_log(
103            "blockdev-add",
104            driver="file",
105            filename=detached_header_img1,
106            node_name="luks-2-header-storage",
107        )
108
109        # Create detached LUKS raw payload
110        self.vm.blockdev_create(
111            {"driver": "file", "filename": detached_payload_raw_img, "size": 0}
112        )
113        self.vm.qmp_log(
114            "blockdev-add",
115            driver="file",
116            filename=detached_payload_raw_img,
117            node_name="luks-2-payload-storage",
118        )
119
120        # Format LUKS disk with detached header
121        result = self.vm.blockdev_create(
122            {
123                "driver": imgfmt,
124                "header": "luks-2-header-storage",
125                "file": "luks-2-payload-storage",
126                "key-secret": "sec0",
127                "preallocation": "full",
128                "size": image_size,
129                "iter-time": 10,
130            }
131        )
132        self.assertEqual(result, None)
133
134        self.vm.shutdown()
135
136        # 3. Create the LUKS disk with detached header (qcow2)
137
138        # Create detached LUKS header using qemu-img
139        res = qemu_img_create(
140            "-f",
141            "luks",
142            "--object",
143            secret_obj,
144            "-o",
145            luks_opts,
146            "-o",
147            "detached-header=true",
148            detached_header_img2,
149        )
150        assert res.returncode == 0
151
152        # Create detached LUKS qcow2 payload
153        res = qemu_img_create(
154            "-f", "qcow2", detached_payload_qcow2_img, str(image_size)
155        )
156        assert res.returncode == 0
157
158    def tearDown(self) -> None:
159        os.remove(luks_img)
160        os.remove(detached_header_img1)
161        os.remove(detached_header_img2)
162        os.remove(detached_payload_raw_img)
163        os.remove(detached_payload_qcow2_img)
164
165        # Check if there was any qemu-io run that failed
166        if "Pattern verification failed" in self.vm.get_log():
167            print("ERROR: Pattern verification failed:")
168            print(self.vm.get_log())
169            self.fail("qemu-io pattern verification failed")
170
171    def test_img_creation(self) -> None:
172        # Check if the images created above are expected
173
174        data = qemu_img_info(luks_img)["format-specific"]
175        self.assertEqual(data["type"], imgfmt)
176        self.assertEqual(data["data"]["detached-header"], False)
177
178        data = qemu_img_info(detached_header_raw_img)["format-specific"]
179        self.assertEqual(data["type"], imgfmt)
180        self.assertEqual(data["data"]["detached-header"], True)
181
182        data = qemu_img_info(detached_header_qcow2_img)["format-specific"]
183        self.assertEqual(data["type"], imgfmt)
184        self.assertEqual(data["data"]["detached-header"], True)
185
186        # Check if preallocation works
187        size = qemu_img_info(detached_payload_raw_img)["actual-size"]
188        self.assertGreaterEqual(size, image_size)
189
190    def test_detached_luks_header(self) -> None:
191        self.vm.launch()
192
193        # 1. Add the disk created above
194
195        # Add normal LUKS disk
196        self.vm.qmp_log(
197            "blockdev-add",
198            driver="file",
199            filename=luks_img,
200            node_name="luks-1-storage",
201        )
202        result = self.vm.qmp_log(
203            "blockdev-add",
204            driver="luks",
205            file="luks-1-storage",
206            key_secret="sec0",
207            node_name="luks-1-format",
208        )
209
210        # Expected result{ "return": {} }
211        self.assert_qmp(result, "return", {})
212
213        # Add detached LUKS header with raw payload
214        self.vm.qmp_log(
215            "blockdev-add",
216            driver="file",
217            filename=detached_header_img1,
218            node_name="luks-header1-storage",
219        )
220
221        self.vm.qmp_log(
222            "blockdev-add",
223            driver="file",
224            filename=detached_payload_raw_img,
225            node_name="luks-2-payload-raw-storage",
226        )
227
228        result = self.vm.qmp_log(
229            "blockdev-add",
230            driver=imgfmt,
231            header="luks-header1-storage",
232            file="luks-2-payload-raw-storage",
233            key_secret="sec0",
234            node_name="luks-2-payload-raw-format",
235        )
236        self.assert_qmp(result, "return", {})
237
238        # Add detached LUKS header with qcow2 payload
239        self.vm.qmp_log(
240            "blockdev-add",
241            driver="file",
242            filename=detached_header_img2,
243            node_name="luks-header2-storage",
244        )
245
246        self.vm.qmp_log(
247            "blockdev-add",
248            driver="file",
249            filename=detached_payload_qcow2_img,
250            node_name="luks-3-payload-qcow2-storage",
251        )
252
253        result = self.vm.qmp_log(
254            "blockdev-add",
255            driver=imgfmt,
256            header="luks-header2-storage",
257            file="luks-3-payload-qcow2-storage",
258            key_secret="sec0",
259            node_name="luks-3-payload-qcow2-format",
260        )
261        self.assert_qmp(result, "return", {})
262
263        # 2. Do I/O test
264
265        # Do some I/O to the image to see whether it still works
266        # (Pattern verification will be checked by tearDown())
267
268        # Normal LUKS disk
269        result = self.vm.qmp_log(
270            "human-monitor-command",
271            command_line='qemu-io luks-1-format "write -P 40 0 64k"',
272        )
273        self.assert_qmp(result, "return", "")
274
275        result = self.vm.qmp_log(
276            "human-monitor-command",
277            command_line='qemu-io luks-1-format "read -P 40 0 64k"',
278        )
279        self.assert_qmp(result, "return", "")
280
281        # Detached LUKS header with raw payload
282        cmd = 'qemu-io luks-2-payload-raw-format "write -P 41 0 64k"'
283        result = self.vm.qmp(
284            "human-monitor-command",
285            command_line=cmd
286        )
287        self.assert_qmp(result, "return", "")
288
289        cmd = 'qemu-io luks-2-payload-raw-format "read -P 41 0 64k"'
290        result = self.vm.qmp(
291            "human-monitor-command",
292            command_line=cmd
293        )
294        self.assert_qmp(result, "return", "")
295
296        # Detached LUKS header with qcow2 payload
297        cmd = 'qemu-io luks-3-payload-qcow2-format "write -P 42 0 64k"'
298        result = self.vm.qmp(
299            "human-monitor-command",
300            command_line=cmd
301        )
302        self.assert_qmp(result, "return", "")
303
304        cmd = 'qemu-io luks-3-payload-qcow2-format "read -P 42 0 64k"'
305        result = self.vm.qmp(
306            "human-monitor-command",
307            command_line=cmd
308        )
309        self.assert_qmp(result, "return", "")
310
311        self.vm.shutdown()
312
313
314if __name__ == "__main__":
315    # Test image creation and I/O
316    iotests.main(supported_fmts=["luks"], supported_protocols=["file"])
317