Add Firewall support (#111)

Signed-off-by: Lukas Kämmerling <lukas.kaemmerling@hetzner-cloud.de>

Author: LKaemmerling

.gitlab-ci.yml

@@ -44,7 +44,7 @@ python38:
 
 python39:
   <<: *tests_template
-  image: python:3.9-rc-alpine
+  image: python:3.9-alpine
   script: tox -e py39
 
 test-style:

docs/api.clients.firewalls.rst

@@ -0,0 +1,22 @@
+FirewallsClient
+==================
+
+
+.. autoclass:: hcloud.firewalls.client.FirewallsClient
+    :members:
+
+.. autoclass:: hcloud.firewalls.client.BoundFirewall
+    :members:
+
+.. autoclass:: hcloud.firewalls.domain.Firewall
+    :members:
+
+.. autoclass:: hcloud.firewalls.domain.FirewallRule
+    :members:
+
+.. autoclass:: hcloud.firewalls.domain.FirewallResource
+    :members:
+
+.. autoclass:: hcloud.firewalls.domain.CreateFirewallResponse
+    :members:
+

hcloud/firewalls/__init__.py

@@ -0,0 +1 @@
+# -*- coding: utf-8 -*-

hcloud/firewalls/client.py

@@ -0,0 +1,163 @@
+# -*- coding: utf-8 -*-
+from dateutil.parser import isoparse
+
+from hcloud.core.domain import BaseDomain
+
+
+class Firewall(BaseDomain):
+    """Firewall Domain
+
+    :param id: int
+           ID of the Firewall
+    :param name: str
+           Name of the Firewall
+    :param labels: dict
+           User-defined labels (key-value pairs)
+    :param rules: List[:class:`FirewallRule <hcloud.firewalls.domain.FirewallRule>`]
+           Rules of the Firewall
+    :param applied_to: List[:class:`FirewallResource <hcloud.firewalls.domain.FirewallResource>`]
+           Resources currently using the Firewall
+    :param created: datetime
+           Point in time when the image was created
+    """
+    __slots__ = (
+        "id",
+        "name",
+        "labels",
+        "rules",
+        "applied_to",
+        "created"
+    )
+
+    def __init__(
+            self,
+            id=None,
+            name=None,
+            labels=None,
+            rules=None,
+            applied_to=None,
+            created=None
+    ):
+        self.id = id
+        self.name = name
+        self.rules = rules
+        self.applied_to = applied_to
+        self.labels = labels
+        self.created = isoparse(created) if created else None
+
+
+class FirewallRule:
+    """Firewall Rule Domain
+
+    :param direction: str
+           The Firewall which was created
+    :param port: str
+           Port to which traffic will be allowed, only applicable for protocols TCP and UDP, specify port ranges by using
+           - as a indicator, Sample: 80-85 means all ports between 80 & 85 (80, 82, 83, 84, 85)
+    :param protocol: str
+           Select traffic direction on which rule should be applied. Use source_ips for direction in and destination_ips for direction out.
+    :param source_ips: List[str]
+            List of permitted IPv4/IPv6 addresses in CIDR notation. Use 0.0.0.0/0 to allow all IPv4 addresses and ::/0 to allow all IPv6 addresses. You can specify 100 CIDRs at most.
+    :param destination_ips: List[str]
+           List of permitted IPv4/IPv6 addresses in CIDR notation. Use 0.0.0.0/0 to allow all IPv4 addresses and ::/0 to allow all IPv6 addresses. You can specify 100 CIDRs at most.
+    """
+    __slots__ = (
+        "direction",
+        "port",
+        "protocol",
+        "source_ips",
+        "destination_ips"
+    )
+
+    DIRECTION_IN = "in"
+    """Firewall Rule Direction In"""
+    DIRECTION_OUT = "out"
+    """Firewall Rule Direction Out"""
+
+    PROTOCOL_UDP = "udp"
+    """Firewall Rule Protocol UDP"""
+    PROTOCOL_ICMP = "icmp"
+    """Firewall Rule Protocol ICMP"""
+    PROTOCOL_TCP = "tcp"
+    """Firewall Rule Protocol TCP"""
+
+    def __init__(
+            self,
+            direction,  # type: str
+            protocol,  # type: str
+            source_ips,  # type: List[str]
+            port=None,  # type: Optional[str]
+            destination_ips=[],  # type: Optional[List[str]]
+    ):
+        self.direction = direction
+        self.port = port
+        self.protocol = protocol
+        self.source_ips = source_ips
+        self.destination_ips = destination_ips
+
+    def to_payload(self):
+        payload = {
+            "direction": self.direction,
+            "protocol": self.protocol,
+            "source_ips": self.source_ips,
+        }
+        if len(self.destination_ips) > 0:
+            payload.update({"destination_ips": self.destination_ips})
+        if self.port is not None:
+            payload.update({"port": self.port})
+        return payload
+
+
+class FirewallResource:
+    """Firewall Used By Domain
+
+    :param type: str
+           Type of resource referenced
+    :param server: Optional[Server]
+           Server the Firewall is applied to
+    """
+    __slots__ = (
+        "type",
+        "server",
+    )
+
+    TYPE_SERVER = "server"
+    """Firewall Used By Type Server"""
+
+    def __init__(
+            self,
+            type,  # type: str
+            server=None,  # type: Optional[Server]
+    ):
+        self.type = type
+        self.server = server
+
+    def to_payload(self):
+        payload = {
+            "type": self.type,
+        }
+        if self.server is not None:
+            payload.update({"server": {"id": self.server.id}})
+        return payload
+
+
+class CreateFirewallResponse(BaseDomain):
+    """Create Firewall Response Domain
+
+    :param firewall: :class:`BoundFirewall <hcloud.firewalls.client.BoundFirewall>`
+           The Firewall which was created
+    :param action: :class:`BoundAction <hcloud.actions.client.BoundAction>`
+           The Action which shows the progress of the Firewall Creation
+    """
+    __slots__ = (
+        "firewall",
+        "action"
+    )
+
+    def __init__(
+            self,
+            firewall,  # type: BoundFirewall
+            action,  # type: BoundAction
+    ):
+        self.firewall = firewall
+        self.action = action

hcloud/firewalls/domain.py

@@ -20,6 +20,7 @@
 from hcloud.load_balancer_types.client import LoadBalancerTypesClient
 
 from .__version__ import VERSION
+from .firewalls.client import FirewallsClient
 
 
 class APIException(Exception):
@@ -133,6 +134,12 @@ def __init__(self, token, api_endpoint="https://api.hetzner.cloud/v1", applicati
         :type: :class:`LoadBalancerTypesClient <hcloud.load_balancer_types.client.LoadBalancerTypesClient>`
         """
 
+        self.firewalls = FirewallsClient(self)
+        """FirewallsClient Instance
+
+        :type: :class:`FirewallsClient <hcloud.firewalls.client.FirewallsClient>`
+        """
+
     def _get_user_agent(self):
         """Get the user agent of the hcloud-python instance with the user application name (if specified)
 

hcloud/hcloud.py

@@ -25,7 +25,7 @@ class LoadBalancer(BaseDomain):
             Private networks information.
     :param algorithm: LoadBalancerAlgorithm
             The algorithm the Load Balancer is currently using
-    :param services: LoadBalancerService
+    :param services: List[LoadBalancerService]
             The services the LoadBalancer is currently serving
     :param targets: LoadBalancerTarget
             The targets the LoadBalancer is currently serving

hcloud/load_balancers/domain.py

@@ -3,10 +3,11 @@
 
 from hcloud.actions.client import BoundAction
 from hcloud.core.domain import add_meta_to_result
+from hcloud.firewalls.client import BoundFirewall
 from hcloud.floating_ips.client import BoundFloatingIP
 from hcloud.isos.client import BoundIso
 from hcloud.servers.domain import Server, CreateServerResponse, ResetPasswordResponse, EnableRescueResponse, \
-    RequestConsoleResponse, PublicNetwork, IPv4Address, IPv6Network, PrivateNet
+    RequestConsoleResponse, PublicNetwork, IPv4Address, IPv6Network, PrivateNet, PublicNetworkFirewall
 from hcloud.volumes.client import BoundVolume
 from hcloud.images.domain import CreateImageResponse
 from hcloud.images.client import BoundImage
@@ -48,11 +49,21 @@ def __init__(self, client, data, complete=True):
             ipv6_network = IPv6Network(**public_net['ipv6'])
             floating_ips = [BoundFloatingIP(client._client.floating_ips, {"id": floating_ip}, complete=False) for
                             floating_ip in public_net['floating_ips']]
-            data['public_net'] = PublicNetwork(ipv4=ipv4_address, ipv6=ipv6_network, floating_ips=floating_ips)
+            firewalls = [
+                PublicNetworkFirewall(
+                    BoundFirewall(client._client.firewalls, {"id": firewall["id"]}, complete=False),
+                    status=firewall["status"]
+                ) for firewall in public_net.get("firewalls", [])
+            ]
+            data['public_net'] = PublicNetwork(ipv4=ipv4_address, ipv6=ipv6_network, floating_ips=floating_ips,
+                                               firewalls=firewalls)
 
         private_nets = data.get("private_net")
         if private_nets:
-            private_nets = [PrivateNet(network=BoundNetwork(client._client.networks, {"id": private_net['network']}, complete=False), ip=private_net['ip'], alias_ips=private_net['alias_ips'], mac_address=private_net['mac_address']) for private_net in private_nets]
+            private_nets = [PrivateNet(
+                network=BoundNetwork(client._client.networks, {"id": private_net['network']}, complete=False),
+                ip=private_net['ip'], alias_ips=private_net['alias_ips'], mac_address=private_net['mac_address']) for
+                private_net in private_nets]
             data['private_net'] = private_nets
 
         super(BoundServer, self).__init__(client, data, complete)
@@ -392,6 +403,7 @@ def create(self,
                image,  # type: Image
                ssh_keys=None,  # type: Optional[List[SSHKey]]
                volumes=None,  # type: Optional[List[Volume]]
+               firewalls=None,  # type: Optional[List[Firewall]]
                networks=None,  # type: Optional[List[Network]]
                user_data=None,  # type: Optional[str]
                labels=None,  # type: Optional[Dict[str, str]]
@@ -444,6 +456,8 @@ def create(self,
             data['volumes'] = [volume.id for volume in volumes]
         if networks is not None:
             data['networks'] = [network.id for network in networks]
+        if firewalls is not None:
+            data['firewalls'] = [firewall.id for firewall in firewalls]
         if user_data is not None:
             data['user_data'] = user_data
         if labels is not None:

hcloud/servers/client.py

@@ -239,21 +239,49 @@ class PublicNetwork(BaseDomain):
     :param ipv4: :class:`IPv4Address <hcloud.servers.domain.IPv4Address>`
     :param ipv6: :class:`IPv6Network <hcloud.servers.domain.IPv6Network>`
     :param floating_ips: List[:class:`BoundFloatingIP <hcloud.floating_ips.client.BoundFloatingIP>`]
+    :param firewalls: List[:class:`PublicNetworkFirewall <hcloud.servers.client.PublicNetworkFirewall>`]
     """
     __slots__ = (
         "ipv4",
         "ipv6",
-        "floating_ips"
+        "floating_ips",
+        "firewalls"
     )
 
     def __init__(self,
                  ipv4,  # type: IPv4Address
                  ipv6,  # type: IPv6Network
                  floating_ips,  # type: List[BoundFloatingIP]
+                 firewalls=None,  # type: List[PublicNetworkFirewall]
                  ):
         self.ipv4 = ipv4
         self.ipv6 = ipv6
         self.floating_ips = floating_ips
+        self.firewalls = firewalls
+
+
+class PublicNetworkFirewall(BaseDomain):
+    """Public Network Domain
+
+    :param firewall: :class:`BoundFirewall <hcloud.firewalls.domain.BoundFirewall>`
+    :param status: str
+    """
+    __slots__ = (
+        "firewall",
+        "status"
+    )
+
+    STATUS_APPLIED = "applied"
+    """Public Network Firewall Status applied"""
+    STATUS_PENDING = "pending"
+    """Public Network Firewall Status pending"""
+
+    def __init__(self,
+                 firewall,  # type: BoundFirewall
+                 status,  # type: str
+                 ):
+        self.firewall = firewall
+        self.status = status
 
 
 class IPv4Address(BaseDomain):