Async python client for LXD REST API (currently under heavy development).
pip install lxdimport asyncio
from pathlib import Path
from lxd import LXDTransport, LXDClient
async def main():
async with LXDTransport(
endpoint_url='https://mylxd:8443/',
cert_path=Path('~/.config/lxc/client.crt'),
key_path=Path('~/.config/lxc/client.key'),
endpoint_cert_path=Path('~/.config/lxc/servercerts/mylxd.crt'),
) as transport:
client = LXDClient(transport)
await client.authenticate(
cert_path=Path('~/.config/lxc/client.crt'),
password='your-trust-password'
)
asyncio.run(main())# Recursion 0 returns only links to objects,
# you can resolve them by awaiting
instance_links = await client.instances.list(recursion=0)
instance = await instance_links[0]
# Recursion 1 returns only some fields
instances = await client.instances.list(recursion=1)
# Recursion 2 returns all possible information
instances = await client.instances.list(recursion=2)from lxd.entities.instances import InstanceAction
instances = await client.instances.list()
operation = await client.instances.update_state(
instances[0].name, action=InstanceAction.STOP
)
await client.operations.wait(operation.id) # wait as long as possible
await client.operations.wait(operation.id, timeout=30) # 30sasync for event in client.server.get_events():
# See Event object for more properties
print(event.type)
print(event.metadata)Get server environment and configuration.
# See lxd.entities.server.Server
info = await client.server.get()
print(info.config)
print(info.environment)Update the entire server configuration.
await client.server.update_configuration({
'core.https_address': '0.0.0.0:8443'
'core.trust_password': 'very-strong-password'
})You may need to send int values as str, otherwise lxd would not be able to parse them:
await client.server.update_configuration({
# 2 (int) value would cause error:
# 'cannot set \'images.remote_cache_expiry\': invalid type float64'
'images.remote_cache_expiry': '2'
})Update a subset of the server configuration.
await client.server.update_configuration_subset({
'core.trust_password': 'very-strong-password'
})You may need to send int values as str, otherwise lxd would not be able to parse them:
await client.server.update_configuration({
# 2 (int) value would cause error:
# 'cannot set \'images.remote_cache_expiry\': invalid type float64'
'images.remote_cache_expiry': '2'
})Gets the hardware information profile of the LXD server.
# See lxd.entities.server.ServerResources
server_resources = await client.server.get_resources()
print(server_resources.cpu)Connect to event API using websocket.
# Listen all events
async for event in client.server.get_events():
print(event.type)
print(event.metadata)
# Listen to specific events
async for event in client.server.get_events(type='operation'):
print(event.metadata.id)
print(event.metadata.status)Returns a list of trusted certificates.
# See lxd.entities.certificates.Certificate
certs = await client.certificates.list()
print(certs[0].fingerprint)If you pass recursion=0 parameter, lxd would return just references,
which are represented in current module as
lxd.entities.certificates.CertificateLink objects.
If you await such link object - you would get object itself (separate http
request is performed for every await call).
cert_links = await client.certificates.list(recursion=0)
certs = await asyncio.gather(*cert_links)Gets a specific certificate entry from the trust store by fingerprint.
from cryptography.x509 import load_pem_x509_certificate
from cryptography.hazmat.primitives import hashes
fprint = '97f267c0fe20fd013b6b4ba3f5440ea3e9361ce8568d41c633f28c620ab37ea0'
cert = await client.certificates.get(fprint)
cert_obj = load_pem_x509_certificate(cert.certificate.encode())
assert cert_obj.fingerprint(hashes.SHA256()).hex() == fprintAdds a certificate to the trust store as trusted user (client certificate should be trusted).
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
private_key = rsa.generate_private_key(
public_exponent=65537, key_size=2048, backend=default_backend()
)
subj = x509.Name([
x509.NameAttribute(NameOID.COMMON_NAME, "alvassin@osx")
])
cert = x509.CertificateBuilder().subject_name(
subj
).issuer_name(
subj
).public_key(
private_key.public_key()
).serial_number(
x509.random_serial_number()
).not_valid_before(
datetime.utcnow()
).not_valid_after(
datetime.utcnow() + timedelta(days=365)
).sign(
private_key=private_key,
algorithm=hashes.SHA256(),
backend=default_backend()
)
await client.certificates.add(
cert.public_bytes(serialization.Encoding.PEM)
)If password argument is specified, adds a certificate to the trust store
as an untrusted user.
await client.certificates.add(
cert.public_bytes(serialization.Encoding.PEM),
password='your-trust-password'
)Update the entire certificate configuration.
await client.certificates.update_configuration(
'97f267c0fe20fd013b6b4ba3f5440ea3e9361ce8568d41c633f28c620ab37ea0',
certificate='-----BEGIN CERTIFICATE-----\n...',
name='new-name',
projects=[],
restricted=False,
type='client'
)Update a subset of the certificate configuration.
await client.certificates.update_configuration_subset(
'97f267c0fe20fd013b6b4ba3f5440ea3e9361ce8568d41c633f28c620ab37ea0',
name='another-name'
)Removes the certificate from the trust store.
await client.certificates.remove(
'97f267c0fe20fd013b6b4ba3f5440ea3e9361ce8568d41c633f28c620ab37ea0'
)Get instances list.
Get a specific instance (basic struct).
Create a new instance on LXD.
Depending on the source, this can create an instance from an existing local image, remote image, existing local instance or snapshot, remote migration stream or backup file.
Deletes a specific instance and anything owned by the instance, such as snapshots and backups.
Gets the runtime state of the instance.
This is a reasonably expensive call as it causes code to be run inside of the instance to retrieve the resource usage and network information.
Changes the running state of the instance.
Get a list of operations.
Get specific operation state.
Wait for the operation to reach a final state (or timeout) and retrieve its final state.
Cancel the operation (if supported by operation).
- Add filtering support.