ansible-modules-syncthing/collection/plugins/module_utils/syncthing_api.py
Peter Surda 2c26c3a53e
All checks were successful
buildbot/multibuild_parent Build done.
buildbot/travis_bionic Build done.
Fixes
2024-04-23 12:12:40 +08:00

112 lines
4.4 KiB
Python

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()