mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-08 19:26:48 +00:00
491819ed64
vrtManager's errors (some 140: a disk that does not move, a network address outside its subnet, a host device given to another VM, the disks a migration cannot copy) were English f-strings, so no catalog could translate them, on pages or in jobs. They are marked texts with their values now; a variant that differs by a word (internal or external, running or persistent, with a subnet or without) is a whole message of its own, and a list of problems is a list of messages. OperationError reads in the active language, so the pages that show one translate it as they are; its args stay English, and so do the logs and the audit log. Host device problems and the Migrate tab's reasons a disk is not copied are messages too.
119 lines
5.5 KiB
Python
119 lines
5.5 KiB
Python
from django.utils.translation import gettext_noop
|
|
import base64
|
|
|
|
from libvirt import VIR_ERR_NO_STORAGE_POOL, VIR_STORAGE_XML_INACTIVE, libvirtError
|
|
from lxml import etree
|
|
from vrtManager import util
|
|
from vrtManager.connection import wvmConnect
|
|
|
|
# the element of <usage> that holds the usage, per usage type
|
|
USAGE_ELEMENTS = {"ceph": "name", "volume": "volume", "iscsi": "target", "tls": "name", "vtpm": "name"}
|
|
|
|
# the usage type a reference by usage means: the type the <secret> gives, or
|
|
# for a pool's <auth> the auth type; a disk's encryption passphrase is a volume's
|
|
REF_USAGE_TYPES = {"ceph": "ceph", "iscsi": "iscsi", "chap": "iscsi", "passphrase": "volume"}
|
|
|
|
|
|
def value_bytes(usage_type, value):
|
|
"""The bytes a value typed on the page stands for: a Ceph key as
|
|
`ceph auth get-key` prints it, in base64; any other secret's password or
|
|
passphrase as it is typed."""
|
|
if usage_type != "ceph":
|
|
return value.encode()
|
|
try:
|
|
return base64.b64decode(value, validate=True)
|
|
except ValueError: # not base64, or not ASCII; the message never holds the value
|
|
raise util.OperationError(gettext_noop("A Ceph key is base64 text, as `ceph auth get-key` prints it")) from None
|
|
|
|
|
|
def secrets_named(tree, path, by_usage):
|
|
"""UUIDs of the secrets the <secret> elements at path name, by UUID or by
|
|
usage; by_usage maps (usage type, usage) to a UUID."""
|
|
uuids = set()
|
|
for ref in tree.xpath(path):
|
|
if ref.get("uuid"):
|
|
uuids.add(ref.get("uuid"))
|
|
else:
|
|
ref_type = ref.get("type") or ref.getparent().get("type")
|
|
uuids.add(by_usage.get((REF_USAGE_TYPES.get(ref_type), ref.get("usage"))))
|
|
return uuids
|
|
|
|
|
|
class wvmSecrets(wvmConnect):
|
|
def pool_trees(self):
|
|
"""Each pool's name with its definitions, parsed: the live one, and
|
|
for a persistent pool the one saved for its next start too, read
|
|
whether it is active or not (it may stop meanwhile). A pool gone
|
|
meanwhile is left out."""
|
|
for pool in self.wvm.listAllStoragePools(0):
|
|
try:
|
|
xmls = [pool.XMLDesc(0)]
|
|
if pool.isPersistent():
|
|
xmls.append(pool.XMLDesc(VIR_STORAGE_XML_INACTIVE))
|
|
except libvirtError as err:
|
|
if err.get_error_code() != VIR_ERR_NO_STORAGE_POOL:
|
|
raise
|
|
continue
|
|
yield pool.name(), [etree.fromstring(xml) for xml in xmls]
|
|
|
|
def get_users(self, secrets):
|
|
"""The VMs and pools that name each of secrets (get_secrets_info), by
|
|
UUID: a VM's disks, host devices and vTPM, a pool's source, in any of
|
|
their definitions. Not seen: a LUKS volume's own encryption, the TLS
|
|
settings of qemu.conf, snapshots."""
|
|
by_usage = {(secret["usage_type"], secret["usage"]): secret["uuid"] for secret in secrets}
|
|
users = {}
|
|
|
|
def add(kind, name, uuids):
|
|
for uuid in uuids - {None}:
|
|
users.setdefault(uuid, {"vms": [], "pools": []})[kind].append(name)
|
|
|
|
for name, trees in self.domain_trees():
|
|
uuids = set()
|
|
for tree in trees:
|
|
uuids.update(secrets_named(tree, "devices//secret", by_usage))
|
|
uuids.update(tree.xpath("devices/tpm/backend/encryption/@secret"))
|
|
add("vms", name, uuids)
|
|
for name, trees in self.pool_trees():
|
|
uuids = set()
|
|
for tree in trees:
|
|
uuids.update(secrets_named(tree, "source//secret", by_usage))
|
|
add("pools", name, uuids)
|
|
return users
|
|
|
|
def create_secret(self, usage_type, usage, value="", ephemeral=False, private=False):
|
|
"""A secret for usage, with value if one is given: a Ceph key that is
|
|
not base64 defines nothing, and a value libvirt refuses undefines the
|
|
new secret again."""
|
|
value = value_bytes(usage_type, value) if value else None
|
|
# built with lxml: the usage is text, whatever characters it holds
|
|
secret = etree.Element("secret", ephemeral="yes" if ephemeral else "no", private="yes" if private else "no")
|
|
usage_element = etree.SubElement(secret, "usage", type=usage_type)
|
|
etree.SubElement(usage_element, USAGE_ELEMENTS[usage_type]).text = usage
|
|
defined = self.wvm.secretDefineXML(etree.tostring(secret, encoding="unicode"))
|
|
if value is not None:
|
|
try:
|
|
defined.setValue(value)
|
|
except libvirtError:
|
|
defined.undefine()
|
|
raise
|
|
|
|
def set_secret_value(self, uuid, value):
|
|
"""value as typed on the page; its bytes follow the secret's usage
|
|
type (value_bytes)."""
|
|
secret = self.wvm.secretLookupByUUIDString(uuid)
|
|
usage_type = etree.fromstring(secret.XMLDesc(0)).xpath("string(usage/@type)")
|
|
secret.setValue(value_bytes(usage_type, value))
|
|
|
|
def delete_secret(self, uuid):
|
|
"""Refused while a VM or a pool names it (get_users): without it that
|
|
VM or pool would no longer start."""
|
|
secret = self.wvm.secretLookupByUUIDString(uuid)
|
|
if users := self.get_users(self.get_secrets_info()).get(uuid):
|
|
named = []
|
|
if users["vms"]:
|
|
named.append(util.message(gettext_noop("VMs %(names)s"), names=", ".join(users["vms"])))
|
|
if users["pools"]:
|
|
named.append(util.message(gettext_noop("pools %(names)s"), names=", ".join(users["pools"])))
|
|
raise util.OperationError(gettext_noop("Secret %(uuid)s is used by %(named)s; remove it from them first"), uuid=uuid, named=named)
|
|
secret.undefine()
|