ansible/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2024-09-14 20:13:21 +02:00
Code Releases Activity

Remove hetzner content. (#1368)

Browse source
This commit is contained in:
Felix Fontein 2020-11-24 06:34:23 +01:00 committed by GitHub
parent 8e3931d9b0
commit eab9a43d2e
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
18 changed files with 27 additions and 3192 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
.github/BOTMETA.yml vendored
View file

@ -47,8 +47,6 @@ files:
maintainers: $team_google
labels: gcp
supershipit: erjohnso rambleraptor
$doc_fragments/hetzner.py:
labels: hetzner
$doc_fragments/hpe3par.py:
maintainers: farhan7500 gautamphegde
labels: hpe3par
@ -465,14 +463,6 @@ files:
maintainers: briceburg
$modules/net_tools/haproxy.py:
maintainers: ravibhure
$modules/net_tools/hetzner_failover_ip.py:
maintainers: felixfontein
$modules/net_tools/hetzner_failover_ip_info.py:
maintainers: felixfontein
$modules/net_tools/hetzner_firewall.py:
maintainers: felixfontein
$modules/net_tools/hetzner_firewall_info.py:
maintainers: felixfontein
$modules/net_tools/:
maintainers: nerzhul
$modules/net_tools/infinity/infinity.py:

14
changelogs/fragments/hetzner-migration-removal.yml Normal file
View file

@ -0,0 +1,14 @@
removed_features:
- >
All ``hetzner`` modules have been removed from this collection.
They have been migrated to the `community.hrobot <https://galaxy.ansible.com/community/hrobot>`_ collection.
If you use ansible-base 2.10 or newer, redirections have been provided.
If you use Ansible 2.9 and installed this collection, you need to adjust the FQCNs (``community.general.hetzner_firewall`` → ``community.hrobot.firewall``) and make sure to install the community.hrobot collection.
breaking_changes:
- >
If you use Ansible 2.9 and the ``hetzner`` modules from this collections, community.general 2.0.0 results in errors when trying to use the hetzner content by FQCN, like ``community.general.hetzner_firewall``.
Since Ansible 2.9 is not able to use redirections, you will have to adjust your playbooks and roles manually to use the new FQCNs (``community.hrobot.firewall`` for the previous example) and to make sure that you have ``community.hrobot`` installed.
If you use ansible-base 2.10 or newer and did not install Ansible 3.0.0, but installed (and/or upgraded) community.general manually, you need to make sure to also install ``community.hrobot`` if you are using any of the ``hetzner`` modules.
While ansible-base 2.10 or newer can use the redirects that community.general 2.0.0 adds, the collection they point to (community.hrobot) must be installed for them to work.

13
meta/runtime.yml
View file

@ -1,3 +1,4 @@
---
requires_ansible: '>=2.9.10'
action_groups:
k8s:
@ -147,6 +148,14 @@ plugin_routing:
deprecation:
removal_version: 3.0.0
warning_text: The helm module in community.general has been deprecated. Use community.kubernetes.helm instead.
hetzner_failover_ip:
redirect: community.hrobot.failover_ip
hetzner_failover_ip_info:
redirect: community.hrobot.failover_ip_info
hetzner_firewall:
redirect: community.hrobot.firewall
hetzner_firewall_info:
redirect: community.hrobot.firewall_info
hpilo_facts:
deprecation:
removal_version: 3.0.0
@ -450,11 +459,15 @@ plugin_routing:
doc_fragments:
docker:
redirect: community.docker.docker
hetzner:
redirect: community.hrobot.robot
module_utils:
docker.common:
redirect: community.docker.common
docker.swarm:
redirect: community.docker.swarm
hetzner:
redirect: community.hrobot.robot
callback:
actionable:
tombstone:

23
plugins/doc_fragments/hetzner.py
View file

@ -1,23 +0,0 @@
# -*- coding: utf-8 -*-
# Copyright: (c) 2019 Felix Fontein <felix@fontein.de>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
class ModuleDocFragment(object):
# Standard files documentation fragment
DOCUMENTATION = r'''
options:
hetzner_user:
description: The username for the Robot webservice user.
type: str
required: yes
hetzner_password:
description: The password for the Robot webservice user.
type: str
required: yes
'''

171
plugins/module_utils/hetzner.py
View file

@ -1,171 +0,0 @@
# -*- coding: utf-8 -*-
# This code is part of Ansible, but is an independent component.
# This particular file snippet, and this file snippet only, is BSD licensed.
# Modules you write using this snippet, which is embedded dynamically by Ansible
# still belong to the author of the module, and may assign their own license
# to the complete work.
#
# Copyright (c), Felix Fontein <felix@fontein.de>, 2019
#
# Simplified BSD License (see licenses/simplified_bsd.txt or https://opensource.org/licenses/BSD-2-Clause)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
from ansible.module_utils.urls import fetch_url
from ansible.module_utils.six.moves.urllib.parse import urlencode
import time
HETZNER_DEFAULT_ARGUMENT_SPEC = dict(
hetzner_user=dict(type='str', required=True),
hetzner_password=dict(type='str', required=True, no_log=True),
)
# The API endpoint is fixed.
BASE_URL = "https://robot-ws.your-server.de"
def fetch_url_json(module, url, method='GET', timeout=10, data=None, headers=None, accept_errors=None):
'''
Make general request to Hetzner's JSON robot API.
'''
module.params['url_username'] = module.params['hetzner_user']
module.params['url_password'] = module.params['hetzner_password']
resp, info = fetch_url(module, url, method=method, timeout=timeout, data=data, headers=headers)
try:
content = resp.read()
except AttributeError:
content = info.pop('body', None)
if not content:
module.fail_json(msg='Cannot retrieve content from {0}'.format(url))
try:
result = module.from_json(content.decode('utf8'))
if 'error' in result:
if accept_errors:
if result['error']['code'] in accept_errors:
return result, result['error']['code']
module.fail_json(msg='Request failed: {0} {1} ({2})'.format(
result['error']['status'],
result['error']['code'],
result['error']['message']
))
return result, None
except ValueError:
module.fail_json(msg='Cannot decode content retrieved from {0}'.format(url))
class CheckDoneTimeoutException(Exception):
def __init__(self, result, error):
super(CheckDoneTimeoutException, self).__init__()
self.result = result
self.error = error
def fetch_url_json_with_retries(module, url, check_done_callback, check_done_delay=10, check_done_timeout=180, skip_first=False, **kwargs):
'''
Make general request to Hetzner's JSON robot API, with retries until a condition is satisfied.
The condition is tested by calling ``check_done_callback(result, error)``. If it is not satisfied,
it will be retried with delays ``check_done_delay`` (in seconds) until a total timeout of
``check_done_timeout`` (in seconds) since the time the first request is started is reached.
If ``skip_first`` is specified, will assume that a first call has already been made and will
directly start with waiting.
'''
start_time = time.time()
if not skip_first:
result, error = fetch_url_json(module, url, **kwargs)
if check_done_callback(result, error):
return result, error
while True:
elapsed = (time.time() - start_time)
left_time = check_done_timeout - elapsed
time.sleep(max(min(check_done_delay, left_time), 0))
result, error = fetch_url_json(module, url, **kwargs)
if check_done_callback(result, error):
return result, error
if left_time < check_done_delay:
raise CheckDoneTimeoutException(result, error)
# #####################################################################################
# ## FAILOVER IP ######################################################################
def get_failover_record(module, ip):
'''
Get information record of failover IP.
See https://robot.your-server.de/doc/webservice/en.html#get-failover-failover-ip
'''
url = "{0}/failover/{1}".format(BASE_URL, ip)
result, error = fetch_url_json(module, url)
if 'failover' not in result:
module.fail_json(msg='Cannot interpret result: {0}'.format(result))
return result['failover']
def get_failover(module, ip):
'''
Get current routing target of failover IP.
The value ``None`` represents unrouted.
See https://robot.your-server.de/doc/webservice/en.html#get-failover-failover-ip
'''
return get_failover_record(module, ip)['active_server_ip']
def set_failover(module, ip, value, timeout=180):
'''
Set current routing target of failover IP.
Return a pair ``(value, changed)``. The value ``None`` for ``value`` represents unrouted.
See https://robot.your-server.de/doc/webservice/en.html#post-failover-failover-ip
and https://robot.your-server.de/doc/webservice/en.html#delete-failover-failover-ip
'''
url = "{0}/failover/{1}".format(BASE_URL, ip)
if value is None:
result, error = fetch_url_json(
module,
url,
method='DELETE',
timeout=timeout,
accept_errors=['FAILOVER_ALREADY_ROUTED']
)
else:
headers = {"Content-type": "application/x-www-form-urlencoded"}
data = dict(
active_server_ip=value,
)
result, error = fetch_url_json(
module,
url,
method='POST',
timeout=timeout,
data=urlencode(data),
headers=headers,
accept_errors=['FAILOVER_ALREADY_ROUTED']
)
if error is not None:
return value, False
else:
return result['failover']['active_server_ip'], True
def get_failover_state(value):
'''
Create result dictionary for failover IP's value.
The value ``None`` represents unrouted.
'''
return dict(
value=value,
state='routed' if value else 'unrouted'
)

1
plugins/modules/hetzner_failover_ip.py
View file

@ -1 +0,0 @@
./net_tools/hetzner_failover_ip.py

1
plugins/modules/hetzner_failover_ip_info.py
View file

@ -1 +0,0 @@
./net_tools/hetzner_failover_ip_info.py

1
plugins/modules/hetzner_firewall.py
View file

@ -1 +0,0 @@
./net_tools/hetzner_firewall.py

1
plugins/modules/hetzner_firewall_info.py
View file

@ -1 +0,0 @@
./net_tools/hetzner_firewall_info.py

141
plugins/modules/net_tools/hetzner_failover_ip.py
View file

@ -1,141 +0,0 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# (c) 2019 Felix Fontein <felix@fontein.de>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
DOCUMENTATION = r'''
---
module: hetzner_failover_ip
short_description: Manage Hetzner's failover IPs
author:
- Felix Fontein (@felixfontein)
description:
- Manage Hetzner's failover IPs.
seealso:
- name: Failover IP documentation
description: Hetzner's documentation on failover IPs.
link: https://wiki.hetzner.de/index.php/Failover/en
- module: community.general.hetzner_failover_ip_info
description: Retrieve information on failover IPs.
extends_documentation_fragment:
- community.general.hetzner
options:
failover_ip:
description: The failover IP address.
type: str
required: yes
state:
description:
- Defines whether the IP will be routed or not.
- If set to C(routed), I(value) must be specified.
type: str
choices:
- routed
- unrouted
default: routed
value:
description:
- The new value for the failover IP address.
- Required when setting I(state) to C(routed).
type: str
timeout:
description:
- Timeout to use when routing or unrouting the failover IP.
- Note that the API call returns when the failover IP has been
successfully routed to the new address, respectively successfully
unrouted.
type: int
default: 180
'''
EXAMPLES = r'''
- name: Set value of failover IP 1.2.3.4 to 5.6.7.8
community.general.hetzner_failover_ip:
hetzner_user: foo
hetzner_password: bar
failover_ip: 1.2.3.4
value: 5.6.7.8
- name: Set value of failover IP 1.2.3.4 to unrouted
community.general.hetzner_failover_ip:
hetzner_user: foo
hetzner_password: bar
failover_ip: 1.2.3.4
state: unrouted
'''
RETURN = r'''
value:
description:
- The value of the failover IP.
- Will be C(none) if the IP is unrouted.
returned: success
type: str
state:
description:
- Will be C(routed) or C(unrouted).
returned: success
type: str
'''
from ansible.module_utils.basic import AnsibleModule
from ansible_collections.community.general.plugins.module_utils.hetzner import (
HETZNER_DEFAULT_ARGUMENT_SPEC,
get_failover,
set_failover,
get_failover_state,
)
def main():
argument_spec = dict(
failover_ip=dict(type='str', required=True),
state=dict(type='str', default='routed', choices=['routed', 'unrouted']),
value=dict(type='str'),
timeout=dict(type='int', default=180),
)
argument_spec.update(HETZNER_DEFAULT_ARGUMENT_SPEC)
module = AnsibleModule(
argument_spec=argument_spec,
supports_check_mode=True,
required_if=(
('state', 'routed', ['value']),
),
)
failover_ip = module.params['failover_ip']
value = get_failover(module, failover_ip)
changed = False
before = get_failover_state(value)
if module.params['state'] == 'routed':
new_value = module.params['value']
else:
new_value = None
if value != new_value:
if module.check_mode:
value = new_value
changed = True
else:
value, changed = set_failover(module, failover_ip, new_value, timeout=module.params['timeout'])
after = get_failover_state(value)
module.exit_json(
changed=changed,
diff=dict(
before=before,
after=after,
),
**after
)
if __name__ == '__main__':
main()

117
plugins/modules/net_tools/hetzner_failover_ip_info.py
View file

@ -1,117 +0,0 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# (c) 2019 Felix Fontein <felix@fontein.de>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
DOCUMENTATION = r'''
---
module: hetzner_failover_ip_info
short_description: Retrieve information on Hetzner's failover IPs
author:
- Felix Fontein (@felixfontein)
description:
- Retrieve information on Hetzner's failover IPs.
seealso:
- name: Failover IP documentation
description: Hetzner's documentation on failover IPs.
link: https://wiki.hetzner.de/index.php/Failover/en
- module: community.general.hetzner_failover_ip
description: Manage failover IPs.
extends_documentation_fragment:
- community.general.hetzner
options:
failover_ip:
description: The failover IP address.
type: str
required: yes
'''
EXAMPLES = r'''
- name: Get value of failover IP 1.2.3.4
community.general.hetzner_failover_ip_info:
hetzner_user: foo
hetzner_password: bar
failover_ip: 1.2.3.4
value: 5.6.7.8
register: result
- name: Print value of failover IP 1.2.3.4 in case it is routed
ansible.builtin.debug:
msg: "1.2.3.4 routes to {{ result.value }}"
when: result.state == 'routed'
'''
RETURN = r'''
value:
description:
- The value of the failover IP.
- Will be C(none) if the IP is unrouted.
returned: success
type: str
state:
description:
- Will be C(routed) or C(unrouted).
returned: success
type: str
failover_ip:
description:
- The failover IP.
returned: success
type: str
sample: '1.2.3.4'
failover_netmask:
description:
- The netmask for the failover IP.
returned: success
type: str
sample: '255.255.255.255'
server_ip:
description:
- The main IP of the server this failover IP is associated to.
- This is I(not) the server the failover IP is routed to.
returned: success
type: str
server_number:
description:
- The number of the server this failover IP is associated to.
- This is I(not) the server the failover IP is routed to.
returned: success
type: int
'''
from ansible.module_utils.basic import AnsibleModule
from ansible_collections.community.general.plugins.module_utils.hetzner import (
HETZNER_DEFAULT_ARGUMENT_SPEC,
get_failover_record,
get_failover_state,
)
def main():
argument_spec = dict(
failover_ip=dict(type='str', required=True),
)
argument_spec.update(HETZNER_DEFAULT_ARGUMENT_SPEC)
module = AnsibleModule(
argument_spec=argument_spec,
supports_check_mode=True,
)
failover = get_failover_record(module, module.params['failover_ip'])
result = get_failover_state(failover['active_server_ip'])
result['failover_ip'] = failover['ip']
result['failover_netmask'] = failover['netmask']
result['server_ip'] = failover['server_ip']
result['server_number'] = failover['server_number']
result['changed'] = False
module.exit_json(**result)
if __name__ == '__main__':
main()

509
plugins/modules/net_tools/hetzner_firewall.py
View file

@ -1,509 +0,0 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# (c) 2019 Felix Fontein <felix@fontein.de>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
DOCUMENTATION = r'''
---
module: hetzner_firewall
version_added: '0.2.0'
short_description: Manage Hetzner's dedicated server firewall
author:
- Felix Fontein (@felixfontein)
description:
- Manage Hetzner's dedicated server firewall.
- Note that idempotency check for TCP flags simply compares strings and doesn't
try to interpret the rules. This might change in the future.
seealso:
- name: Firewall documentation
description: Hetzner's documentation on the stateless firewall for dedicated servers
link: https://wiki.hetzner.de/index.php/Robot_Firewall/en
- module: community.general.hetzner_firewall_info
description: Retrieve information on firewall configuration.
extends_documentation_fragment:
- community.general.hetzner
options:
server_ip:
description: The server's main IP address.
required: yes
type: str
port:
description:
- Switch port of firewall.
type: str
choices: [ main, kvm ]
default: main
state:
description:
- Status of the firewall.
- Firewall is active if state is C(present), and disabled if state is C(absent).
type: str
default: present
choices: [ present, absent ]
whitelist_hos:
description:
- Whether Hetzner services have access.
type: bool
rules:
description:
- Firewall rules.
type: dict
suboptions:
input:
description:
- Input firewall rules.
type: list
elements: dict
suboptions:
name:
description:
- Name of the firewall rule.
type: str
ip_version:
description:
- Internet protocol version.
- Note that currently, only IPv4 is supported by Hetzner.
required: yes
type: str
choices: [ ipv4, ipv6 ]
dst_ip:
description:
- Destination IP address or subnet address.
- CIDR notation.
type: str
dst_port:
description:
- Destination port or port range.
type: str
src_ip:
description:
- Source IP address or subnet address.
- CIDR notation.
type: str
src_port:
description:
- Source port or port range.
type: str
protocol:
description:
- Protocol above IP layer
type: str
tcp_flags:
description:
- TCP flags or logical combination of flags.
- Flags supported by Hetzner are C(syn), C(fin), C(rst), C(psh) and C(urg).
- They can be combined with C(|) (logical or) and C(&) (logical and).
- See L(the documentation,https://wiki.hetzner.de/index.php/Robot_Firewall/en#Parameter)
for more information.
type: str
action:
description:
- Action if rule matches.
required: yes
type: str
choices: [ accept, discard ]
update_timeout:
description:
- Timeout to use when configuring the firewall.
- Note that the API call returns before the firewall has been
successfully set up.
type: int
default: 30
wait_for_configured:
description:
- Whether to wait until the firewall has been successfully configured before
determining what to do, and before returning from the module.
- The API returns status C(in progress) when the firewall is currently
being configured. If this happens, the module will try again until
the status changes to C(active) or C(disabled).
- Please note that there is a request limit. If you have to do multiple
updates, it can be better to disable waiting, and regularly use
M(community.general.hetzner_firewall_info) to query status.
type: bool
default: yes
wait_delay:
description:
- Delay to wait (in seconds) before checking again whether the firewall has
been configured.
type: int
default: 10
timeout:
description:
- Timeout (in seconds) for waiting for firewall to be configured.
type: int
default: 180
'''
EXAMPLES = r'''
- name: Configure firewall for server with main IP 1.2.3.4
community.general.hetzner_firewall:
hetzner_user: foo
hetzner_password: bar
server_ip: 1.2.3.4
state: present
whitelist_hos: yes
rules:
input:
- name: Allow everything to ports 20-23 from 4.3.2.1/24
ip_version: ipv4
src_ip: 4.3.2.1/24
dst_port: '20-23'
action: accept
- name: Allow everything to port 443
ip_version: ipv4
dst_port: '443'
action: accept
- name: Drop everything else
ip_version: ipv4
action: discard
register: result
- ansible.builtin.debug:
msg: "{{ result }}"
'''
RETURN = r'''
firewall:
description:
- The firewall configuration.
type: dict
returned: success
contains:
port:
description:
- Switch port of firewall.
- C(main) or C(kvm).
type: str
sample: main
server_ip:
description:
- Server's main IP address.
type: str
sample: 1.2.3.4
server_number:
description:
- Hetzner's internal server number.
type: int
sample: 12345
status:
description:
- Status of the firewall.
- C(active) or C(disabled).
- Will be C(in process) if the firewall is currently updated, and
I(wait_for_configured) is set to C(no) or I(timeout) to a too small value.
type: str
sample: active
whitelist_hos:
description:
- Whether Hetzner services have access.
type: bool
sample: true
rules:
description:
- Firewall rules.
type: dict
contains:
input:
description:
- Input firewall rules.
type: list
elements: dict
contains:
name:
description:
- Name of the firewall rule.
type: str
sample: Allow HTTP access to server
ip_version:
description:
- Internet protocol version.
type: str
sample: ipv4
dst_ip:
description:
- Destination IP address or subnet address.
- CIDR notation.
type: str
sample: 1.2.3.4/32
dst_port:
description:
- Destination port or port range.
type: str
sample: "443"
src_ip:
description:
- Source IP address or subnet address.
- CIDR notation.
type: str
sample: null
src_port:
description:
- Source port or port range.
type: str
sample: null
protocol:
description:
- Protocol above IP layer
type: str
sample: tcp
tcp_flags:
description:
- TCP flags or logical combination of flags.
type: str
sample: null
action:
description:
- Action if rule matches.
- C(accept) or C(discard).
type: str
sample: accept
'''
from ansible.module_utils.basic import AnsibleModule
from ansible_collections.community.general.plugins.module_utils.compat import ipaddress as compat_ipaddress
from ansible_collections.community.general.plugins.module_utils.hetzner import (
HETZNER_DEFAULT_ARGUMENT_SPEC,
BASE_URL,
fetch_url_json,
fetch_url_json_with_retries,
CheckDoneTimeoutException,
)
from ansible.module_utils.six.moves.urllib.parse import urlencode
from ansible.module_utils._text import to_native, to_text
RULE_OPTION_NAMES = [
'name', 'ip_version', 'dst_ip', 'dst_port', 'src_ip', 'src_port',
'protocol', 'tcp_flags', 'action',
]
RULES = ['input']
def restrict_dict(dictionary, fields):
result = dict()
for k, v in dictionary.items():
if k in fields:
result[k] = v
return result
def restrict_firewall_config(config):
result = restrict_dict(config, ['port', 'status', 'whitelist_hos'])
result['rules'] = dict()
for ruleset in RULES:
result['rules'][ruleset] = [
restrict_dict(rule, RULE_OPTION_NAMES)
for rule in config['rules'].get(ruleset) or []
]
return result
def update(before, after, params, name):
bv = before.get(name)
after[name] = bv
changed = False
pv = params[name]
if pv is not None:
changed = pv != bv
if changed:
after[name] = pv
return changed
def normalize_ip(ip, ip_version):
if ip is None:
return ip
if '/' in ip:
ip, range = ip.split('/')
else:
ip, range = ip, ''
ip_addr = to_native(compat_ipaddress.ip_address(to_text(ip)).compressed)
if range == '':
range = '32' if ip_version.lower() == 'ipv4' else '128'
return ip_addr + '/' + range
def update_rules(before, after, params, ruleset):
before_rules = before['rules'][ruleset]
after_rules = after['rules'][ruleset]
params_rules = params['rules'][ruleset]
changed = len(before_rules) != len(params_rules)
for no, rule in enumerate(params_rules):
rule['src_ip'] = normalize_ip(rule['src_ip'], rule['ip_version'])
rule['dst_ip'] = normalize_ip(rule['dst_ip'], rule['ip_version'])
if no < len(before_rules):
before_rule = before_rules[no]
before_rule['src_ip'] = normalize_ip(before_rule['src_ip'], before_rule['ip_version'])
before_rule['dst_ip'] = normalize_ip(before_rule['dst_ip'], before_rule['ip_version'])
if before_rule != rule:
changed = True
after_rules.append(rule)
return changed
def encode_rule(output, rulename, input):
for i, rule in enumerate(input['rules'][rulename]):
for k, v in rule.items():
if v is not None:
output['rules[{0}][{1}][{2}]'.format(rulename, i, k)] = v
def create_default_rules_object():
rules = dict()
for ruleset in RULES:
rules[ruleset] = []
return rules
def firewall_configured(result, error):
return result['firewall']['status'] != 'in process'
def main():
argument_spec = dict(
server_ip=dict(type='str', required=True),
port=dict(type='str', default='main', choices=['main', 'kvm']),
state=dict(type='str', default='present', choices=['present', 'absent']),
whitelist_hos=dict(type='bool'),
rules=dict(type='dict', options=dict(
input=dict(type='list', elements='dict', options=dict(
name=dict(type='str'),
ip_version=dict(type='str', required=True, choices=['ipv4', 'ipv6']),
dst_ip=dict(type='str'),
dst_port=dict(type='str'),
src_ip=dict(type='str'),
src_port=dict(type='str'),
protocol=dict(type='str'),
tcp_flags=dict(type='str'),
action=dict(type='str', required=True, choices=['accept', 'discard']),
)),
)),
update_timeout=dict(type='int', default=30),
wait_for_configured=dict(type='bool', default=True),
wait_delay=dict(type='int', default=10),
timeout=dict(type='int', default=180),
)
argument_spec.update(HETZNER_DEFAULT_ARGUMENT_SPEC)
module = AnsibleModule(
argument_spec=argument_spec,
supports_check_mode=True,
)
# Sanitize input
module.params['status'] = 'active' if (module.params['state'] == 'present') else 'disabled'
if module.params['rules'] is None:
module.params['rules'] = {}
if module.params['rules'].get('input') is None:
module.params['rules']['input'] = []
server_ip = module.params['server_ip']
# https://robot.your-server.de/doc/webservice/en.html#get-firewall-server-ip
url = "{0}/firewall/{1}".format(BASE_URL, server_ip)
if module.params['wait_for_configured']:
try:
result, error = fetch_url_json_with_retries(
module,
url,
check_done_callback=firewall_configured,
check_done_delay=module.params['wait_delay'],
check_done_timeout=module.params['timeout'],
)
except CheckDoneTimeoutException as dummy:
module.fail_json(msg='Timeout while waiting for firewall to be configured.')
else:
result, error = fetch_url_json(module, url)
if not firewall_configured(result, error):
module.fail_json(msg='Firewall configuration cannot be read as it is not configured.')
full_before = result['firewall']
if not full_before.get('rules'):
full_before['rules'] = create_default_rules_object()
before = restrict_firewall_config(full_before)
# Build wanted (after) state and compare
after = dict(before)
changed = False
changed |= update(before, after, module.params, 'port')
changed |= update(before, after, module.params, 'status')
changed |= update(before, after, module.params, 'whitelist_hos')
after['rules'] = create_default_rules_object()
if module.params['status'] == 'active':
for ruleset in RULES:
changed |= update_rules(before, after, module.params, ruleset)
# Update if different
construct_result = True
construct_status = None
if changed and not module.check_mode:
# https://robot.your-server.de/doc/webservice/en.html#post-firewall-server-ip
url = "{0}/firewall/{1}".format(BASE_URL, server_ip)
headers = {"Content-type": "application/x-www-form-urlencoded"}
data = dict(after)
data['whitelist_hos'] = str(data['whitelist_hos']).lower()
del data['rules']
for ruleset in RULES:
encode_rule(data, ruleset, after)
result, error = fetch_url_json(
module,
url,
method='POST',
timeout=module.params['update_timeout'],
data=urlencode(data),
headers=headers,
)
if module.params['wait_for_configured'] and not firewall_configured(result, error):
try:
result, error = fetch_url_json_with_retries(
module,
url,
check_done_callback=firewall_configured,
check_done_delay=module.params['wait_delay'],
check_done_timeout=module.params['timeout'],
skip_first=True,
)
except CheckDoneTimeoutException as e:
result, error = e.result, e.error
module.warn('Timeout while waiting for firewall to be configured.')
full_after = result['firewall']
if not full_after.get('rules'):
full_after['rules'] = create_default_rules_object()
construct_status = full_after['status']
if construct_status != 'in process':
# Only use result if configuration is done, so that diff will be ok
after = restrict_firewall_config(full_after)
construct_result = False
if construct_result:
# Construct result (used for check mode, and configuration still in process)
full_after = dict(full_before)
for k, v in after.items():
if k != 'rules':
full_after[k] = after[k]
if construct_status is not None:
# We want 'in process' here
full_after['status'] = construct_status
full_after['rules'] = dict()
for ruleset in RULES:
full_after['rules'][ruleset] = after['rules'][ruleset]
module.exit_json(
changed=changed,
diff=dict(
before=before,
after=after,
),
firewall=full_after,
)
if __name__ == '__main__':
main()

226
plugins/modules/net_tools/hetzner_firewall_info.py
View file

@ -1,226 +0,0 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# (c) 2019 Felix Fontein <felix@fontein.de>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
DOCUMENTATION = r'''
---
module: hetzner_firewall_info
version_added: '0.2.0'
short_description: Manage Hetzner's dedicated server firewall
author:
- Felix Fontein (@felixfontein)
description:
- Manage Hetzner's dedicated server firewall.
seealso:
- name: Firewall documentation
description: Hetzner's documentation on the stateless firewall for dedicated servers
link: https://wiki.hetzner.de/index.php/Robot_Firewall/en
- module: community.general.hetzner_firewall
description: Configure firewall.
extends_documentation_fragment:
- community.general.hetzner
options:
server_ip:
description: The server's main IP address.
type: str
required: yes
wait_for_configured:
description:
- Whether to wait until the firewall has been successfully configured before
determining what to do, and before returning from the module.
- The API returns status C(in progress) when the firewall is currently
being configured. If this happens, the module will try again until
the status changes to C(active) or C(disabled).
- Please note that there is a request limit. If you have to do multiple
updates, it can be better to disable waiting, and regularly use
M(community.general.hetzner_firewall_info) to query status.
type: bool
default: yes
wait_delay:
description:
- Delay to wait (in seconds) before checking again whether the firewall has
been configured.
type: int
default: 10
timeout:
description:
- Timeout (in seconds) for waiting for firewall to be configured.
type: int
default: 180
'''
EXAMPLES = r'''
- name: Get firewall configuration for server with main IP 1.2.3.4
community.general.hetzner_firewall_info:
hetzner_user: foo
hetzner_password: bar
server_ip: 1.2.3.4
register: result
- ansible.builtin.debug:
msg: "{{ result.firewall }}"
'''
RETURN = r'''
firewall:
description:
- The firewall configuration.
type: dict
returned: success
contains:
port:
description:
- Switch port of firewall.
- C(main) or C(kvm).
type: str
sample: main
server_ip:
description:
- Server's main IP address.
type: str
sample: 1.2.3.4
server_number:
description:
- Hetzner's internal server number.
type: int
sample: 12345
status:
description:
- Status of the firewall.
- C(active) or C(disabled).
- Will be C(in process) if the firewall is currently updated, and
I(wait_for_configured) is set to C(no) or I(timeout) to a too small value.
type: str
sample: active
whitelist_hos:
description:
- Whether Hetzner services have access.
type: bool
sample: true
rules:
description:
- Firewall rules.
type: dict
contains:
input:
description:
- Input firewall rules.
type: list
elements: dict
contains:
name:
description:
- Name of the firewall rule.
type: str
sample: Allow HTTP access to server
ip_version:
description:
- Internet protocol version.
type: str
sample: ipv4
dst_ip:
description:
- Destination IP address or subnet address.
- CIDR notation.
type: str
sample: 1.2.3.4/32
dst_port:
description:
- Destination port or port range.
type: str
sample: "443"
src_ip:
description:
- Source IP address or subnet address.
- CIDR notation.
type: str
sample: null
src_port:
description:
- Source port or port range.
type: str
sample: null
protocol:
description:
- Protocol above IP layer
type: str
sample: tcp
tcp_flags:
description:
- TCP flags or logical combination of flags.
type: str
sample: null
action:
description:
- Action if rule matches.
- C(accept) or C(discard).
type: str
sample: accept
'''
from ansible.module_utils.basic import AnsibleModule
from ansible_collections.community.general.plugins.module_utils.hetzner import (
HETZNER_DEFAULT_ARGUMENT_SPEC,
BASE_URL,
fetch_url_json,
fetch_url_json_with_retries,
CheckDoneTimeoutException,
)
def firewall_configured(result, error):
return result['firewall']['status'] != 'in process'
def main():
argument_spec = dict(
server_ip=dict(type='str', required=True),
wait_for_configured=dict(type='bool', default=True),
wait_delay=dict(type='int', default=10),
timeout=dict(type='int', default=180),
)
argument_spec.update(HETZNER_DEFAULT_ARGUMENT_SPEC)
module = AnsibleModule(
argument_spec=argument_spec,
supports_check_mode=True,
)
server_ip = module.params['server_ip']
# https://robot.your-server.de/doc/webservice/en.html#get-firewall-server-ip
url = "{0}/firewall/{1}".format(BASE_URL, server_ip)
if module.params['wait_for_configured']:
try:
result, error = fetch_url_json_with_retries(
module,
url,
check_done_callback=firewall_configured,
check_done_delay=module.params['wait_delay'],
check_done_timeout=module.params['timeout'],
)
except CheckDoneTimeoutException as dummy:
module.fail_json(msg='Timeout while waiting for firewall to be configured.')
else:
result, error = fetch_url_json(module, url)
firewall = result['firewall']
if not firewall.get('rules'):
firewall['rules'] = dict()
for ruleset in ['input']:
firewall['rules'][ruleset] = []
module.exit_json(
changed=False,
firewall=firewall,
)
if __name__ == '__main__':
main()

268
tests/unit/plugins/module_utils/test_hetzner.py
View file

@ -1,268 +0,0 @@
# Copyright: (c) 2017 Ansible Project
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import copy
import json
import pytest
from mock import MagicMock
from ansible_collections.community.general.plugins.module_utils import hetzner
class ModuleFailException(Exception):
def __init__(self, msg, **kwargs):
super(ModuleFailException, self).__init__(msg)
self.fail_msg = msg
self.fail_kwargs = kwargs
def get_module_mock():
def f(msg, **kwargs):
raise ModuleFailException(msg, **kwargs)
module = MagicMock()
module.fail_json = f
module.from_json = json.loads
return module
# ########################################################################################
FETCH_URL_JSON_SUCCESS = [
(
(None, dict(
body=json.dumps(dict(
a='b'
)).encode('utf-8'),
)),
None,
(dict(
a='b'
), None)
),
(
(None, dict(
body=json.dumps(dict(
error=dict(
code="foo",
status=400,
message="bar",
),
a='b'
)).encode('utf-8'),
)),
['foo'],
(dict(
error=dict(
code="foo",
status=400,
message="bar",
),
a='b'
), 'foo')
),
]
FETCH_URL_JSON_FAIL = [
(
(None, dict(
body=json.dumps(dict(
error=dict(
code="foo",
status=400,
message="bar",
),
)).encode('utf-8'),
)),
None,
'Request failed: 400 foo (bar)'
),
(
(None, dict(
body=json.dumps(dict(
error=dict(
code="foo",
status=400,
message="bar",
),
)).encode('utf-8'),
)),
['bar'],
'Request failed: 400 foo (bar)'
),
]
@pytest.mark.parametrize("return_value, accept_errors, result", FETCH_URL_JSON_SUCCESS)
def test_fetch_url_json(monkeypatch, return_value, accept_errors, result):
module = get_module_mock()
hetzner.fetch_url = MagicMock(return_value=return_value)
assert hetzner.fetch_url_json(module, 'https://foo/bar', accept_errors=accept_errors) == result
@pytest.mark.parametrize("return_value, accept_errors, result", FETCH_URL_JSON_FAIL)
def test_fetch_url_json_fail(monkeypatch, return_value, accept_errors, result):
module = get_module_mock()
hetzner.fetch_url = MagicMock(return_value=return_value)
with pytest.raises(ModuleFailException) as exc:
hetzner.fetch_url_json(module, 'https://foo/bar', accept_errors=accept_errors)
assert exc.value.fail_msg == result
assert exc.value.fail_kwargs == dict()
# ########################################################################################
GET_FAILOVER_SUCCESS = [
(
'1.2.3.4',
(None, dict(
body=json.dumps(dict(
failover=dict(
active_server_ip='1.1.1.1',
ip='1.2.3.4',
netmask='255.255.255.255',
)
)).encode('utf-8'),
)),
'1.1.1.1',
dict(
active_server_ip='1.1.1.1',
ip='1.2.3.4',
netmask='255.255.255.255',
)
),
]
GET_FAILOVER_FAIL = [
(
'1.2.3.4',
(None, dict(
body=json.dumps(dict(
error=dict(
code="foo",
status=400,
message="bar",
),
)).encode('utf-8'),
)),
'Request failed: 400 foo (bar)'
),
]
@pytest.mark.parametrize("ip, return_value, result, record", GET_FAILOVER_SUCCESS)
def test_get_failover_record(monkeypatch, ip, return_value, result, record):
module = get_module_mock()
hetzner.fetch_url = MagicMock(return_value=copy.deepcopy(return_value))
assert hetzner.get_failover_record(module, ip) == record
@pytest.mark.parametrize("ip, return_value, result", GET_FAILOVER_FAIL)
def test_get_failover_record_fail(monkeypatch, ip, return_value, result):
module = get_module_mock()
hetzner.fetch_url = MagicMock(return_value=copy.deepcopy(return_value))
with pytest.raises(ModuleFailException) as exc:
hetzner.get_failover_record(module, ip)
assert exc.value.fail_msg == result
assert exc.value.fail_kwargs == dict()
@pytest.mark.parametrize("ip, return_value, result, record", GET_FAILOVER_SUCCESS)
def test_get_failover(monkeypatch, ip, return_value, result, record):
module = get_module_mock()
hetzner.fetch_url = MagicMock(return_value=copy.deepcopy(return_value))
assert hetzner.get_failover(module, ip) == result
@pytest.mark.parametrize("ip, return_value, result", GET_FAILOVER_FAIL)
def test_get_failover_fail(monkeypatch, ip, return_value, result):
module = get_module_mock()
hetzner.fetch_url = MagicMock(return_value=copy.deepcopy(return_value))
with pytest.raises(ModuleFailException) as exc:
hetzner.get_failover(module, ip)
assert exc.value.fail_msg == result
assert exc.value.fail_kwargs == dict()
# ########################################################################################
SET_FAILOVER_SUCCESS = [
(
'1.2.3.4',
'1.1.1.1',
(None, dict(
body=json.dumps(dict(
failover=dict(
active_server_ip='1.1.1.2',
)
)).encode('utf-8'),
)),
('1.1.1.2', True)
),
(
'1.2.3.4',
'1.1.1.1',
(None, dict(
body=json.dumps(dict(
error=dict(
code="FAILOVER_ALREADY_ROUTED",
status=400,
message="Failover already routed",
),
)).encode('utf-8'),
)),
('1.1.1.1', False)
),
]
SET_FAILOVER_FAIL = [
(
'1.2.3.4',
'1.1.1.1',
(None, dict(
body=json.dumps(dict(
error=dict(
code="foo",
status=400,
message="bar",
),
)).encode('utf-8'),
)),
'Request failed: 400 foo (bar)'
),
]
@pytest.mark.parametrize("ip, value, return_value, result", SET_FAILOVER_SUCCESS)
def test_set_failover(monkeypatch, ip, value, return_value, result):
module = get_module_mock()
hetzner.fetch_url = MagicMock(return_value=copy.deepcopy(return_value))
assert hetzner.set_failover(module, ip, value) == result
@pytest.mark.parametrize("ip, value, return_value, result", SET_FAILOVER_FAIL)
def test_set_failover_fail(monkeypatch, ip, value, return_value, result):
module = get_module_mock()
hetzner.fetch_url = MagicMock(return_value=copy.deepcopy(return_value))
with pytest.raises(ModuleFailException) as exc:
hetzner.set_failover(module, ip, value)
assert exc.value.fail_msg == result
assert exc.value.fail_kwargs == dict()

219
tests/unit/plugins/modules/net_tools/test_hetzner_failover_ip.py
View file

@ -1,219 +0,0 @@
# (c) 2020 Felix Fontein <felix@fontein.de>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
from ansible_collections.community.internal_test_tools.tests.unit.utils.fetch_url_module_framework import (
FetchUrlCall,
BaseTestModule,
)
from ansible_collections.community.general.plugins.module_utils.hetzner import BASE_URL
from ansible_collections.community.general.plugins.modules.net_tools import hetzner_failover_ip
class TestHetznerFailoverIP(BaseTestModule):
MOCK_ANSIBLE_MODULEUTILS_BASIC_ANSIBLEMODULE = 'ansible_collections.community.general.plugins.modules.net_tools.hetzner_failover_ip.AnsibleModule'
MOCK_ANSIBLE_MODULEUTILS_URLS_FETCH_URL = 'ansible_collections.community.general.plugins.module_utils.hetzner.fetch_url'
# Tests for state idempotence (routed and unrouted)
def test_unrouted(self, mocker):
result = self.run_module_success(mocker, hetzner_failover_ip, {
'hetzner_user': '',
'hetzner_password': '',
'failover_ip': '1.2.3.4',
'state': 'unrouted',
}, [
FetchUrlCall('GET', 200)
.result_json({
'failover': {
'ip': '1.2.3.4',
'netmask': '255.255.255.255',
'server_ip': '2.3.4.5',
'server_number': 2345,
'active_server_ip': None,
},
})
.expect_url('{0}/failover/1.2.3.4'.format(BASE_URL)),
])
assert result['changed'] is False
assert result['value'] is None
assert result['state'] == 'unrouted'
def test_routed(self, mocker):
result = self.run_module_success(mocker, hetzner_failover_ip, {
'hetzner_user': '',
'hetzner_password': '',
'failover_ip': '1.2.3.4',
'state': 'routed',
'value': '4.3.2.1',
}, [
FetchUrlCall('GET', 200)
.result_json({
'failover': {
'ip': '1.2.3.4',
'netmask': '255.255.255.255',
'server_ip': '2.3.4.5',
'server_number': 2345,
'active_server_ip': '4.3.2.1',
},
})
.expect_url('{0}/failover/1.2.3.4'.format(BASE_URL)),
])
assert result['changed'] is False
assert result['value'] == '4.3.2.1'
assert result['state'] == 'routed'
# Tests for changing state (unrouted to routed, vice versa)
def test_unrouted_to_routed(self, mocker):
result = self.run_module_success(mocker, hetzner_failover_ip, {
'hetzner_user': '',
'hetzner_password': '',
'failover_ip': '1.2.3.4',
'state': 'routed',
'value': '4.3.2.1',
}, [
FetchUrlCall('GET', 200)
.result_json({
'failover': {
'ip': '1.2.3.4',
'netmask': '255.255.255.255',
'server_ip': '2.3.4.5',
'server_number': 2345,
'active_server_ip': None,
},
})
.expect_url('{0}/failover/1.2.3.4'.format(BASE_URL)),
FetchUrlCall('POST', 200)
.result_json({
'failover': {
'ip': '1.2.3.4',
'netmask': '255.255.255.255',
'server_ip': '2.3.4.5',
'server_number': 2345,
'active_server_ip': '4.3.2.1',
},
})
.expect_form_value('active_server_ip', '4.3.2.1')
.expect_url('{0}/failover/1.2.3.4'.format(BASE_URL)),
])
assert result['changed'] is True
assert result['value'] == '4.3.2.1'
assert result['state'] == 'routed'
def test_routed_to_unrouted(self, mocker):
result = self.run_module_success(mocker, hetzner_failover_ip, {
'hetzner_user': '',
'hetzner_password': '',
'failover_ip': '1.2.3.4',
'state': 'unrouted',
}, [
FetchUrlCall('GET', 200)
.result_json({
'failover': {
'ip': '1.2.3.4',
'netmask': '255.255.255.255',
'server_ip': '2.3.4.5',
'server_number': 2345,
'active_server_ip': '4.3.2.1',
},
})
.expect_url('{0}/failover/1.2.3.4'.format(BASE_URL)),
FetchUrlCall('DELETE', 200)
.result_json({
'failover': {
'ip': '1.2.3.4',
'netmask': '255.255.255.255',
'server_ip': '2.3.4.5',
'server_number': 2345,
'active_server_ip': None,
},
})
.expect_url('{0}/failover/1.2.3.4'.format(BASE_URL)),
])
assert result['changed'] is True
assert result['value'] is None
assert result['state'] == 'unrouted'
# Tests for re-routing
def test_rerouting(self, mocker):
result = self.run_module_success(mocker, hetzner_failover_ip, {
'hetzner_user': '',
'hetzner_password': '',
'failover_ip': '1.2.3.4',
'state': 'routed',
'value': '4.3.2.1',
}, [
FetchUrlCall('GET', 200)
.result_json({
'failover': {
'ip': '1.2.3.4',
'netmask': '255.255.255.255',
'server_ip': '2.3.4.5',
'server_number': 2345,
'active_server_ip': '5.4.3.2',
},
})
.expect_url('{0}/failover/1.2.3.4'.format(BASE_URL)),
FetchUrlCall('POST', 200)
.result_json({
'failover': {
'ip': '1.2.3.4',
'netmask': '255.255.255.255',
'server_ip': '2.3.4.5',
'server_number': 2345,
'active_server_ip': '4.3.2.1',
},
})
.expect_form_value('active_server_ip', '4.3.2.1')
.expect_url('{0}/failover/1.2.3.4'.format(BASE_URL)),
])
assert result['changed'] is True
assert result['value'] == '4.3.2.1'
assert result['state'] == 'routed'
def test_rerouting_already_routed(self, mocker):
result = self.run_module_success(mocker, hetzner_failover_ip, {
'hetzner_user': '',
'hetzner_password': '',
'failover_ip': '1.2.3.4',
'state': 'routed',
'value': '4.3.2.1',
}, [
FetchUrlCall('GET', 200)
.result_json({
'failover': {
'ip': '1.2.3.4',
'netmask': '255.255.255.255',
'server_ip': '2.3.4.5',
'server_number': 2345,
'active_server_ip': '5.4.3.2',
},
})
.expect_url('{0}/failover/1.2.3.4'.format(BASE_URL)),
FetchUrlCall('POST', 409)
.result_json({
'error': {
'status': 409,
'code': 'FAILOVER_ALREADY_ROUTED',
'message': 'Failover already routed',
},
'failover': {
'ip': '1.2.3.4',
'netmask': '255.255.255.255',
'server_ip': '2.3.4.5',
'server_number': 2345,
'active_server_ip': '4.3.2.1',
},
})
.expect_form_value('active_server_ip', '4.3.2.1')
.expect_url('{0}/failover/1.2.3.4'.format(BASE_URL)),
])
assert result['changed'] is False
assert result['value'] == '4.3.2.1'
assert result['state'] == 'routed'

71
tests/unit/plugins/modules/net_tools/test_hetzner_failover_ip_info.py
View file

@ -1,71 +0,0 @@
# (c) 2020 Felix Fontein <felix@fontein.de>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
from ansible_collections.community.internal_test_tools.tests.unit.utils.fetch_url_module_framework import (
FetchUrlCall,
BaseTestModule,
)
from ansible_collections.community.general.plugins.module_utils.hetzner import BASE_URL
from ansible_collections.community.general.plugins.modules.net_tools import hetzner_failover_ip_info
class TestHetznerFailoverIPInfo(BaseTestModule):
MOCK_ANSIBLE_MODULEUTILS_BASIC_ANSIBLEMODULE = 'ansible_collections.community.general.plugins.modules.net_tools.hetzner_failover_ip_info.AnsibleModule'
MOCK_ANSIBLE_MODULEUTILS_URLS_FETCH_URL = 'ansible_collections.community.general.plugins.module_utils.hetzner.fetch_url'
# Tests for state (routed and unrouted)
def test_unrouted(self, mocker):
result = self.run_module_success(mocker, hetzner_failover_ip_info, {
'hetzner_user': '',
'hetzner_password': '',
'failover_ip': '1.2.3.4',
}, [
FetchUrlCall('GET', 200)
.result_json({
'failover': {
'ip': '1.2.3.4',
'netmask': '255.255.255.255',
'server_ip': '2.3.4.5',
'server_number': 2345,
'active_server_ip': None,
},
})
.expect_url('{0}/failover/1.2.3.4'.format(BASE_URL)),
])
assert result['changed'] is False
assert result['value'] is None
assert result['state'] == 'unrouted'
assert result['failover_ip'] == '1.2.3.4'
assert result['server_ip'] == '2.3.4.5'
assert result['server_number'] == 2345
def test_routed(self, mocker):
result = self.run_module_success(mocker, hetzner_failover_ip_info, {
'hetzner_user': '',
'hetzner_password': '',
'failover_ip': '1.2.3.4',
}, [
FetchUrlCall('GET', 200)
.result_json({
'failover': {
'ip': '1.2.3.4',
'netmask': '255.255.255.255',
'server_ip': '2.3.4.5',
'server_number': 2345,
'active_server_ip': '4.3.2.1',
},
})
.expect_url('{0}/failover/1.2.3.4'.format(BASE_URL)),
])
assert result['changed'] is False
assert result['value'] == '4.3.2.1'
assert result['state'] == 'routed'
assert result['failover_ip'] == '1.2.3.4'
assert result['server_ip'] == '2.3.4.5'
assert result['server_number'] == 2345

1193
tests/unit/plugins/modules/net_tools/test_hetzner_firewall.py
View file

File diff suppressed because it is too large Load diff

240
tests/unit/plugins/modules/net_tools/test_hetzner_firewall_info.py
View file

@ -1,240 +0,0 @@
# (c) 2019 Felix Fontein <felix@fontein.de>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
from ansible_collections.community.internal_test_tools.tests.unit.utils.fetch_url_module_framework import (
FetchUrlCall,
BaseTestModule,
)
from ansible_collections.community.general.plugins.module_utils.hetzner import BASE_URL
from ansible_collections.community.general.plugins.modules.net_tools import hetzner_firewall_info
class TestHetznerFirewallInfo(BaseTestModule):
MOCK_ANSIBLE_MODULEUTILS_BASIC_ANSIBLEMODULE = 'ansible_collections.community.general.plugins.modules.net_tools.hetzner_firewall_info.AnsibleModule'
MOCK_ANSIBLE_MODULEUTILS_URLS_FETCH_URL = 'ansible_collections.community.general.plugins.module_utils.hetzner.fetch_url'
# Tests for state (absent and present)
def test_absent(self, mocker):
result = self.run_module_success(mocker, hetzner_firewall_info, {
'hetzner_user': '',
'hetzner_password': '',
'server_ip': '1.2.3.4',
}, [
FetchUrlCall('GET', 200)
.result_json({
'firewall': {
'server_ip': '1.2.3.4',
'server_number': 1,
'status': 'disabled',
'whitelist_hos': False,
'port': 'main',
'rules': {
'input': [],
},
},
})
.expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)),
])
assert result['changed'] is False
assert result['firewall']['status'] == 'disabled'
assert result['firewall']['server_ip'] == '1.2.3.4'
assert result['firewall']['server_number'] == 1
def test_present(self, mocker):
result = self.run_module_success(mocker, hetzner_firewall_info, {
'hetzner_user': '',
'hetzner_password': '',
'server_ip': '1.2.3.4',
}, [
FetchUrlCall('GET', 200)
.result_json({
'firewall': {
'server_ip': '1.2.3.4',
'server_number': 1,
'status': 'active',
'whitelist_hos': False,
'port': 'main',
'rules': {
'input': [],
},
},
})
.expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)),
])
assert result['changed'] is False
assert result['firewall']['status'] == 'active'
assert result['firewall']['server_ip'] == '1.2.3.4'
assert result['firewall']['server_number'] == 1
assert len(result['firewall']['rules']['input']) == 0
def test_present_w_rules(self, mocker):
result = self.run_module_success(mocker, hetzner_firewall_info, {
'hetzner_user': '',
'hetzner_password': '',
'server_ip': '1.2.3.4',
}, [
FetchUrlCall('GET', 200)
.result_json({
'firewall': {
'server_ip': '1.2.3.4',
'server_number': 1,
'status': 'active',
'whitelist_hos': False,
'port': 'main',
'rules': {
'input': [
{
'name': 'Accept HTTPS traffic',
'ip_version': 'ipv4',
'dst_ip': None,
'dst_port': '443',
'src_ip': None,
'src_port': None,
'protocol': 'tcp',
'tcp_flags': None,
'action': 'accept',
},
{
'name': None,
'ip_version': 'ipv4',
'dst_ip': None,
'dst_port': None,
'src_ip': None,
'src_port': None,
'protocol': None,
'tcp_flags': None,
'action': 'discard',
}
],
},
},
})
.expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)),
])
assert result['changed'] is False
assert result['firewall']['status'] == 'active'
assert result['firewall']['server_ip'] == '1.2.3.4'
assert result['firewall']['server_number'] == 1
assert len(result['firewall']['rules']['input']) == 2
assert result['firewall']['rules']['input'][0]['name'] == 'Accept HTTPS traffic'
assert result['firewall']['rules']['input'][0]['dst_port'] == '443'
assert result['firewall']['rules']['input'][0]['action'] == 'accept'
assert result['firewall']['rules']['input'][1]['dst_port'] is None
assert result['firewall']['rules']['input'][1]['action'] == 'discard'
# Tests for wait_for_configured in getting status
def test_wait_get(self, mocker):
mocker.patch('time.sleep', lambda duration: None)
result = self.run_module_success(mocker, hetzner_firewall_info, {
'hetzner_user': '',
'hetzner_password': '',
'server_ip': '1.2.3.4',
'wait_for_configured': True,
}, [
FetchUrlCall('GET', 200)
.result_json({
'firewall': {
'server_ip': '1.2.3.4',
'server_number': 1,
'status': 'in process',
'whitelist_hos': False,
'port': 'main',
'rules': {
'input': [],
},
},
})
.expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)),
FetchUrlCall('GET', 200)
.result_json({
'firewall': {
'server_ip': '1.2.3.4',
'server_number': 1,
'status': 'active',
'whitelist_hos': False,
'port': 'main',
'rules': {
'input': [],
},
},
})
.expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)),
])
assert result['changed'] is False
assert result['firewall']['status'] == 'active'
assert result['firewall']['server_ip'] == '1.2.3.4'
assert result['firewall']['server_number'] == 1
def test_wait_get_timeout(self, mocker):
mocker.patch('time.sleep', lambda duration: None)
result = self.run_module_failed(mocker, hetzner_firewall_info, {
'hetzner_user': '',
'hetzner_password': '',
'server_ip': '1.2.3.4',
'wait_for_configured': True,
'timeout': 0,
}, [
FetchUrlCall('GET', 200)
.result_json({
'firewall': {
'server_ip': '1.2.3.4',
'server_number': 1,
'status': 'in process',
'whitelist_hos': False,
'port': 'main',
'rules': {
'input': [],
},
},
})
.expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)),
FetchUrlCall('GET', 200)
.result_json({
'firewall': {
'server_ip': '1.2.3.4',
'server_number': 1,
'status': 'in process',
'whitelist_hos': False,
'port': 'main',
'rules': {
'input': [],
},
},
})
.expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)),
])
assert result['msg'] == 'Timeout while waiting for firewall to be configured.'
def test_nowait_get(self, mocker):
result = self.run_module_success(mocker, hetzner_firewall_info, {
'hetzner_user': '',
'hetzner_password': '',
'server_ip': '1.2.3.4',
'wait_for_configured': False,
}, [
FetchUrlCall('GET', 200)
.result_json({
'firewall': {
'server_ip': '1.2.3.4',
'server_number': 1,
'status': 'in process',
'whitelist_hos': False,
'port': 'main',
'rules': {
'input': [],
},
},
})
.expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)),
])
assert result['changed'] is False
assert result['firewall']['status'] == 'in process'
assert result['firewall']['server_ip'] == '1.2.3.4'
assert result['firewall']['server_number'] == 1