qemu/tests/qemu-iotests/207

#!/usr/bin/env python
#
# Test ssh image creation
#
# Copyright (C) 2018 Red Hat, Inc.
#
# Creator/Owner: Kevin Wolf <kwolf@redhat.com>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#

import iotests
import subprocess
import re

iotests.verify_image_format(supported_fmts=['raw'])
iotests.verify_protocol(supported=['ssh'])

def filter_hash(msg):
    return re.sub('"hash": "[0-9a-f]+"', '"hash": HASH', msg)

def blockdev_create(vm, options):
    result = vm.qmp_log('blockdev-create', job_id='job0', options=options,
                        filters=[iotests.filter_testfiles, filter_hash])

    if 'return' in result:
        assert result['return'] == {}
        vm.run_job('job0')
    iotests.log("")

with iotests.FilePath('t.img') as disk_path, \
     iotests.VM() as vm:

    remote_path = iotests.remote_filename(disk_path)

    #
    # Successful image creation (defaults)
    #
    iotests.log("=== Successful image creation (defaults) ===")
    iotests.log("")

    vm.launch()
    blockdev_create(vm, { 'driver': 'ssh',
                          'location': {
                              'path': disk_path,
                              'server': {
                                  'host': '127.0.0.1',
                                  'port': '22'
                              }
                          },
                          'size': 4194304 })
    vm.shutdown()

    iotests.img_info_log(remote_path, filter_path=disk_path)
    iotests.log("")
    iotests.img_info_log(disk_path)

    #
    # Test host-key-check options
    #
    iotests.log("=== Test host-key-check options ===")
    iotests.log("")

    vm.launch()
    blockdev_create(vm, { 'driver': 'ssh',
                          'location': {
                              'path': disk_path,
                              'server': {
                                  'host': '127.0.0.1',
                                  'port': '22'
                              },
                              'host-key-check': {
                                  'mode': 'none'
                              }
                          },
                          'size': 8388608 })
    vm.shutdown()

    iotests.img_info_log(remote_path, filter_path=disk_path)

    vm.launch()
    blockdev_create(vm, { 'driver': 'ssh',
                          'location': {
                              'path': disk_path,
                              'server': {
                                  'host': '127.0.0.1',
                                  'port': '22'
                              },
                              'host-key-check': {
                                  'mode': 'known_hosts'
                              }
                          },
                          'size': 4194304 })
    vm.shutdown()

    iotests.img_info_log(remote_path, filter_path=disk_path)

    md5_key = subprocess.check_output(
        'ssh-keyscan -t rsa 127.0.0.1 2>/dev/null | grep -v "\\^#" | ' +
        'cut -d" " -f3 | base64 -d | md5sum -b | cut -d" " -f1',
        shell=True).rstrip().decode('ascii')

    vm.launch()
    blockdev_create(vm, { 'driver': 'ssh',
                          'location': {
                              'path': disk_path,
                              'server': {
                                  'host': '127.0.0.1',
                                  'port': '22'
                              },
                              'host-key-check': {
                                  'mode': 'hash',
                                  'type': 'md5',
                                  'hash': 'wrong',
                              }
                          },
                          'size': 2097152 })
    blockdev_create(vm, { 'driver': 'ssh',
                          'location': {
                              'path': disk_path,
                              'server': {
                                  'host': '127.0.0.1',
                                  'port': '22'
                              },
                              'host-key-check': {
                                  'mode': 'hash',
                                  'type': 'md5',
                                  'hash': md5_key,
                              }
                          },
                          'size': 8388608 })
    vm.shutdown()

    iotests.img_info_log(remote_path, filter_path=disk_path)

    sha1_key = subprocess.check_output(
        'ssh-keyscan -t rsa 127.0.0.1 2>/dev/null | grep -v "\\^#" | ' +
        'cut -d" " -f3 | base64 -d | sha1sum -b | cut -d" " -f1',
        shell=True).rstrip().decode('ascii')

    vm.launch()
    blockdev_create(vm, { 'driver': 'ssh',
                          'location': {
                              'path': disk_path,
                              'server': {
                                  'host': '127.0.0.1',
                                  'port': '22'
                              },
                              'host-key-check': {
                                  'mode': 'hash',
                                  'type': 'sha1',
                                  'hash': 'wrong',
                              }
                          },
                          'size': 2097152 })
    blockdev_create(vm, { 'driver': 'ssh',
                          'location': {
                              'path': disk_path,
                              'server': {
                                  'host': '127.0.0.1',
                                  'port': '22'
                              },
                              'host-key-check': {
                                  'mode': 'hash',
                                  'type': 'sha1',
                                  'hash': sha1_key,
                              }
                          },
                          'size': 4194304 })
    vm.shutdown()

    iotests.img_info_log(remote_path, filter_path=disk_path)

    #
    # Invalid path and user
    #
    iotests.log("=== Invalid path and user ===")
    iotests.log("")

    vm.launch()
    blockdev_create(vm, { 'driver': 'ssh',
                          'location': {
                              'path': '/this/is/not/an/existing/path',
                              'server': {
                                  'host': '127.0.0.1',
                                  'port': '22'
                              },
                              'host-key-check': {
                                  'mode': 'none'
                              }
                          },
                          'size': 4194304 })
    blockdev_create(vm, { 'driver': 'ssh',
                          'location': {
                              'path': disk_path,
                              'user': 'invalid user',
                              'server': {
                                  'host': '127.0.0.1',
                                  'port': '22'
                              },
                              'host-key-check': {
                                  'mode': 'none'
                              }
                          },
                          'size': 4194304 })
    vm.shutdown()