From 3e91ec28b8f858cf9b5e2819f5904d8a945c06dc Mon Sep 17 00:00:00 2001 From: Ricardo Carrillo Cruz Date: Wed, 9 Jan 2019 11:25:18 +0100 Subject: [PATCH] Checkpoint access rule (#49937) * WIP checkpoint_access_rule module * Add fixes and docstrings * Add dunder init * Fix sanity tests issues * Fix sanity test * Add RETURN and EXAMPLES * Fix example * Fix pep8 * Add tests * Fix pep8 * Fix pep8 --- .../modules/network/checkpoint/__init__.py | 0 .../checkpoint/checkpoint_access_rule.py | 235 ++++++++++++++++++ lib/ansible/plugins/httpapi/checkpoint.py | 2 + .../modules/network/checkpoint/__init__.py | 0 .../checkpoint/test_checkpoint_access_rule.py | 105 ++++++++ 5 files changed, 342 insertions(+) create mode 100644 lib/ansible/modules/network/checkpoint/__init__.py create mode 100644 lib/ansible/modules/network/checkpoint/checkpoint_access_rule.py create mode 100644 test/units/modules/network/checkpoint/__init__.py create mode 100644 test/units/modules/network/checkpoint/test_checkpoint_access_rule.py diff --git a/lib/ansible/modules/network/checkpoint/__init__.py b/lib/ansible/modules/network/checkpoint/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/lib/ansible/modules/network/checkpoint/checkpoint_access_rule.py b/lib/ansible/modules/network/checkpoint/checkpoint_access_rule.py new file mode 100644 index 0000000000..0b82e10897 --- /dev/null +++ b/lib/ansible/modules/network/checkpoint/checkpoint_access_rule.py @@ -0,0 +1,235 @@ +#!/usr/bin/python +# +# This file is part of Ansible +# +# Ansible is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# Ansible is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with Ansible. If not, see . +# + +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + + +ANSIBLE_METADATA = {'metadata_version': '1.1', + 'status': ['preview'], + 'supported_by': 'network'} + + +DOCUMENTATION = """ +--- +module: checkpoint_access_rule +short_description: Manages access rules on Checkpoint over Web Services API +description: + - Manages access rules on Checkpoint devices including creating, updating, removing access rules objects, + All operations are performed over Web Services API. +version_added: "2.8" +author: "Ansible by Red Hat (@rcarrillocruz)" +options: + name: + description: + - Name of the access rule. + type: str + layer: + description: + - Layer to attach the access rule to. + required: True + type: str + position: + description: + - Position of the access rule. + type: str + source: + description: + - Source object of the access rule. + type: str + destination: + description: + - Destionation object of the access rule. + type: str + action: + description: + - Action of the access rule (accept, drop, inform, etc). + type: str + default: drop + enabled: + description: + - Enabled or disabled flag. + type: bool + default: True + state: + description: + - State of the access rule (present or absent). Defaults to present. + type: str + default: present +""" + +EXAMPLES = """ +- name: Create access rule + checkpoint_access_rule: + layer: Network + name: "Drop attacker" + position: top + source: attacker + destination: Any + action: Drop + +- name: Delete access rule + checkpoint_access_rule: + layer: Network + name: "Drop attacker" +""" + +RETURN = """ +checkpoint_access_rules: + description: The checkpoint access rule object created or updated. + returned: always, except when deleting the access rule. + type: list +""" + + +from ansible.module_utils.basic import AnsibleModule +from ansible.module_utils.connection import Connection +from ansible.module_utils.network.checkpoint.checkpoint import publish, install_policy +import json + + +def get_access_rule(module, connection): + name = module.params['name'] + layer = module.params['layer'] + + payload = {'name': name, 'layer': layer} + + code, response = connection.send_request('/web_api/show-access-rule', payload) + + return code, response + + +def create_access_rule(module, connection): + name = module.params['name'] + layer = module.params['layer'] + position = module.params['position'] + source = module.params['source'] + destination = module.params['destination'] + action = module.params['action'] + + payload = {'name': name, + 'layer': layer, + 'position': position, + 'source': source, + 'destination': destination, + 'action': action} + + code, response = connection.send_request('/web_api/add-access-rule', payload) + + return code, response + + +def update_access_rule(module, connection): + name = module.params['name'] + layer = module.params['layer'] + position = module.params['position'] + source = module.params['source'] + destination = module.params['destination'] + action = module.params['action'] + enabled = module.params['enabled'] + + payload = {'name': name, + 'layer': layer, + 'position': position, + 'source': source, + 'destination': destination, + 'action': action, + 'enabled': enabled} + + code, response = connection.send_request('/web_api/set-access-rule', payload) + + return code, response + + +def delete_access_rule(module, connection): + name = module.params['name'] + layer = module.params['layer'] + + payload = {'name': name, + 'layer': layer, + } + + code, response = connection.send_request('/web_api/delete-access-rule', payload) + + return code, response + + +def needs_update(module, access_rule): + res = False + + if module.params['source'] and module.params['source'] != access_rule['source'][0]['name']: + res = True + if module.params['destination'] and module.params['destination'] != access_rule['destination'][0]['name']: + res = True + if module.params['action'] != access_rule['action']['name']: + res = True + if module.params['enabled'] != access_rule['enabled']: + res = True + + return res + + +def main(): + argument_spec = dict( + name=dict(type='str', required=True), + layer=dict(type='str'), + position=dict(type='str'), + source=dict(type='str'), + destination=dict(type='str'), + action=dict(type='str', default='drop'), + enabled=dict(type='bool', default=True), + state=dict(type='str', default='present') + ) + + required_if = [('state', 'present', ('layer', 'position'))] + module = AnsibleModule(argument_spec=argument_spec, required_if=required_if) + connection = Connection(module._socket_path) + code, response = get_access_rule(module, connection) + result = {'changed': False} + + if module.params['state'] == 'present': + if code == 200: + if needs_update(module, response): + code, response = update_access_rule(module, connection) + publish(module, connection) + install_policy(module, connection) + result['changed'] = True + result['checkpoint_access_rules'] = response + else: + pass + elif code == 404: + code, response = create_access_rule(module, connection) + publish(module, connection) + install_policy(module, connection) + result['changed'] = True + result['checkpoint_access_rules'] = response + else: + if code == 200: + # Handle deletion + code, response = delete_access_rule(module, connection) + publish(module, connection) + install_policy(module, connection) + result['changed'] = True + elif code == 404: + pass + + module.exit_json(**result) + + +if __name__ == '__main__': + main() diff --git a/lib/ansible/plugins/httpapi/checkpoint.py b/lib/ansible/plugins/httpapi/checkpoint.py index 78d084134b..1a77a44db8 100644 --- a/lib/ansible/plugins/httpapi/checkpoint.py +++ b/lib/ansible/plugins/httpapi/checkpoint.py @@ -58,6 +58,8 @@ class HttpApi(HttpApiBase): value = self._get_response_value(response_data) return response.getcode(), self._response_to_json(value) + except AnsibleConnectionFailure as e: + return 404, 'Object not found' except HTTPError as e: error = json.loads(e.read()) return e.code, error diff --git a/test/units/modules/network/checkpoint/__init__.py b/test/units/modules/network/checkpoint/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/test/units/modules/network/checkpoint/test_checkpoint_access_rule.py b/test/units/modules/network/checkpoint/test_checkpoint_access_rule.py new file mode 100644 index 0000000000..4e17f71b16 --- /dev/null +++ b/test/units/modules/network/checkpoint/test_checkpoint_access_rule.py @@ -0,0 +1,105 @@ +# Copyright (c) 2018 Red Hat +# +# This file is part of Ansible +# +# Ansible is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# Ansible is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with Ansible. If not, see . +# + +from __future__ import absolute_import + +import pytest +from units.modules.utils import set_module_args, exit_json, fail_json, AnsibleFailJson, AnsibleExitJson + +from ansible.module_utils import basic +from ansible.modules.network.checkpoint import checkpoint_access_rule + +OBJECT = {'layer': 'foo', 'position': 'bar', 'name': 'baz', + 'source': [{'name': 'lol'}], 'destination': [{'name': 'Any'}], + 'action': {'name': 'drop'}, 'enabled': True} +PAYLOAD = {'layer': 'foo', 'position': 'bar', 'name': 'baz'} + + +class TestCheckpointAccessRule(object): + module = checkpoint_access_rule + + @pytest.fixture(autouse=True) + def module_mock(self, mocker): + return mocker.patch.multiple(basic.AnsibleModule, exit_json=exit_json, fail_json=fail_json) + + @pytest.fixture + def connection_mock(self, mocker): + connection_class_mock = mocker.patch('ansible.modules.network.checkpoint.checkpoint_access_rule.Connection') + return connection_class_mock.return_value + + @pytest.fixture + def get_access_rule_200(self, mocker): + mock_function = mocker.patch('ansible.modules.network.checkpoint.checkpoint_access_rule.get_access_rule') + mock_function.return_value = (200, OBJECT) + return mock_function.return_value + + @pytest.fixture + def get_access_rule_404(self, mocker): + mock_function = mocker.patch('ansible.modules.network.checkpoint.checkpoint_access_rule.get_access_rule') + mock_function.return_value = (404, 'Object not found') + return mock_function.return_value + + def test_create(self, get_access_rule_404, connection_mock): + connection_mock.send_request.return_value = (200, OBJECT) + result = self._run_module(PAYLOAD) + + assert result['changed'] + assert 'checkpoint_access_rules' in result + + def test_create_idempotent(self, get_access_rule_200, connection_mock): + connection_mock.send_request.return_value = (200, PAYLOAD) + result = self._run_module(PAYLOAD) + + assert not result['changed'] + + def test_update(self, get_access_rule_200, connection_mock): + payload_for_update = {'enabled': False} + payload_for_update.update(PAYLOAD) + connection_mock.send_request.return_value = (200, payload_for_update) + result = self._run_module(payload_for_update) + + assert result['changed'] + assert not result['checkpoint_access_rules']['enabled'] + + def test_delete(self, get_access_rule_200, connection_mock): + connection_mock.send_request.return_value = (200, OBJECT) + payload_for_delete = {'state': 'absent'} + payload_for_delete.update(PAYLOAD) + result = self._run_module(payload_for_delete) + + assert result['changed'] + + def test_delete_idempotent(self, get_access_rule_404, connection_mock): + payload = {'name': 'baz', 'state': 'absent'} + connection_mock.send_request.return_value = (200, OBJECT) + result = self._run_module(payload) + + assert not result['changed'] + + def _run_module(self, module_args): + set_module_args(module_args) + with pytest.raises(AnsibleExitJson) as ex: + self.module.main() + return ex.value.args[0] + + def _run_module_with_fail_json(self, module_args): + set_module_args(module_args) + with pytest.raises(AnsibleFailJson) as exc: + self.module.main() + result = exc.value.args[0] + return result