# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Brightbox Driver
"""
import base64
from libcloud.utils.py3 import b, httplib
from libcloud.compute.base import Node, NodeSize, NodeImage, NodeDriver, NodeLocation
from libcloud.compute.types import Provider, NodeState
from libcloud.common.brightbox import BrightboxConnection
API_VERSION = "1.0"
def _extract(d, keys):
return {k: d[k] for k in keys if k in d and d[k] is not None}
[docs]class BrightboxNodeDriver(NodeDriver):
"""
Brightbox node driver
"""
connectionCls = BrightboxConnection
type = Provider.BRIGHTBOX
name = "Brightbox"
website = "http://www.brightbox.co.uk/"
NODE_STATE_MAP = {
"creating": NodeState.PENDING,
"active": NodeState.RUNNING,
"inactive": NodeState.UNKNOWN,
"deleting": NodeState.UNKNOWN,
"deleted": NodeState.TERMINATED,
"failed": NodeState.UNKNOWN,
"unavailable": NodeState.UNKNOWN,
}
def __init__(
self,
key,
secret=None,
secure=True,
host=None,
port=None,
api_version=API_VERSION,
**kwargs,
):
super().__init__(
key=key,
secret=secret,
secure=secure,
host=host,
port=port,
api_version=api_version,
**kwargs,
)
def _to_node(self, data):
extra_data = _extract(
data,
[
"fqdn",
"user_data",
"status",
"interfaces",
"snapshots",
"server_groups",
"hostname",
"started_at",
"created_at",
"deleted_at",
],
)
extra_data["zone"] = self._to_location(data["zone"])
ipv6_addresses = [
interface["ipv6_address"]
for interface in data["interfaces"]
if "ipv6_address" in interface
]
private_ips = [
interface["ipv4_address"]
for interface in data["interfaces"]
if "ipv4_address" in interface
]
public_ips = [cloud_ip["public_ip"] for cloud_ip in data["cloud_ips"]]
public_ips += ipv6_addresses
return Node(
id=data["id"],
name=data["name"],
state=self.NODE_STATE_MAP[data["status"]],
private_ips=private_ips,
public_ips=public_ips,
driver=self.connection.driver,
size=self._to_size(data["server_type"]),
image=self._to_image(data["image"]),
extra=extra_data,
)
def _to_image(self, data):
extra_data = _extract(
data,
[
"arch",
"compatibility_mode",
"created_at",
"description",
"disk_size",
"min_ram",
"official",
"owner",
"public",
"source",
"source_type",
"status",
"username",
"virtual_size",
"licence_name",
],
)
if data.get("ancestor", None):
extra_data["ancestor"] = self._to_image(data["ancestor"])
return NodeImage(id=data["id"], name=data["name"], driver=self, extra=extra_data)
def _to_size(self, data):
return NodeSize(
id=data["id"],
name=data["name"],
ram=data["ram"],
disk=data["disk_size"],
bandwidth=0,
price=0,
driver=self,
)
def _to_location(self, data):
if data:
return NodeLocation(id=data["id"], name=data["handle"], country="GB", driver=self)
else:
return None
def _post(self, path, data={}):
headers = {"Content-Type": "application/json"}
return self.connection.request(path, data=data, headers=headers, method="POST")
def _put(self, path, data={}):
headers = {"Content-Type": "application/json"}
return self.connection.request(path, data=data, headers=headers, method="PUT")
[docs] def create_node(self, name, size, image, location=None, ex_userdata=None, ex_servergroup=None):
"""Create a new Brightbox node
Reference: https://api.gb1.brightbox.com/1.0/#server_create_server
@inherits: :class:`NodeDriver.create_node`
:keyword ex_userdata: User data
:type ex_userdata: ``str``
:keyword ex_servergroup: Name or list of server group ids to
add server to
:type ex_servergroup: ``str`` or ``list`` of ``str``
"""
data = {
"name": name,
"server_type": size.id,
"image": image.id,
}
if ex_userdata:
data["user_data"] = base64.b64encode(b(ex_userdata)).decode("ascii")
if location:
data["zone"] = location.id
if ex_servergroup:
if not isinstance(ex_servergroup, list):
ex_servergroup = [ex_servergroup]
data["server_groups"] = ex_servergroup
data = self._post("/%s/servers" % self.api_version, data).object
return self._to_node(data)
[docs] def destroy_node(self, node):
response = self.connection.request(
"/{}/servers/{}".format(self.api_version, node.id), method="DELETE"
)
return response.status == httplib.ACCEPTED
[docs] def list_nodes(self):
data = self.connection.request("/%s/servers" % self.api_version).object
return list(map(self._to_node, data))
[docs] def list_images(self, location=None):
data = self.connection.request("/%s/images" % self.api_version).object
return list(map(self._to_image, data))
[docs] def list_sizes(self):
data = self.connection.request("/%s/server_types" % self.api_version).object
return list(map(self._to_size, data))
[docs] def list_locations(self):
data = self.connection.request("/%s/zones" % self.api_version).object
return list(map(self._to_location, data))
[docs] def ex_list_cloud_ips(self):
"""
List Cloud IPs
@note: This is an API extension for use on Brightbox
:rtype: ``list`` of ``dict``
"""
return self.connection.request("/%s/cloud_ips" % self.api_version).object
[docs] def ex_create_cloud_ip(self, reverse_dns=None):
"""
Requests a new cloud IP address for the account
@note: This is an API extension for use on Brightbox
:param reverse_dns: Reverse DNS hostname
:type reverse_dns: ``str``
:rtype: ``dict``
"""
params = {}
if reverse_dns:
params["reverse_dns"] = reverse_dns
return self._post("/%s/cloud_ips" % self.api_version, params).object
[docs] def ex_update_cloud_ip(self, cloud_ip_id, reverse_dns):
"""
Update some details of the cloud IP address
@note: This is an API extension for use on Brightbox
:param cloud_ip_id: The id of the cloud ip.
:type cloud_ip_id: ``str``
:param reverse_dns: Reverse DNS hostname
:type reverse_dns: ``str``
:rtype: ``dict``
"""
response = self._put(
"/{}/cloud_ips/{}".format(self.api_version, cloud_ip_id),
{"reverse_dns": reverse_dns},
)
return response.status == httplib.OK
[docs] def ex_map_cloud_ip(self, cloud_ip_id, interface_id):
"""
Maps (or points) a cloud IP address at a server's interface
or a load balancer to allow them to respond to public requests
@note: This is an API extension for use on Brightbox
:param cloud_ip_id: The id of the cloud ip.
:type cloud_ip_id: ``str``
:param interface_id: The Interface ID or LoadBalancer ID to
which this Cloud IP should be mapped to
:type interface_id: ``str``
:return: True if the mapping was successful.
:rtype: ``bool``
"""
response = self._post(
"/{}/cloud_ips/{}/map".format(self.api_version, cloud_ip_id),
{"destination": interface_id},
)
return response.status == httplib.ACCEPTED
[docs] def ex_unmap_cloud_ip(self, cloud_ip_id):
"""
Unmaps a cloud IP address from its current destination making
it available to remap. This remains in the account's pool
of addresses
@note: This is an API extension for use on Brightbox
:param cloud_ip_id: The id of the cloud ip.
:type cloud_ip_id: ``str``
:return: True if the unmap was successful.
:rtype: ``bool``
"""
response = self._post("/{}/cloud_ips/{}/unmap".format(self.api_version, cloud_ip_id))
return response.status == httplib.ACCEPTED
[docs] def ex_destroy_cloud_ip(self, cloud_ip_id):
"""
Release the cloud IP address from the account's ownership
@note: This is an API extension for use on Brightbox
:param cloud_ip_id: The id of the cloud ip.
:type cloud_ip_id: ``str``
:return: True if the unmap was successful.
:rtype: ``bool``
"""
response = self.connection.request(
"/{}/cloud_ips/{}".format(self.api_version, cloud_ip_id), method="DELETE"
)
return response.status == httplib.OK