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
community.general/plugins/httpapi/fortimanager.py
Ansible Core Team aebc1b03fd Initial commit
2020-03-09 09:11:07 +00:00

451 lines
17 KiB
Python
Raw Blame History

# Copyright (c) 2018 Fortinet and/or its affiliates.
#
# 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 <http://www.gnu.org/licenses/>.
#
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
DOCUMENTATION = '''
---
author:
- Luke Weighall (@lweighall)
- Andrew Welsh (@Ghilli3)
- Jim Huber (@p4r4n0y1ng)
httpapi : fortimanager
short_description: HttpApi Plugin for Fortinet FortiManager Appliance or VM.
description:
- This HttpApi plugin provides methods to connect to Fortinet FortiManager Appliance or VM via JSON RPC API.
'''
import json
from ansible.plugins.httpapi import HttpApiBase
from ansible.module_utils.basic import to_text
from ansible_collections.fortinet.fortios.plugins.module_utils.network.fortimanager.common import BASE_HEADERS
from ansible_collections.fortinet.fortios.plugins.module_utils.network.fortimanager.common import FMGBaseException
from ansible_collections.fortinet.fortios.plugins.module_utils.network.fortimanager.common import FMGRCommon
from ansible_collections.fortinet.fortios.plugins.module_utils.network.fortimanager.common import FMGRMethods
class HttpApi(HttpApiBase):
def __init__(self, connection):
super(HttpApi, self).__init__(connection)
self._req_id = 0
self._sid = None
self._url = "/jsonrpc"
self._host = None
self._tools = FMGRCommon
self._debug = False
self._connected_fmgr = None
self._last_response_msg = None
self._last_response_code = None
self._last_data_payload = None
self._last_url = None
self._last_response_raw = None
self._locked_adom_list = list()
self._locked_adoms_by_user = list()
self._uses_workspace = False
self._uses_adoms = False
self._adom_list = list()
self._logged_in_user = None
def set_become(self, become_context):
"""
ELEVATION IS NOT REQUIRED ON FORTINET DEVICES - SKIPPED.
:param become_context: Unused input.
:return: None
"""
return None
def update_auth(self, response, response_data):
"""
TOKENS ARE NOT USED SO NO NEED TO UPDATE AUTH.
:param response: Unused input.
:param response_data Unused_input.
:return: None
"""
return None
def login(self, username, password):
"""
This function will log the plugin into FortiManager, and return the results.
:param username: Username of FortiManager Admin
:param password: Password of FortiManager Admin
:return: Dictionary of status if it logged in or not.
"""
self._logged_in_user = username
self.send_request(FMGRMethods.EXEC, self._tools.format_request(FMGRMethods.EXEC, "sys/login/user",
passwd=password, user=username, ))
if "FortiManager object connected to FortiManager" in self.__str__():
# If Login worked, then inspect the FortiManager for Workspace Mode, and it's system information.
self.inspect_fmgr()
return
else:
raise FMGBaseException(msg="Unknown error while logging in...connection was lost during login operation...."
" Exiting")
def inspect_fmgr(self):
# CHECK FOR WORKSPACE MODE TO SEE IF WE HAVE TO ENABLE ADOM LOCKS
status = self.get_system_status()
if status[0] == -11:
# THE CONNECTION GOT LOST SOMEHOW, REMOVE THE SID AND REPORT BAD LOGIN
self.logout()
raise FMGBaseException(msg="Error -11 -- the Session ID was likely malformed somehow. Contact authors."
" Exiting")
elif status[0] == 0:
try:
self.check_mode()
if self._uses_adoms:
self.get_adom_list()
if self._uses_workspace:
self.get_locked_adom_list()
self._connected_fmgr = status[1]
self._host = self._connected_fmgr["Hostname"]
except BaseException:
pass
return
def logout(self):
"""
This function will logout of the FortiManager.
"""
if self.sid is not None:
# IF WE WERE USING WORKSPACES, THEN CLEAN UP OUR LOCKS IF THEY STILL EXIST
if self.uses_workspace:
self.get_lock_info()
self.run_unlock()
ret_code, response = self.send_request(FMGRMethods.EXEC,
self._tools.format_request(FMGRMethods.EXEC, "sys/logout"))
self.sid = None
return ret_code, response
def send_request(self, method, params):
"""
Responsible for actual sending of data to the connection httpapi base plugin. Does some formatting too.
:param params: A formatted dictionary that was returned by self.common_datagram_params()
before being called here.
:param method: The preferred API Request method (GET, ADD, POST, etc....)
:type method: basestring
:return: Dictionary of status, if it logged in or not.
"""
try:
if self.sid is None and params[0]["url"] != "sys/login/user":
try:
self.connection._connect()
except Exception as err:
raise FMGBaseException(
msg="An problem happened with the httpapi plugin self-init connection process. "
"Error: " + to_text(err))
except IndexError:
raise FMGBaseException("An attempt was made at communicating with a FMG with "
"no valid session and an incorrectly formatted request.")
except Exception as err:
raise FMGBaseException("An attempt was made at communicating with a FMG with "
"no valid session and an unexpected error was discovered. \n Error: " + to_text(err))
self._update_request_id()
json_request = {
"method": method,
"params": params,
"session": self.sid,
"id": self.req_id,
"verbose": 1
}
data = json.dumps(json_request, ensure_ascii=False).replace('\\\\', '\\')
try:
# Sending URL and Data in Unicode, per Ansible Specifications for Connection Plugins
response, response_data = self.connection.send(path=to_text(self._url), data=to_text(data),
headers=BASE_HEADERS)
# Get Unicode Response - Must convert from StringIO to unicode first so we can do a replace function below
result = json.loads(to_text(response_data.getvalue()))
self._update_self_from_response(result, self._url, data)
return self._handle_response(result)
except Exception as err:
raise FMGBaseException(err)
def _handle_response(self, response):
self._set_sid(response)
if isinstance(response["result"], list):
result = response["result"][0]
else:
result = response["result"]
if "data" in result:
return result["status"]["code"], result["data"]
else:
return result["status"]["code"], result
def _update_self_from_response(self, response, url, data):
self._last_response_raw = response
if isinstance(response["result"], list):
result = response["result"][0]
else:
result = response["result"]
if "status" in result:
self._last_response_code = result["status"]["code"]
self._last_response_msg = result["status"]["message"]
self._last_url = url
self._last_data_payload = data
def _set_sid(self, response):
if self.sid is None and "session" in response:
self.sid = response["session"]
def return_connected_fmgr(self):
"""
Returns the data stored under self._connected_fmgr
:return: dict
"""
try:
if self._connected_fmgr:
return self._connected_fmgr
except Exception:
raise FMGBaseException("Couldn't Retrieve Connected FMGR Stats")
def get_system_status(self):
"""
Returns the system status page from the FortiManager, for logging and other uses.
return: status
"""
status = self.send_request(FMGRMethods.GET, self._tools.format_request(FMGRMethods.GET, "sys/status"))
return status
@property
def debug(self):
return self._debug
@debug.setter
def debug(self, val):
self._debug = val
@property
def req_id(self):
return self._req_id
@req_id.setter
def req_id(self, val):
self._req_id = val
def _update_request_id(self, reqid=0):
self.req_id = reqid if reqid != 0 else self.req_id + 1
@property
def sid(self):
return self._sid
@sid.setter
def sid(self, val):
self._sid = val
def __str__(self):
if self.sid is not None and self.connection._url is not None:
return "FortiManager object connected to FortiManager: " + to_text(self.connection._url)
return "FortiManager object with no valid connection to a FortiManager appliance."
##################################
# BEGIN DATABASE LOCK CONTEXT CODE
##################################
@property
def uses_workspace(self):
return self._uses_workspace
@uses_workspace.setter
def uses_workspace(self, val):
self._uses_workspace = val
@property
def uses_adoms(self):
return self._uses_adoms
@uses_adoms.setter
def uses_adoms(self, val):
self._uses_adoms = val
def add_adom_to_lock_list(self, adom):
if adom not in self._locked_adom_list:
self._locked_adom_list.append(adom)
def remove_adom_from_lock_list(self, adom):
if adom in self._locked_adom_list:
self._locked_adom_list.remove(adom)
def check_mode(self):
"""
Checks FortiManager for the use of Workspace mode
"""
url = "/cli/global/system/global"
code, resp_obj = self.send_request(FMGRMethods.GET,
self._tools.format_request(FMGRMethods.GET,
url,
fields=["workspace-mode", "adom-status"]))
try:
if resp_obj["workspace-mode"] == "workflow":
self.uses_workspace = True
elif resp_obj["workspace-mode"] == "disabled":
self.uses_workspace = False
except KeyError:
raise FMGBaseException(msg="Couldn't determine workspace-mode in the plugin")
try:
if resp_obj["adom-status"] in [1, "enable"]:
self.uses_adoms = True
else:
self.uses_adoms = False
except KeyError:
raise FMGBaseException(msg="Couldn't determine adom-status in the plugin")
def run_unlock(self):
"""
Checks for ADOM status, if locked, it will unlock
"""
for adom_locked in self._locked_adoms_by_user:
adom = adom_locked["adom"]
self.unlock_adom(adom)
def lock_adom(self, adom=None, *args, **kwargs):
"""
Locks an ADOM for changes
"""
if adom:
if adom.lower() == "global":
url = "/dvmdb/global/workspace/lock/"
else:
url = "/dvmdb/adom/{adom}/workspace/lock/".format(adom=adom)
else:
url = "/dvmdb/adom/root/workspace/lock"
code, respobj = self.send_request(FMGRMethods.EXEC, self._tools.format_request(FMGRMethods.EXEC, url))
if code == 0 and respobj["status"]["message"].lower() == "ok":
self.add_adom_to_lock_list(adom)
return code, respobj
def unlock_adom(self, adom=None, *args, **kwargs):
"""
Unlocks an ADOM after changes
"""
if adom:
if adom.lower() == "global":
url = "/dvmdb/global/workspace/unlock/"
else:
url = "/dvmdb/adom/{adom}/workspace/unlock/".format(adom=adom)
else:
url = "/dvmdb/adom/root/workspace/unlock"
code, respobj = self.send_request(FMGRMethods.EXEC, self._tools.format_request(FMGRMethods.EXEC, url))
if code == 0 and respobj["status"]["message"].lower() == "ok":
self.remove_adom_from_lock_list(adom)
return code, respobj
def commit_changes(self, adom=None, aux=False, *args, **kwargs):
"""
Commits changes to an ADOM
"""
if adom:
if aux:
url = "/pm/config/adom/{adom}/workspace/commit".format(adom=adom)
else:
if adom.lower() == "global":
url = "/dvmdb/global/workspace/commit/"
else:
url = "/dvmdb/adom/{adom}/workspace/commit".format(adom=adom)
else:
url = "/dvmdb/adom/root/workspace/commit"
return self.send_request(FMGRMethods.EXEC, self._tools.format_request(FMGRMethods.EXEC, url))
def get_lock_info(self, adom=None):
"""
Gets ADOM lock info so it can be displayed with the error messages. Or if determined to be locked by ansible
for some reason, then unlock it.
"""
if not adom or adom == "root":
url = "/dvmdb/adom/root/workspace/lockinfo"
else:
if adom.lower() == "global":
url = "/dvmdb/global/workspace/lockinfo/"
else:
url = "/dvmdb/adom/{adom}/workspace/lockinfo/".format(adom=adom)
datagram = {}
data = self._tools.format_request(FMGRMethods.GET, url, **datagram)
resp_obj = self.send_request(FMGRMethods.GET, data)
code = resp_obj[0]
if code != 0:
self._module.fail_json(msg=("An error occurred trying to get the ADOM Lock Info. "
"Error: " + to_text(resp_obj)))
elif code == 0:
try:
if resp_obj[1]["status"]["message"] == "OK":
self._lock_info = None
except Exception:
self._lock_info = resp_obj[1]
return resp_obj
def get_adom_list(self):
"""
Gets the list of ADOMs for the FortiManager
"""
if self.uses_adoms:
url = "/dvmdb/adom"
datagram = {}
data = self._tools.format_request(FMGRMethods.GET, url, **datagram)
resp_obj = self.send_request(FMGRMethods.GET, data)
code = resp_obj[0]
if code != 0:
self._module.fail_json(msg=("An error occurred trying to get the ADOM Info. "
"Error: " + to_text(resp_obj)))
elif code == 0:
num_of_adoms = len(resp_obj[1])
append_list = ['root', ]
for adom in resp_obj[1]:
if adom["tab_status"] != "":
append_list.append(to_text(adom["name"]))
self._adom_list = append_list
return resp_obj
def get_locked_adom_list(self):
"""
Gets the list of locked adoms
"""
try:
locked_list = list()
locked_by_user_list = list()
for adom in self._adom_list:
adom_lock_info = self.get_lock_info(adom=adom)
try:
if adom_lock_info[1]["status"]["message"] == "OK":
continue
except IndexError as err:
pass
try:
if adom_lock_info[1][0]["lock_user"]:
locked_list.append(to_text(adom))
if adom_lock_info[1][0]["lock_user"] == self._logged_in_user:
locked_by_user_list.append({"adom": to_text(adom),
"user": to_text(adom_lock_info[1][0]["lock_user"])})
except Exception as err:
raise FMGBaseException(err)
self._locked_adom_list = locked_list
self._locked_adoms_by_user = locked_by_user_list
except Exception as err:
raise FMGBaseException(msg=("An error occurred while trying to get the locked adom list. Error: "
+ to_text(err)))
################################
# END DATABASE LOCK CONTEXT CODE
################################
View git blame Copy permalink