fortrace.core package

Submodules

fortrace.core.qemu_monitor module

Module for interaction with the QEMU Monitor of a domain.

From the QEMU Monitor documentation (https://www.qemu.org/docs/master/system/monitor.html). The QEMU monitor is used to give complex commands to the QEMU emulator.

In case of ForTrace++ it is used to control the mouse and the keyboard of a domain. Furthermore, it can be used to mount devices on a domain.

class fortrace.core.qemu_monitor.Mouse(domain: virDomain, take_screenshot: Callable[[int | None], bytes])

Bases: object

Mouse pointer, controlled via QEMUMonitor.

click(button: MouseButton = MouseButton.LEFT, number_of_clicks: int = 1)

Perform n mouse clicks at current position.

Parameters:

• button – which button to press

• number_of_clicks – how many clicks to perform

drag(dx: int, dy: int, dz: int | None = None, absolute: bool = False, button: MouseButton = MouseButton.LEFT)

Drag an item from current position to specified position.

Press a mouse button at the current position, go to specified position and release the button.

Parameters:

• dx – x-direction (positive: move to right, negative: move to left)

• dy – y-direction (positive: move down, negative: move up)

• dz – z-direction=scroll-wheel

• absolute – place the mouse to an absolute position

• button – which button to press

find_and_zero_mouse_ptr()

Method to find the mouse pointer on the domain screen.

The determined position is put on the movement stack.

get_position(position: int = -1) → tuple[int, int]

Return n-th position in movement stack.

Parameters:

position – use ‘-1’ to return the last position,

Returns:

(x,y) coordinates of the mouse pointer

init()

Initialize the mouse class

move(dx: int = 0, dy: int = 0, dz: int = 0, absolute: bool = False)

Move the mouse to specified coordinates.

The origin is the top left corner of the screen.

Parameters:

• dx – x-direction [px] (positive: move to right, negative: move to left)

• dy – y-direction [px] (positive: move down, negative: move up)

• dz – z-direction [scroll wheel steps] (positive: move up, negative move down)

• absolute – place the mouse to an absolute position specified by dx and dy

move_and_click(dx: int, dy: int, dz: int | None = None, absolute: bool = False, button: MouseButton = MouseButton.LEFT, number_of_clicks: int = 1)

Move the mouse to a specific position and perform a click there.

Parameters:

• dx – x-direction [px] (positive: move to right, negative: move to left)

• dy – y-direction [px] (positive: move down, negative: move up)

• dz – z-direction [scroll wheel steps] (positive: move up, negative: move down)

• absolute – place the mouse to an absolute position

• button – which button to press

• number_of_clicks – how many clicks to perform

move_back(position: int)

Move mouse back to specified position.

Make the selected position the latest in movement stack of mouse.

Parameters:

position – position in mouse movement stack to move back to

class fortrace.core.qemu_monitor.MouseButton(*values)

Bases: Enum

Map QEMUs codes to mouse buttons (taken from documentation).

LEFT = 1

MIDDLE = 2

RIGHT = 4

class fortrace.core.qemu_monitor.QEMUMonitorSession(domain: virDomain, take_screenshot: Callable[[int | None], bytes], domain_output_path: Path)

Bases: object

Representation of the QEMU Monitor connection to one specific domain.

This class wraps the libvirt_qemu bindings to provide handy methods for interaction with a domain.

_mouse

Handle to the mouse pointer object

Type:

fortrace.core.qemu_monitor.Mouse

_special_keys_mapping

list of key-codes used by qemu on a US layout

direct_command(command: str) → str

Directly send a command to QEMU monitor.

Parameters:

command – qemu-monitor command to be sent

Returns:

the output of the command

property mouse

save_screenshot(name: str | None = None)

Save a screenshot of the domain to path using qemu monitor command.

Parameters:

name – if no name is specified, the current time is used. File ending .ppm is automatically appended

send_key_combination(key_combination: str, times: int = 1)

Send a key combination via the monitor.

Parameters:

• key_combination – string must consist of qemu-monitor keycodes separated by ‘-’, e.g. ‘alt-f4’

• times – how many times the key combination should be entered

send_text(text: str, end_ret: bool = False)

Send the specified text to the active domain.

Parameters:

• text – text to be sent

• end_ret – should return be pressed afterward?

take_screenshot(screen: int = 0)

See documentation of VirshDomain.take_screenshot.

fortrace.core.simulation_monitor module

fortrace.core.virsh_console module

class fortrace.core.virsh_console.VirshConsole(domain: str, user: str, password: str, shell: ShellType, hypervisor: str = 'qemu:///system', timeout: int = 2, sudo_timeout: float = 300)

Bases: object

Class to handle interaction with virsh console connections.

This class is only supported on Linux and macOS domains, since Windows does not implement this technique.

close()

Close the console connection.

install_packages(packages: list[str], package_manager: PackageManager)

Install packages via a package manager through the console.

Provide a list of packages that need to be installed. Make sure the packages exist before calling this method.

Parameters:

• packages – packages to be installed (use the names of the package manager)

• package_manager – As each package installer has a different approach to install packages, provide it here

Returns:

open_console_application(command: str, elevated: bool = False, timeout: int | None = -1, new_prompt: str | None = None) → GenericConsoleApplication

Open a console application and retrieve the associated object for interaction.

Opens an application on the terminal, which might render REPLWrapper useless, as sometimes there will be no prompt until the application is closed.

Parameters:

• command – command (do NOT prepend sudo here)

• elevated – should the command run elevated?

• timeout – timeout in seconds for command

• new_prompt – if there is a new prompt expected after a command is run, specify it here (otherwise a TIMEOUT exception will be thrown)

Returns:

Handle the opened console application.

remove_packages(packages: list[str], package_manager: PackageManager)

Provide a list of packages to be removed from the guest.

Make sure that any package in the list of provided ones is really installed on the system.

Parameters:

• packages – packages to be removed

• package_manager – As each package installer has a different approach to install packages, provide it here

restore_prompt()

Restores the prompt of the console to the initial one.

Call this method BEFORE exiting a program with a different prompt (can be called in their __del__ method)

Returns:

run_command(command: str, elevated: bool = False, timeout: int | None = -1, new_prompt: str | None = None) → str | list[str]

Run a program in the opened console session.

Run a command in pty that might be elevated or result in a new prompt. Do not use this method to call editors or start graphical programs!

Parameters:

• command – the command to be run

• elevated – should the command run elevated, e.g. sudo

• timeout – overwrite the standard timeout for a command (e.g. when downloading something from the internet)

• new_prompt – if the command changes the prompt, provide the new expected prompt here (see tests, for example)

Returns:

adjusted string written to stdout

property shell_type

property virsh

fortrace.core.virsh_domain module

This module bundles the VirshDomain and its child with Graphical extension.

class fortrace.core.virsh_domain.GraphicalVirshDomain(domain_name: str, connection: virConnect, domain_network: list[str] | None, os_type: OSType, session_output: PathLike, desktop_env: DesktopEnvironmentType)

Bases: VirshDomain

Extension of the VirshDomain class with desktop environment attribute.

property env

class fortrace.core.virsh_domain.VirshDomain(domain_name: str, connection: virConnect, domain_network: list[str] | None, os_type: OSType, session_output: PathLike)

Bases: object

Single VM/domain under control of ForTrace++.

attach_usb_device(vendor_id: str, product_id: str)

Attach an available USB device to the running domain.

Parameters:

• vendor_id – vendor ID of the USB device, e.g., ‘0x1234’

• product_id – product ID of the USB device, e.g., ‘0x5678’

Note

The device is attached only to the live domain. It is NOT attached across hibernation or boot cycles. You need to attach it again.

boot(start_sniffer: bool = True, snapshot: str | None = None)

Boot the VM with the option to start a sniffer or use a specific snapshot.

Parameters:

• start_sniffer – start sniffer once the guest has an IP address?

• snapshot – Specify a snapshot from which the guest should be booted. If the snapshot does not exist, it will be created from the current state of the guest and is available for future iterations

create_snapshot(name: str, flags: int = 0) → virDomainSnapshot

Create a snapshot of the domain.

The guest might be running or powered off.

Parameters:

• name – specify a unique name for the snapshot

• flags – virDomainSnapshotCreateFlags (https://libvirt.org/html/libvirt-libvirt-domain-snapshot.html#virDomainSnapshotCreateFlags)

Returns:

The newly created snapshot

delete()

Delete a domain and every associated data from the host.

Use this method with caution and only on inactive domains.

delete_snapshot(name: str, flags: int = 0)

Delete a snapshot of the domain.

Parameters:

• name – name of the snapshot to be deleted

• flags – virDomainSnapshotDeleteFlags (https://libvirt.org/html/libvirt-libvirt-domain-snapshot.html#virDomainSnapshotDeleteFlags)

destroy()

Destroy the domain.

This is a more powerful variant of shutdown, and can be used if the guest OS is known to ignore a normal shutdown request. The method tries a graceful shutdown first (it will fail after a certain timeout). If it fails, the ‘virtual power cord’ is pulled out of the guest, which may result in data loss.

It might be more convenient to power the guest down via its PTY or the GUI (if there is one).

detach_usb_device(vendor_id: str, product_id: str)

Detach an available USB device of the running domain.

Parameters:

• vendor_id – vendor ID of the USB device, e.g., ‘0x1234’

• product_id – product ID of the USB device, e.g., ‘0x5678’

property domain: virDomain

property domain_name: str

dump_image(images: list[str] | None = None)

Saves images of a domain as raw files into the domain_output_path.

Each image is named after its device name provided in the domain configuration. Domain has to be active to perform a live backup. The method will block until the backup is completed.

Parameters:

images – list of images to be dumped (None if all images should be dumped). The names must be the same as those used in the libvirt configuration.

dump_memory(compression: Literal['elf', 'kdump-zlib', 'kdump-lzo', 'kdump-snappy'] = 'elf', name: str | None = None)

Dumps the domain memory via libvirt into the domain’s simulation directory.

The memory dump will have the current timestamp as its name, to ensure its

unique and sortable. The file suffix is determined based on the selected compression algorithm.

Parameters:

• compression – which compression type to be used elf: the default, uncompressed format kdump-zlib: kdump-compressed format with zlib compression kdump-lzo: kdump-compressed format with LZO compression kdump-snappy: kdump-compressed format with Snappy compression

• name – give te RAM dump a unique name (defaults to timestamp)

get_time() → datetime

Returns UTC time of guest, based on the settings of guest’s configuration.

Note

libvirt’s getTime is only available with a running guest agent, which we

don’t want

Returns:

the datetime of the domain

insert_cd(iso_path: PathLike)

Insert/change a CD-ROM into the (running) guest domain.

Please make sure that a CD-ROM drive is already added to the guest’s configuration (should look like the format string below, without the source attribute).

Parameters:

iso_path – provide the path to the ISO file here

property ipv4: IPv4Address

list_snapshots(flags: int = 0) → list[str]

Get a list of all snapshot names

Parameters:

flags – virDomainSnapshotListFlags (https://libvirt.org/html/libvirt-libvirt-domain-snapshot.html#virDomainSnapshotListFlags)

Returns:

list of snapshot names

open_pty(user: str, password: str, shell: ShellType, timeout: int = 2) → VirshConsole

Open a pty for the specified user in the domain.

Checks whether there already is an active PTY session, and if so, returns it

instead.

Parameters:

• user – username

• password – user’s password

• shell – the shell type (important for prompt change)

• timeout – timeout for pty command in seconds

Returns:

connection to an active console

property pty: VirshConsole | None

reset_time()

Restores the guest clock to the host clock.

revert_to_snapshot(name: str, flags: int = 0)

Revert the guest to the specified snapshot.

If the snapshot name is unknown, ForTrace++ stops.

Parameters:

• name – give the unique name of the snapshot

• flags – virDomainSnapshotRevertFlags (https://libvirt.org/html/libvirt-libvirt-domain-snapshot.html#virDomainSnapshotRevertFlags)

set_time(datetime_guest: datetime | ZoneInfo)

Change the time of the guest.

Set the time of a guest by modifying the configuration file and specifying a

relative offset to the hardware clock or a timezone.

Parameters:

datetime_guest – The time or timezone to set the guest to

Note

One must disable all NTP services on the guest, as it will update the offset

(e.g., on Linux ‘timedatectl set-ntp 0’)

shutdown(blocking: bool = False)

Send a shutdown signal to the guest.

The guest OS might ignore the request. It might be more convenient to power the

guest down via its PTY or the GUI (if there is one).

Notes

To forcefully shut down the domain use destroy.

Parameters:

blocking – should the call block until the guest is shutdown (use with caution, as there is no timeout)

start_screen_recording(interval: float = 5.0)

Periodically take screenshots of the domain to document the execution.

The screenshots are placed in the domain’s output directory under ‘screen_recording’, e.g., /var/tmp/ForTrace/win10/screen_recording/.

Each image bears the name of the time it was taken. The images can be combined into a hyperlapse, i.e., with ffmpeg:

> ffmpeg -framerate 24 -pattern_type glob -i “*.png” -c:v libx264 output.mp4

Parameters:

interval – delay in seconds between screenshots

start_sniffer()

Start the network sniffer on the session network interface.

Note

dumpcap filters for domain related packages, meaning the package must come from or must be addressed to the domain. One sniffer can be started per domain.

stop_screen_recording()

Stop the screen recording thread.

stop_sniffer()

Stop the network sniffer.

take_screenshot(screen: int = 0) → bytes

Takes a screenshot of the domain’s specified screen

A non-graphical guest also shows a console screen, thus this method is also available in this class.

Parameters:

screen – ID of the virtualized screen to take a screenshot from (If the domain only has one screen, use ‘0’)

Returns:

byte sequence of specified screen. Image format is hypervisor specific. Refer to the appropriate documentation.

Warning

Do NOT change this method without reviewing QEMUMonitor_Session

property template: ElementTree

transfer_file(source: PathLike, destination: PathLike, target_dev: str = 'vda')

Transfer a file/directory to guest using guestfs.

The domain should be powered down, so no corruption of the guest’s filesystem might occur.

Due to a license conflict, guestfs is not part of PyPi and has to be installed manually. Thus, it is imported at function level, so ForTrace can run without guestfs and only imports it on demand. Besides the Python bindings, libguestfish has to be installed on the system.

For more information visit https://www.libguestfs.org/

Parameters:

• source – file or folder to be transferred from the host

• destination – path on the guest file system

• target_dev –

  name of the target device of XML config. If domain has only one disk, ‘vda’ should work. If domain has more disks, they are usually called ‘vdb’, ‘vdc’, … ->

  look into the config

fortrace.core.virsh_session module

class fortrace.core.virsh_session.VirshSession(session_output: PathLike, hypervisor: str = 'qemu:///system')

Bases: object

ForTrace’s representation of the hypervisor connection.

clone(domain_name: str) → str

Clone the specified domain.

Returns:

name of the cloned domain

create_domain(domain_name: str, os_type: OSType, domain_network: str | None = 'default', desktop_environment: DesktopEnvironmentType | None = None) → VirshDomain | GraphicalVirshDomain

Creates a new domain with the given name.

The domain is not started, which is done via the domain object.

Parameters:

• domain_name – name or UUID of the domain to be created

• os_type – type of OS to expect

• domain_network – name/UUIDString of the network the domain should connect to

• desktop_environment – if domain is a graphical target, please provide the expected desktop environment

Note

‘Create’ uses the meaning as it is defined by libvirt. A domain to be

created has to be installed on the system.

Returns:

a handle to the domain (graphical or non-graphical)

property domains

get_domain_by_id(domain_id: str) → VirshDomain | None

Get the domain with the specified domain from the list of session domains.

Parameters:

domain_id – UUIDString of the domain to be retrieved

Returns:

The VirshDomain with the given UUID if it is in the list of session domains.

get_domain_by_name(domain_name: str) → VirshDomain

Get a domain based on its name of the ForTrace related domains.

Parameters:

domain_name – name of the domain to retrieve

Returns:

Domain, if there is one with the given name.

get_network_by_id(network_id: str) → virNetwork | None

Get a network based on its UUID of the ForTrace networks.

Parameters:

network_id – UUID string of the network to retrieve

Returns:

The libvirt virNetwork object or None

get_network_by_name(network_name: str) → virNetwork | None

Get a network based on its name of the ForTrace networks.

Parameters:

network_name – name of the network to retrieve

Returns:

The libvirt virNetwork object or None

list_available_domains() → list[tuple[str, str]]

List all available domains of the hypervisor with their UUID.

list_available_networks() → list[tuple[str, str]]

List all available networks of the hypervisor with their UUID.

property networks

remove_domain(domain: VirshDomain)

Remove the provided domain from the session.

The removed domain will be powered down (with force, if necessary).

Module contents

The core package contains the implementation of the Virtual Machine Monitor (vmm.py), Guest control (guest.py) and Agent which will be used on a running guest to control it (agent.py).