mirror of
https://github.com/zahodi/ansible-mikrotik.git
synced 2025-06-21 01:25:47 +02:00
237 lines
7.9 KiB
Python
237 lines
7.9 KiB
Python
#!/usr/bin/env python
|
|
from ansible.module_utils import mt_api
|
|
import re
|
|
import sys
|
|
import socket
|
|
|
|
|
|
def list_to_string(list):
|
|
list_string = ""
|
|
list_string = ','.join(map(str, list))
|
|
return list_string
|
|
|
|
|
|
def clean_params(params):
|
|
'''
|
|
remove keys with empty values
|
|
modify keys with '_' to match mikrotik parameters
|
|
convert yes/no to true/false
|
|
'''
|
|
if isinstance(params, dict):
|
|
for key in list(params):
|
|
if params[key] is None:
|
|
del params[key]
|
|
continue
|
|
|
|
new_key = re.sub('_', '-', key)
|
|
if new_key != key:
|
|
params[new_key] = str(params[key])
|
|
del params[key]
|
|
continue
|
|
|
|
if params[key] == "yes":
|
|
params[key] = "true"
|
|
if params[key] == "no":
|
|
params[key] = "false"
|
|
else:
|
|
print("Must be a dictionary")
|
|
|
|
|
|
class MikrotikIdempotent():
|
|
'''
|
|
MikrotikIdempotent Class
|
|
- A helper class for Ansible modules to abstract common functions.
|
|
|
|
Example Usage:
|
|
mt_obj = MikrotikIdempotent(
|
|
hostname = params['hostname'],
|
|
username = params['username'],
|
|
password = params['password'],
|
|
state = None,
|
|
desired_params = params['settings'],
|
|
idempotent_param= 'name',
|
|
api_path = '/interface/ethernet',
|
|
)
|
|
|
|
mt_obj.sync_state()
|
|
'''
|
|
|
|
def __init__(
|
|
self, hostname, username, password, desired_params, api_path,
|
|
state, idempotent_param, check_mode=False):
|
|
|
|
self.hostname = hostname
|
|
self.username = username
|
|
self.password = password
|
|
self.state = state
|
|
self.desired_params = desired_params
|
|
self.idempotent_param = idempotent_param
|
|
self.current_params = {}
|
|
self.api_path = api_path
|
|
self.check_mode = check_mode
|
|
|
|
self.login_success = False
|
|
self.changed = False
|
|
self.changed_msg = []
|
|
self.failed = False
|
|
self.failed_msg = []
|
|
|
|
self.login()
|
|
|
|
def login(self):
|
|
self.mk = mt_api.Mikrotik(
|
|
self.hostname,
|
|
self.username,
|
|
self.password,
|
|
)
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
result = sock.connect_ex((self.hostname, 8728))
|
|
if result == 0:
|
|
try:
|
|
self.mk.login()
|
|
self.login_success = True
|
|
except Exception as e:
|
|
self.failed_msg = "Could not log into Mikrotik device." + " Check the username and password. Exception {} - {}".format(type(e), e),
|
|
else:
|
|
self.failed_msg = "Could not access RouterOS api." + " Verify API service is enabled and not blocked by firewall.",
|
|
|
|
|
|
def get_current_params(self):
|
|
clean_params(self.desired_params)
|
|
self.param_id = None
|
|
self.current_param = None
|
|
self.current_params = self.mk.api_print(base_path=self.api_path)
|
|
|
|
# When state and idempotent_param is None we are working
|
|
# on editable params only and we are grabbing the only item from response
|
|
if self.state is None and self.idempotent_param is None:
|
|
self.current_param = self.current_params[0][1]
|
|
|
|
# Else we iterate over every item in the list until we find the matching
|
|
# params
|
|
# We also set the param_id here to reference later for editing or removal
|
|
else:
|
|
for current_param in self.current_params:
|
|
if self.idempotent_param in current_param[1]:
|
|
if self.desired_params[self.idempotent_param] == current_param[1][self.idempotent_param]:
|
|
self.current_param = current_param[1]
|
|
self.param_id = current_param[1]['.id']
|
|
# current_param now is a dict, something like:
|
|
# {
|
|
# ".id": "*1",
|
|
# "full-duplex": "true",
|
|
# "mac-address": "08:00:27:6F:4C:22",
|
|
# "mtu": "1500",
|
|
# "name": "ether1",
|
|
# ...
|
|
# }
|
|
|
|
def add(self):
|
|
# When current_param is empty we need to call api_add method to add
|
|
# all the parameters in the desired_params
|
|
if self.current_param is None:
|
|
# check if we have a list within the dictionary
|
|
# convert the list to string to pass to mikrotik
|
|
for i in self.desired_params:
|
|
if isinstance(self.desired_params[i], list):
|
|
self.desired_params[i] = list_to_string(self.desired_params[i])
|
|
self.new_params = self.desired_params
|
|
self.old_params = ""
|
|
if not self.check_mode:
|
|
self.mk.api_add(
|
|
base_path = self.api_path,
|
|
params = self.desired_params,
|
|
)
|
|
self.changed = True
|
|
|
|
# Else we need to determing what the difference between the currently
|
|
# and the desired
|
|
else:
|
|
self.edit()
|
|
|
|
def rem(self):
|
|
# if param_id is set this means there is currently a matching item
|
|
# which we will remove
|
|
if self.param_id:
|
|
self.new_params = "item removed"
|
|
self.old_params = self.desired_params
|
|
if not self.check_mode:
|
|
self.mk.api_remove(
|
|
base_path=self.api_path,
|
|
remove_id=self.param_id,
|
|
)
|
|
self.changed = True
|
|
|
|
def edit(self):
|
|
# out_params is used to pass to api_edit() to make changes
|
|
# to a mikrotik device
|
|
out_params = {}
|
|
# old_params used storing old values that are going to be changed
|
|
# to aid in the diff output
|
|
old_params = {} # used to store values of params we change
|
|
|
|
# iterate over items in desired params and
|
|
# match against items in current_param
|
|
# to figure out the difference
|
|
for desired_param in self.desired_params:
|
|
# check if a desired item is already set in mikrotik
|
|
if desired_param in self.current_param:
|
|
# check if we have a list within the dictionary
|
|
# convert mikrotik string to list to get a diff
|
|
if isinstance(self.desired_params[desired_param], list):
|
|
if desired_param in self.current_param:
|
|
current_param_list = self.current_param[desired_param].split(',')
|
|
if set(self.desired_params[desired_param]) != set(current_param_list):
|
|
out_params[desired_param] = list_to_string(self.desired_params[desired_param])
|
|
old_params[desired_param] = str(self.current_param[desired_param])
|
|
else:
|
|
out_params[desired_param] = list_to_string(self.desired_params[desired_param])
|
|
# value is not a list, move on and identify difference
|
|
else:
|
|
if self.current_param[desired_param] != str(self.desired_params[desired_param]):
|
|
out_params[desired_param] = str(self.desired_params[desired_param])
|
|
old_params[desired_param] = str(self.current_param[desired_param])
|
|
# since we didn't get a matching key from mikrotik settings
|
|
# we'll it the out_params to whatever is desired_param
|
|
else:
|
|
if isinstance(desired_param, list):
|
|
out_params[desired_param] = list_to_string(self.desired_params[desired_param])
|
|
else:
|
|
out_params[desired_param] = str(self.desired_params[desired_param])
|
|
|
|
# When out_params has been set it means we found our diff
|
|
# and will set it on the mikrotik
|
|
if out_params:
|
|
if self.param_id is not None:
|
|
out_params['.id'] = self.current_param['.id']
|
|
|
|
if not self.check_mode:
|
|
self.mk.api_edit(
|
|
base_path = self.api_path,
|
|
params = out_params,
|
|
)
|
|
|
|
# we don't need to show the .id in the changed message
|
|
if '.id' in out_params:
|
|
del out_params['.id']
|
|
|
|
self.changed_msg.append({
|
|
"new_params": out_params,
|
|
"old_params": old_params,
|
|
})
|
|
|
|
self.new_params = out_params
|
|
self.old_params = old_params
|
|
self.changed = True
|
|
|
|
def sync_state(self):
|
|
self.get_current_params()
|
|
|
|
# When state and idempotent_param are not set we are working
|
|
# on editable parameters only that we can't add or remove
|
|
if self.state is None and self.idempotent_param is None:
|
|
self.edit()
|
|
elif self.state == "absent":
|
|
self.rem()
|
|
elif self.state == "present" or self.idempotent_param:
|
|
self.add()
|