ansible-modules-syncthing/collection/plugins/module_utils/syncthing_api.py

import json
import os
import platform
from xml.etree.ElementTree import parse

from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.urls import fetch_url, url_argument_spec

if platform.system() == 'Windows':
    DEFAULT_ST_CONFIG_LOCATION = '%localappdata%/Syncthing/config.xml'
elif platform.system() == 'Darwin':
    DEFAULT_ST_CONFIG_LOCATION = '$HOME/Library/Application Support/Syncthing/config.xml'
else:
    DEFAULT_ST_CONFIG_LOCATION = '$HOME/.local/state/syncthing/config.xml'


class SyncthingModule(AnsibleModule):
    def _make_headers(self, target=None):
        url = '{}{}{}{}'.format(
                                self.params['host'],
                                self.api_url,
                                '/' if target else '',
                                target if target else '')
        headers = {'X-Api-Key': self.params['api_key'] }
        return url, headers

    def _get_key_from_filesystem(self):
        try:
            if self.params['config_file']:
                stconfigfile = self.params['config_file']
            else:
                stconfigfile = os.path.expandvars(DEFAULT_ST_CONFIG_LOCATION)
            stconfig = parse(stconfigfile)
            root = stconfig.getroot()
            gui = root.find('gui')
            self.params['api_key'] = gui.find('apikey').text
        except Exception:
            self.fail_json(msg="Auto-configuration failed. Please specify")

    def _api_call(self, method='GET', data=None, target=None, missing_ok=False):
        unix_socket = None
        if 'unix_socket' in self.params:
            unix_socket = self.params['unix_socket']
        url, headers = self._make_headers(target=target)
        if data:
            headers['Content-Type'] = 'application/json'
            data = json.dumps(data)
        self.result = {
            "changed": method != 'GET',
            "response": None,
        }
        resp, info = fetch_url(self,
                               url=url, unix_socket=unix_socket,
                               data=data, headers=headers,
                               method=method,
                               timeout=self.params['timeout'])
        if info:
            self.result['response'] = info
        else:
            self.fail_json(msg='Error occured while calling host')
        if info['status'] not in [200, 404]:
            self.fail_json(msg='Error occured while calling host')
        if info['status'] == 404:
            if missing_ok:
                return {}
            else:
                self.fail_json(msg='Error occured while calling host')
        try:
            content = resp.read()
        except AttributeError:
            self.result['content'] = info.pop('body', '')
            self.fail_json(msg='Error occured while reading response')
        try:
            return json.loads(content)
        except json.decoder.JSONDecodeError:
            return {'content': content}

    def get_call(self, target=None):
        return self._api_call(missing_ok=True, target=target)
    def delete_call(self, target):
        return self._api_call(method='DELETE', target=target)
    def post_call(self, data=None, target=None):
        return self._api_call(method='POST', data=data, target=target)
    def put_call(self, data=None, target=None):
        return self._api_call(method='PUT', data=data, target=target)
    def patch_call(self, data=None, target=None):
        return self._api_call(method='PATCH', data=data, target=target)

    def exit_json(self):
        super().exit_json(**self.result)
    def fail_json(self, msg=""):
        super().fail_json(msg, **self.result)

    def __init__(self, api_url='/', argument_spec=None, supports_check_mode=True):
        self.api_url = api_url
        argument_spec_temp = url_argument_spec()
        argument_spec_temp.update(dict(
            host=dict(type='str', default='http://127.0.0.1:8384'),
            unix_socket=dict(type='str', required=False), 
            api_key=dict(type='str', required=False, no_log=True),
            config_file=dict(type='str', required=False),
            timeout=dict(type='int', default=30),
        ))
        if argument_spec is None:
            argument_spec = {}
        argument_spec_temp.update(argument_spec)
        super().__init__(argument_spec=argument_spec_temp, supports_check_mode=True)

        # Auto-configuration: Try to fetch API key from filesystem
        if not self.params['api_key']:
            self._get_key_from_filesystem()