qubesadmin package

Subpackages

Submodules

qubesadmin.app module

Main Qubes() class and related classes.

class qubesadmin.app.QubesBase[source]

Bases: PropertyHolder

Main Qubes application.

This is a base abstract class, don’t use it directly. Use specialized class in py:class:qubesadmin.Qubes instead, which points at QubesLocal or QubesRemote.

add_new_vm(cls: str | type[QubesVM], name: str, label: str, template: str | QubesVM | None = None, pool: str | None = None, pools: dict | None = None) → QubesVM[source]

Create new Virtual Machine

Example usage with custom storage pools:

>>> app = qubesadmin.Qubes()
>>> pools = {'private': 'external'}
>>> vm = app.add_new_vm('AppVM', 'my-new-vm', 'red',
>>>    'my-template', pools=pools)
>>> vm.netvm = app.domains['sys-whonix']
Parameters:

  • cls (str) – name of VM class (AppVM, TemplateVM etc)

  • name (str) – name of VM

  • label (str) – label color for new VM

  • template (str) – template to use (if apply for given VM class), can be also VM object; use None for default value

  • pool (str) – storage pool to use instead of default one

  • pools (dict) – storage pool for specific volumes

:return new VM object

add_pool(name: str, driver: str, **kwargs) → None[source]

Add a storage pool to config

Parameters:

  • name – name of storage pool to create

  • driver – driver to use, see pool_drivers() for available drivers

  • kwargs – configuration parameters for storage pool, see pool_driver_parameters() for a list

blind_mode: bool = False

do not check for object (VM, label etc) existence before really needed

cache_enabled: bool = False

cache retrieved properties values

clone_vm(src_vm: str | QubesVM, new_name: str, new_cls: str | None = None, *, pool: str | None = None, pools: dict | None = None, ignore_errors: bool = False, ignore_volumes: list | None = None, ignore_devices: bool = False) → QubesVM[source]

Clone Virtual Machine

Example usage with custom storage pools:

>>> app = qubesadmin.Qubes()
>>> pools = {'private': 'external'}
>>> src_vm = app.domains['personal']
>>> vm = app.clone_vm(src_vm, 'my-new-vm', pools=pools)
>>> vm.label = app.labels['green']
Parameters:

  • src_vm (QubesVM or str) – source VM

  • new_name (str) – name of new VM

  • new_cls (str) – name of VM class (AppVM, TemplateVM etc) - use None to copy it from src_vm

  • pool (str) – storage pool to use instead of default one

  • pools (dict) – storage pool for specific volumes

  • ignore_errors (bool) – should errors on meta-data setting be only logged, or abort the whole operation?

  • ignore_volumes (list) – do not clone volumes on this list, like ‘private’ or ‘root’

  • ignore_devices (bool) – if True, do not copy device assignments

:return new VM object

domains: VMCollection

domains (VMs) collection

get_label(label: str | int) → Label[source]

Get label as identified by index or name

Throws QubesLabelNotFoundError:

when label is not found

static get_vm_class(clsname: str) → str[source]

Find the class for a domain.

Compatibility function, client tools use str to identify domain classes.

Parameters:

clsname (str) – name of the class

Return str:

class

labels: WrapperObjectsCollection[Label]

labels collection

list_deviceclass() → list[str][source]

Call Qubesd in order to obtain the device classes list

list_vmclass() → list[str][source]

Call Qubesd in order to obtain the vm classes list

property local_name: str

Get localhost name

log: Logger

logger

pool_driver_parameters(driver: str) → list[str][source]

Parameters to initialize storage pool using given driver

property pool_drivers: Iterable[str]

Available storage pool drivers

pools: WrapperObjectsCollection[Pool]

storage pools

qubesd_call(dest: str | None, method: str, arg: str | None = None, payload: bytes | None = None, payload_stream: IO | None = None) → bytes[source]

Execute Admin API method.

If payload and payload_stream are both specified, they will be sent in that order.

Parameters:

  • dest – Destination VM name

  • method – Full API method name (‘admin…’)

  • arg – Method argument (if any)

  • payload – Payload send to the method

  • payload_stream – file-like object to read payload from

Returns:

Data returned by qubesd (string)

Warning

payload_stream will get closed by this function

qubesd_connection_type: Literal['socket', 'qrexec']

type of qubesd connection: either ‘socket’ or ‘qrexec’

remove_pool(name: str) → None[source]

Remove a storage pool

run_service(dest: str, service: str, user: str | None = None, *, filter_esc: bool = False, localcmd: str | None = None, wait: bool = True, autostart: bool = True, prefix_data: bytes | None = None, **kwargs) → Popen[source]

Run qrexec service in a given destination

kwargs are passed verbatim to subprocess.Popen().

Parameters:

  • dest (str) – Destination - may be a VM name or empty string for default (for a given service)

  • service (str) – service name

  • filter_esc (bool) – filter escape sequences to protect terminal emulator

  • user (str) – username to run service as

  • localcmd (str) – Command to connect stdin/stdout to

  • wait (bool) – Wait service run

  • autostart (bool) – Automatically start the target VM

Return type:

subprocess.Popen

class qubesadmin.app.QubesLocal[source]

Bases: QubesBase

Application object communicating through local socket.

Used when running in dom0.

qubesd_call(dest: str | None, method: str, arg: str | None = None, payload: bytes | None = None, payload_stream: IO | None = None) → bytes[source]

Execute Admin API method.

If payload and payload_stream are both specified, they will be sent in that order.

Parameters:

  • dest – Destination VM name

  • method – Full API method name (‘admin…’)

  • arg – Method argument (if any)

  • payload – Payload send to the method

  • payload_stream – file-like object to read payload from

Returns:

Data returned by qubesd (string)

Warning

payload_stream will get closed by this function

qubesd_connection_type = 'socket'

type of qubesd connection: either ‘socket’ or ‘qrexec’

run_service(dest: str, service: str, user: str | None = None, *, filter_esc: bool = False, localcmd: str | None = None, wait: bool = True, autostart: bool = True, prefix_data: bytes | None = None, **kwargs) → Popen[source]

Run qrexec service in a given destination

Parameters:

  • dest (str) – Destination - may be a VM name or empty string for default (for a given service)

  • service (str) – service name

  • filter_esc (bool) – filter escape sequences to protect terminal emulator

  • user (str) – username to run service as

  • localcmd (str) – Command to connect stdin/stdout to

  • wait (bool) – wait for remote process to finish

Return type:

subprocess.Popen

class qubesadmin.app.QubesRemote[source]

Bases: QubesBase

Application object communicating through qrexec services.

Used when running in VM.

qubesd_call(dest: str | None, method: str, arg: str | None = None, payload: bytes | None = None, payload_stream: IO | None = None) → bytes[source]

Execute Admin API method.

If payload and payload_stream are both specified, they will be sent in that order.

Parameters:

  • dest – Destination VM name

  • method – Full API method name (‘admin…’)

  • arg – Method argument (if any)

  • payload – Payload send to the method

  • payload_stream – file-like object to read payload from

Returns:

Data returned by qubesd (string)

Warning

payload_stream will get closed by this function

qubesd_connection_type = 'qrexec'

type of qubesd connection: either ‘socket’ or ‘qrexec’

run_service(dest: str, service: str, user: str | None = None, *, filter_esc: bool = False, localcmd: str | None = None, wait: bool = True, autostart: bool = True, prefix_data: bytes | None = None, **kwargs) → Popen[source]

Run qrexec service in a given destination

Parameters:

  • dest (str) – Destination - may be a VM name or empty string for default (for a given service)

  • service (str) – service name

  • filter_esc (bool) – filter escape sequences to protect terminal emulator

  • user (str) – username to run service as

  • localcmd (str) – Command to connect stdin/stdout to

  • wait (bool) – wait for process to finish

Return type:

subprocess.Popen

class qubesadmin.app.VMCollection(app: QubesBase)[source]

Bases: object

Collection of VMs objects

T = ~T
clear_cache(invalidate_name: str | None = None) → None[source]

Clear cached list of VMs If invalidate_name is given, remove that object from cache explicitly too.

get(item: str | QubesVM) → QubesVM[source]
get(item: str | QubesVM, default: T) → QubesVM | T

Get a VM object, or return default if it can’t be found.

get_blind(item: str) → QubesVM[source]

Get a vm without downloading the list and checking if exists

keys() → Iterable[str][source]

Get list of VM names.

refresh_cache(force: bool = False) → None[source]

Refresh cached list of VMs

values() → list[QubesVM][source]

Get list of VM objects.

qubesadmin.base module

Base classes for managed objects

class qubesadmin.base.PropertyHolder(app: QubesBase, method_prefix: str, method_dest: str)[source]

Bases: object

A base class for object having properties retrievable using mgmt API.

Warning: each (non-private) local attribute needs to be defined at class level, even if initialized in __init__; otherwise will be treated as property retrievable using mgmt call.

app: QubesBase
clear_cache() → None[source]

Clear property cache.

clone_properties(src: PropertyHolder, proplist: list[str] | None = None) → None[source]

Clone properties from other object.

Parameters:

  • src (PropertyHolder) – source object

  • proplist (list) – list of properties (None or omit for all properties)

property_get_default(item: str) → Any[source]

Get default property value, regardless of the current value

Parameters:

item (str) – name of property

Returns:

default value

property_help(name: str) → str[source]

Get description of a property.

Returns:

property help text

property_is_default(item: str) → bool[source]

Check if given property have default value

Parameters:

item (str) – name of property

Returns:

bool

property_list() → list[str][source]

List available properties (their names).

Returns:

list of strings

qubesd_call(dest: str | None, method: str, arg: str | None = None, payload: bytes | None = None, payload_stream: BinaryIO | None = None) → bytes[source]

Call into qubesd using appropriate mechanism. This method should be defined by a subclass.

Only one of payload and payload_stream can be specified.

Parameters:

  • dest – Destination VM name

  • method – Full API method name (‘admin…’)

  • arg – Method argument (if any)

  • payload – Payload send to the method

  • payload_stream – file-like object to read payload from

Returns:

Data returned by qubesd (string)

class qubesadmin.base.WrapperObjectsCollection(app: QubesBase, list_method: str, object_class: type[T])[source]

Bases: Generic[T]

Collection of simple named objects

clear_cache(invalidate_name: int | str | None = None) → None[source]

Clear cached list of names. If invalidate_name is given, remove that object from cache explicitly too.

get_blind(item: int | str) → T[source]

Get a property without downloading the list and checking if it’s present

items() → list[tuple[int | str, T]][source]

Get list of (key, value) pairs

keys() → list[int | str][source]

Get list of names.

refresh_cache(force: bool = False) → None[source]

Refresh cached list of names

values() → list[T][source]

Get list of objects

qubesadmin.config module

Configuration variables/constants

qubesadmin.config.QUBESD_SOCKET = '/var/run/qubesd.sock'

path to qubesd socket

qubesadmin.devices module

API for various types of devices.

Main concept is that some domain main expose (potentially multiple) devices, which can be attached to other domains. Devices can be of different classes (like ‘pci’, ‘usb’, etc.). Each device class is implemented by an extension.

Devices are identified by pair of (backend domain, port_id), where port_id is str.

class qubesadmin.devices.DeviceCollection(vm: QubesVM, class_: str)[source]

Bases: object

Bag for devices.

Used as default value for DeviceManager.__missing__() factory.

Parameters:

  • vm – VM for which we manage devices

  • class – device class

assign(assignment: DeviceAssignment) → None[source]

Assign device to domain (add to qubes.xml).

Parameters:

assignment (DeviceAssignment) – device object

attach(assignment: DeviceAssignment) → None[source]

Attach (add) device to domain.

Parameters:

assignment (DeviceAssignment) – device object

clear_cache() → None[source]

Clear cache of available devices.

detach(assignment: DeviceAssignment) → None[source]

Detach (remove) device from domain.

Parameters:

assignment (DeviceAssignment) – device to detach (obtained from assignments())

get_assigned_devices(required_only: bool = False) → Iterable[DeviceAssignment][source]

Devices assigned to this vm (included in qubes.xml).

Safe to access before libvirt bootstrap.

get_attached_devices() → Iterable[DeviceAssignment][source]

List devices which are attached to this vm.

get_dedicated_devices() → Iterable[DeviceAssignment][source]

List devices which are attached or assigned to this vm.

get_exposed_devices() → Iterable[DeviceInfo][source]

List devices exposed by this vm.

unassign(assignment: DeviceAssignment) → None[source]

Unassign device from domain (remove from qubes.xml).

Parameters:

assignment (DeviceAssignment) – device to unassign (obtained from assignments())

update_assignment(device: VirtualDevice, required: AssignmentMode) → None[source]

Update assignment of already attached device.

Parameters:

  • device (VirtualDevice) – device for which change required flag

  • required (bool) – new assignment: None -> unassign device from qube False -> device will be auto-attached to qube True -> device is required to start qube

class qubesadmin.devices.DeviceManager(vm: QubesVM)[source]

Bases: dict

Device manager that hold all devices by their classes.

Parameters:

vm – VM for which we manage devices

allow(*interfaces: Iterable[DeviceInterface]) → None[source]

Remove given interfaces from denied list.

clear_cache() → None[source]

Clear cache of all available device classes

deny(*interfaces: Iterable[DeviceInterface]) → None[source]

Deny a device with any of the given interfaces from attaching to the VM.

keys() → a set-like object providing a view on D's keys[source]

qubesadmin.exc module

Qubes OS exception hierarchy

exception qubesadmin.exc.BackupAlreadyRunningError(message_format: str, *args, **kwargs)[source]

Bases: QubesException

Thrown at user when they try to run the same backup twice at the same time

exception qubesadmin.exc.BackupCancelledError(message_format: str, *args, **kwargs)[source]

Bases: QubesException

Thrown at user when backup was manually cancelled

exception qubesadmin.exc.BackupRestoreError(msg: str, backup_log: bytes | None = None)[source]

Bases: QubesException

Restoring a backup failed

exception qubesadmin.exc.DeviceAlreadyAssigned(message_format: str, *args, **kwargs)[source]

Bases: QubesException, KeyError

Trying to assign already assigned device.

exception qubesadmin.exc.DeviceAlreadyAttached(message_format: str, *args, **kwargs)[source]

Bases: QubesException, KeyError

Trying to attach already attached device.

exception qubesadmin.exc.DeviceNotAssigned(message_format: str, *args, **kwargs)[source]

Bases: QubesException, KeyError

Trying to unassign not assigned device.

exception qubesadmin.exc.PermissionDenied[source]

Bases: Exception

Raised deliberately by handlers when we decide not to cooperate

exception qubesadmin.exc.ProtocolError[source]

Bases: AssertionError

Raised when something is wrong with data received

exception qubesadmin.exc.QubesDaemonAccessError(message_format: str, *args, **kwargs)[source]

Bases: QubesDaemonCommunicationError

Got empty response from qubesd. This can be lack of permission, or some server-side issue.

exception qubesadmin.exc.QubesDaemonCommunicationError(message_format: str, *args, **kwargs)[source]

Bases: QubesException

Error while communicating with qubesd, may mean insufficient permissions as well

qubesadmin.exc.QubesDaemonNoResponseError

alias of QubesDaemonAccessError

exception qubesadmin.exc.QubesException(message_format: str, *args, **kwargs)[source]

Bases: Exception

Exception that can be shown to the user

exception qubesadmin.exc.QubesFeatureNotFoundError(message_format: str, *args, **kwargs)[source]

Bases: QubesException, KeyError

Feature not set for a given domain

exception qubesadmin.exc.QubesLabelNotFoundError(message_format: str, *args, **kwargs)[source]

Bases: QubesException, KeyError

Label does not exists

exception qubesadmin.exc.QubesMemoryError(message_format: str, *args, **kwargs)[source]

Bases: QubesVMError, MemoryError

Cannot start domain, because not enough memory is available

exception qubesadmin.exc.QubesNoSuchPropertyError(message_format: str, *args, **kwargs)[source]

Bases: QubesException, AttributeError

Requested property does not exist

exception qubesadmin.exc.QubesNoTemplateError(message_format: str, *args, **kwargs)[source]

Bases: QubesVMError

Cannot start domain, because there is no template

exception qubesadmin.exc.QubesNotImplementedError(message_format: str, *args, **kwargs)[source]

Bases: QubesException, NotImplementedError

Thrown at user when some feature is not implemented

exception qubesadmin.exc.QubesNotesError(message_format: str, *args, **kwargs)[source]

Bases: QubesException

Some problem with qube notes.

exception qubesadmin.exc.QubesPoolInUseError(message_format: str, *args, **kwargs)[source]

Bases: QubesException

VM is in use, cannot remove.

exception qubesadmin.exc.QubesPropertyAccessError(prop: str)[source]

Bases: QubesDaemonAccessError, AttributeError

Failed to read/write property value, cause is unknown (insufficient permissions, no such property, invalid value, other)

exception qubesadmin.exc.QubesPropertyValueError(message_format: str, *args, **kwargs)[source]

Bases: QubesValueError

Cannot set value of qubes.property, because user-supplied value is wrong.

exception qubesadmin.exc.QubesTagNotFoundError(message_format: str, *args, **kwargs)[source]

Bases: QubesException, KeyError

Tag not set for a given domain

exception qubesadmin.exc.QubesVMAlreadyStartedError(message_format: str, *args, **kwargs)[source]

Bases: QubesVMError

Requested qube to start, but it’s already running

exception qubesadmin.exc.QubesVMError(message_format: str, *args, **kwargs)[source]

Bases: QubesException

Some problem with domain state.

exception qubesadmin.exc.QubesVMInUseError(message_format: str, *args, **kwargs)[source]

Bases: QubesVMError

VM is in use, cannot remove.

exception qubesadmin.exc.QubesVMInvalidUUIDError(message_format: str, *args, **kwargs)[source]

Bases: QubesException

Domain UUID is invalid

exception qubesadmin.exc.QubesVMNotFoundError(message_format: str, *args, **kwargs)[source]

Bases: QubesException, KeyError

Domain cannot be found in the system

exception qubesadmin.exc.QubesVMNotHaltedError(message_format: str, *args, **kwargs)[source]

Bases: QubesVMError

Domain is not halted.

This exception is thrown when machine should be halted, but is not (either running or paused).

exception qubesadmin.exc.QubesVMNotPausedError(message_format: str, *args, **kwargs)[source]

Bases: QubesVMNotStartedError

Domain is not paused.

This exception is thrown when machine should be paused, but is not.

exception qubesadmin.exc.QubesVMNotRunningError(message_format: str, *args, **kwargs)[source]

Bases: QubesVMNotStartedError

Domain is not running.

This exception is thrown when machine should be running but is either halted or paused.

exception qubesadmin.exc.QubesVMNotStartedError(message_format: str, *args, **kwargs)[source]

Bases: QubesVMError

Domain is not started.

This exception is thrown when machine is halted, but should be started (that is, either running or paused).

exception qubesadmin.exc.QubesVMNotSuspendedError(message_format: str, *args, **kwargs)[source]

Bases: QubesVMError

Domain is not suspended.

This exception is thrown when machine should be suspended but is either halted or running.

exception qubesadmin.exc.QubesVMShutdownTimeoutError(message_format: str, *args, **kwargs)[source]

Bases: QubesVMError

Domain shutdown timed out.

exception qubesadmin.exc.QubesValueError(message_format: str, *args, **kwargs)[source]

Bases: QubesException, ValueError

Cannot set some value, because it is invalid, out of bounds, etc.

exception qubesadmin.exc.StoragePoolException(message_format: str, *args, **kwargs)[source]

Bases: QubesException

A general storage exception

exception qubesadmin.exc.UnexpectedDeviceProperty(message_format: str, *args, **kwargs)[source]

Bases: QubesException, ValueError

Device has unexpected property such as backend_domain, devclass etc.

exception qubesadmin.exc.UnrecognizedDevice(message_format: str, *args, **kwargs)[source]

Bases: QubesException, ValueError

Device identity is not as expected.

qubesadmin.features module

VM features interface

class qubesadmin.features.Features(vm: QubesVM)[source]

Bases: object

Manager of the features.

Features can have three distinct values: no value (not present in mapping, which is closest thing to None), empty string (which is interpreted as False) and non-empty string, which is True. Anything assigned to the mapping is coerced to strings, however if you assign instances of bool, they are converted as described above. Be aware that assigning the number 0 (which is considered false in Python) will result in string ‘0’, which is considered true.

NO_DEFAULT = <object object>
check_with_template(item: str) → str | None[source]
check_with_template(item: str, default: T) → str | T

Check if the vm’s template has the specified feature.

get(item: str) → str | None[source]
get(item: str, default: T) → str | T

Get a feature, return default value if missing.

items() → Generator[tuple[str, str]][source]

Return iterable of pairs (feature, value)

keys() → Iterator[str]

qubesadmin.firewall module

Firewall configuration interface

class qubesadmin.firewall.Action(value: object)[source]

Bases: RuleChoice

Rule action

accept: str = 'accept'
drop: str = 'drop'
property rule: str

API representation of this rule element

class qubesadmin.firewall.Comment(value: object)[source]

Bases: RuleOption

User comment

property rule: str

API representation of this rule element

class qubesadmin.firewall.DstHost(value: str, prefixlen: int | None = None)[source]

Bases: RuleOption

Represent host/network address: either IPv4, IPv6, or DNS name

property rule: str | None

API representation of this rule element

class qubesadmin.firewall.DstPorts(value: int | str)[source]

Bases: RuleOption

Destination port(s), for TCP/UDP only

property rule: str

API representation of this rule element

class qubesadmin.firewall.Expire(value: SupportsInt)[source]

Bases: RuleOption

Rule expire time

property expired: bool

Has this rule expired already?

property pretty_value: str

Human readable representation

property rule: str

API representation of this rule element

class qubesadmin.firewall.Firewall(vm: QubesVM)[source]

Bases: object

Firewal manager for a VM

load_rules() → None[source]

Force (re-)loading firewall rules

property policy: Action

Default action to take if no rule matches

reload() → None[source]

Force reload the same firewall rules.

Can be used for example to force again names resolution.

property rules: list[Rule]

Firewall rules

You can either copy them, edit and then assign new rules list to this property, or edit in-place and call save_rules(). Once rules are loaded, they are cached. To reload rules, call load_rules().

save_rules(rules: list[Rule] | None = None) → None[source]

Save firewall rules. Needs to be called after in-place editing rules.

class qubesadmin.firewall.IcmpType(value: SupportsInt)[source]

Bases: RuleOption

ICMP packet type

property rule: str

API representation of this rule element

class qubesadmin.firewall.Proto(value: object)[source]

Bases: RuleChoice

Protocol name

icmp: str = 'icmp'
property rule: str

API representation of this rule element

tcp: str = 'tcp'
udp: str = 'udp'
class qubesadmin.firewall.Rule(rule: str | None, **kwargs)[source]

Bases: object

A single firewall rule

property action: Action | None

rule action

property comment: Comment | None

User comment

property dsthost: DstHost | None

destination host/network

property dstports: DstPorts | None

‘Destination port(s) (for ‘tcp’ and ‘udp’ protocol only)

property expire: Expire | None

Timestamp (UNIX epoch) on which this rule expire

property icmptype: IcmpType | None

ICMP packet type (for ‘icmp’ protocol only)

property proto: Proto | None

protocol to match

property rule: str

API representation of this rule

property specialtarget: SpecialTarget | None

Special target, for now only ‘dns’ supported

class qubesadmin.firewall.RuleChoice(value: object)[source]

Bases: RuleOption

Base class for multiple-choices rule elements

class qubesadmin.firewall.RuleOption(value: object)[source]

Bases: object

Base class for a single rule element

property pretty_value: str

Human readable representation

property rule: str

API representation of this rule element

class qubesadmin.firewall.SpecialTarget(value: object)[source]

Bases: RuleChoice

Special destination

dns: str = 'dns'
property rule: str

API representation of this rule element

qubesadmin.label module

VM Labels

class qubesadmin.label.Label(app: QubesBase, name: str)[source]

Bases: object

Label definition for virtual machines

Label specifies colour of the qube icon displayed next to VM’s name.

Parameters:

  • color (str) – colour specification as in HTML (#abcdef)

  • name (str) – label’s name like “red” or “green”

property color: str

color specification as in HTML (#abcdef)

property icon: str

freedesktop icon name, suitable for use in PyQt4.QtGui.QIcon.fromTheme()

property index: int

label numeric identifier

property name: str

label’s name like “red” or “green”

qubesadmin.log module

Qubes logging routines

See also: qubes.vm.qubesvm.QubesVM.log

qubesadmin.log.enable() → None[source]

Enable global logging

Use logging module from standard library to log messages.

>>> import qubes.log
>>> qubes.log.enable()
>>> import logging
>>> logging.warning('Foobar')
qubesadmin.log.enable_debug() → None[source]

Enable debug logging

Enable more messages and additional info to message format.

qubesadmin.spinner module

Qubes CLI spinner

A novice asked the master: “In the east there is a great tree-structure that men call ‘Corporate Headquarters’. It is bloated out of shape with vice presidents and accountants. It issues a multitude of memos, each saying ‘Go, Hence!’ or ‘Go, Hither!’ and nobody knows what is meant. Every year new names are put onto the branches, but all to no avail. How can such an unnatural entity be?”

The master replied: “You perceive this immense structure and are disturbed that it has no rational purpose. Can you not take amusement from its endless gyrations? Do you not enjoy the untroubled ease of programming beneath its sheltering branches? Why are you bothered by its uselessness?”

(Geoffrey James, “The Tao of Programming”, 7.1)

class qubesadmin.spinner.AbstractSpinner(stream: IO, charset: str = '-\\|/')[source]

Bases: object

The base class for all Spinners

Parameters:

  • stream – file-like object with .write() method

  • charset (str) – the sequence of characters to display

The spinner should be used as follows:

  1. exactly one call to show()

  2. zero or more calls to update()

  3. exactly one call to hide()

hide() → None[source]

Hide the spinner and the prompt

show(prompt: str) → None[source]

Show the spinner, with a prompt

Parameters:

prompt (str) – prompt, like “please wait”

update() → None[source]

Show next spinner character

class qubesadmin.spinner.DummySpinner(stream: IO, charset: str = '-\\|/')[source]

Bases: AbstractSpinner

Dummy spinner, does not do anything

hide() → None[source]

Hide the spinner and the prompt

show(prompt: str) → None[source]

Show the spinner, with a prompt

Parameters:

prompt (str) – prompt, like “please wait”

update() → None[source]

Show next spinner character

class qubesadmin.spinner.QubesSpinner(*args, **kwargs)[source]

Bases: AbstractSpinner

Basic spinner

This spinner uses standard ASCII control characters

hide() → None[source]

Hide the spinner and the prompt

show(prompt: str) → None[source]

Show the spinner, with a prompt

Parameters:

prompt (str) – prompt, like “please wait”

update() → None[source]

Show next spinner character

class qubesadmin.spinner.QubesSpinnerEnterpriseEdition(stream: IO, charset: str | None = None)[source]

Bases: QubesSpinner

Enterprise spinner

This is tty- and terminfo-aware spinner. Recommended.

hide() → None[source]

Hide the spinner and the prompt

qubesadmin.storage module

Storage subsystem.

class qubesadmin.storage.Pool(app: QubesBase, name: str)[source]

Bases: object

A Pool is used to manage different kind of volumes (File based/LVM/Btrfs/…).

property config: dict[str, str]

Storage pool config

property driver: str

Storage pool driver

property ephemeral_volatile: bool

Whether volatile volumes in this pool should be encrypted with an ephemeral key in dom0

property revisions_to_keep: int

Number of revisions to keep around

property size: int | None

Storage pool size, in bytes

property usage: int | None

Space used in the pool, in bytes

property usage_details: dict[str, int]

Storage pool usage details (current - not cached)

property volumes: Generator[Volume]

Volumes managed by this pool

class qubesadmin.storage.Volume(app: QubesBase, pool: str | None = None, vid: str | None = None, vm: str | None = None, vm_name: str | None = None)[source]

Bases: object

Storage volume.

clear_data() → None[source]

Clear existing volume content.

clone(source: Volume) → None[source]

Clone data from sane volume of another VM.

This function override existing volume content. This operation is implemented for VM volumes - those in vm.volumes collection (not pool.volumes).

Parameters:

source – source volume object

property ephemeral: bool

True if volume is read-write.

import_data(stream: BinaryIO) → None[source]

Import volume data from a given file-like object.

This function overrides existing volume content.

Parameters:

stream – file-like object, must support fileno()

import_data_with_size(stream: IO, size: object) → None[source]

Import volume data from a given file-like object, informing qubesd that data has a specific size.

This function overrides existing volume content.

Parameters:

  • stream – file-like object, must support fileno()

  • size – size of data in bytes

is_outdated() → bool[source]

Returns True if this snapshot of a source volume (for `snap_on_start`=True) is outdated.

property name: str | None

per-VM volume name, if available

property pool: str

Storage volume pool name.

resize(size: object) → None[source]

Resize volume.

Currently only extending is supported.

Parameters:

size (int) – new size in bytes.

revert(revision: str) → None[source]

Revert volume to previous revision

Parameters:

revision (str) – Revision identifier to revert to

property revisions: list[str]

Returns iterable containing revision identifiers

property revisions_to_keep: int

Number of revisions to keep around

property rw: bool

True if volume is read-write.

property save_on_stop: bool

Commit changes to original volume on VM stop.

property size: int

Size of volume, in bytes.

property snap_on_start: bool

Create a snapshot from source on VM start.

property source: str | None

Volume ID of source volume (for snap_on_start).

If None, this volume itself will be used.

property usage: int

Used volume space, in bytes.

property vid: str

Storage volume id, unique within given pool.

qubesadmin.tags module

VM tags interface

class qubesadmin.tags.Tags(vm: QubesVM)[source]

Bases: object

Manager of the tags.

Tags are simple: tag either can be present on qube or not. Tag is a simple string consisting of ASCII alphanumeric characters, plus _ and -.

add(elem: str) → None[source]

Add a tag

discard(elem: str) → None[source]

Remove a tag if present

remove(elem: str) → None[source]

Remove a tag

update(*others) → None[source]

Add tags from iterable(s)

qubesadmin.utils module

Various utility functions.

class qubesadmin.utils.DriveAction(option_strings, dest='drive', *, prefix='cdrom:', metavar='IMAGE', required=False, help='Attach drive')[source]

Bases: Action

Action for argument parser that stores drive image path. Intended use for device attachment before domain is started.

class qubesadmin.utils.LockFile(path: str, nonblock: bool = False)[source]

Bases: object

Simple locking context manager. It opens a file with an advisory lock taken (fcntl.lockf)

acquire() → None[source]

Lock the opened file

release() → None[source]

Unlock the file and close the file object

qubesadmin.utils.bytes_to_kmg(size: int) → str[source]

Convert bytes to human readable format.

qubesadmin.utils.encode_for_vmexec(args: Iterable[str]) → str[source]

Encode an argument list for qubes.VMExec call.

async qubesadmin.utils.generic_action(domains: list[QubesVM], action: str | Callable, *args, **kwargs)[source]

Asynchronously run action on qubes and return ones that failed.

qubesadmin.utils.get_drive_assignment(app, drive_str)[source]

Prepare qubesadmin.device_protocol.DeviceAssignment object for a given drive. Intended to be used during before domain is started.

If running in dom0, it will also take care about creating the appropriate loop device (if necessary). Otherwise, only existing block devices are supported.

Parameters:

  • app – Qubes() instance

  • drive_str – drive argument

Returns:

DeviceAssignment matching drive_str

qubesadmin.utils.kbytes_to_kmg(size: float | int) → str[source]

Convert kbytes to human readable format.

async qubesadmin.utils.kill(domains: list[QubesVM], **kwargs)[source]

Asynchronously kill qubes and return ones that failed.

qubesadmin.utils.mbytes_to_kmg(size: float | int) → str[source]

Convert mbytes to human readable format.

qubesadmin.utils.parse_size(size: str) → int[source]

Parse human readable size into bytes.

async qubesadmin.utils.pause(domains: list[QubesVM], **kwargs)[source]

Asynchronously pause qubes and return ones that failed.

qubesadmin.utils.qbool(value: str | int | bool) → bool[source]

Property setter for boolean properties.

It accepts (case-insensitive) '0', 'no' and false as False and '1', 'yes' and 'true' as True.

async qubesadmin.utils.resume(domains: list[QubesVM], **kwargs)[source]

Asynchronously resume qubes and return ones that failed.

async qubesadmin.utils.shutdown(domains: list[QubesVM], **kwargs)[source]

Asynchronously resume shutdown and return ones that failed.

qubesadmin.utils.size_to_human(size: int) → str[source]

Humane readable size, with 1/10 precision

async qubesadmin.utils.start(domains: list[QubesVM], **kwargs)[source]

Asynchronously start qubes and return ones that failed.

qubesadmin.utils.start_expert(domain, skip_if_running: bool = False, drive: str | None = None)[source]

Start the domain, optionally specifying a drive.

async qubesadmin.utils.suspend(domains: list[QubesVM], **kwargs)[source]

Asynchronously suspend qubes and return ones that failed.

async qubesadmin.utils.unpause(domains: list[QubesVM], **kwargs)[source]

Asynchronously unpause qubes and return ones that failed.

qubesadmin.utils.updates_vms_status(qvm_collection: QubesBase) → bool | None[source]

Check whether all VMs have the same check-updates value; if yes, return it; otherwise, return None

qubesadmin.utils.vm_dependencies(app: QubesBase, reference_vm: QubesVM) → list[tuple[QubesVM | None, str]][source]

Helper function that returns a list of all the places a given VM is used in. Output is a list of tuples (property_holder, property_name), with None as property_holder for global properties

Module contents

Qubes OS management client.